flush

Co-authored-by: Tim <12827757+laibe@users.noreply.github.com>
Co-authored-by: joshieDo <93316087+joshieDo@users.noreply.github.com>
Co-authored-by: Amp <amp@ampcode.com>

.changelog/dry-lakes-sleep.md

@@ -0,0 +1,6 @@
+---
+reth-trie-common: minor
+reth-trie: minor
+---
+
+Added `contains_range` method to `PrefixSet` for checking if any key falls within a half-open range. Added prefix set support to `ProofCalculator` via `with_prefix_set`, enabling stale cached hash invalidation and branch collapse detection when keys are inserted or removed; propagated storage prefix sets through `SyncAccountValueEncoder`.

.changelog/dull-seals-laugh.md

@@ -0,0 +1,5 @@
+---
+reth-trie-sparse: minor
+---
+
+Fixed a bug in `ArenaParallelSparseTrie` where subtrie updates that would completely empty a subtrie were incorrectly dispatched to parallel workers instead of being processed inline, preventing correct branch collapse detection when blinded siblings are present. Refactored the `SparseTrie` test suite to accept a `fn() -> T` factory instead of requiring `T: Default`, enabling a new `arena_parallel_sparse_trie_always_parallel` test variant that exercises all tests with parallelism thresholds set to 1. Added `test_branch_collapse_multi_empty_subtries_blinded_remaining` to cover the case where removing multiple revealed leaves empties their subtries and leaves a single blinded sibling requiring a proof.

.changelog/nice-whales-buzz.md

@@ -0,0 +1,5 @@
+---
+reth-engine-tree: patch
+---
+
+Added idle-time pre-computation of account trie upper hashes in the sparse trie payload processor when no pending proof results are available.

.github/scripts/bench-job-summary.js

@@ -0,0 +1,106 @@
+// Generates a rich GitHub Actions job summary for reth-bench results.
+//
+// Reads from environment:
+//   BENCH_WORK_DIR       – Directory containing summary.json
+//   BENCH_PR             – PR number (may be empty)
+//   BENCH_ACTOR          – GitHub user who triggered the bench
+//   BENCH_CORES          – CPU core limit (0 = all)
+//   BENCH_WARMUP_BLOCKS  – Number of warmup blocks
+//   BENCH_SAMPLY         – 'true' if samply profiling was enabled
+//   BENCH_ABBA           – 'true' if ABBA interleaved order was used
+//
+// Usage from actions/github-script:
+//   const jobSummary = require('./.github/scripts/bench-job-summary.js');
+//   await jobSummary({ core, context, chartSha, grafanaUrl, runId });
+
+const fs = require('fs');
+const { verdict, loadSamplyUrls, blocksLabel, metricRows, waitTimeRows } = require('./bench-utils');
+
+module.exports = async function ({ core, context, chartSha, grafanaUrl, runId }) {
+  let summary;
+  try {
+    summary = JSON.parse(fs.readFileSync(process.env.BENCH_WORK_DIR + '/summary.json', 'utf8'));
+  } catch (e) {
+    await core.summary.addRaw('⚠️ Benchmark completed but failed to load summary.').write();
+    return;
+  }
+
+  const repo = `${context.repo.owner}/${context.repo.repo}`;
+  const prNumber = process.env.BENCH_PR;
+  const actor = process.env.BENCH_ACTOR;
+  const commitUrl = `https://github.com/${repo}/commit`;
+
+  const { emoji, label } = verdict(summary.changes);
+  const baselineLink = `[\`${summary.baseline.name}\`](${commitUrl}/${summary.baseline.ref})`;
+  const featureLink = `[\`${summary.feature.name}\`](${commitUrl}/${summary.feature.ref})`;
+  const diffUrl = `https://github.com/${repo}/compare/${summary.baseline.ref}...${summary.feature.ref}`;
+
+  // Header & metadata
+  const metaParts = [];
+  if (prNumber) metaParts.push(`**[PR #${prNumber}](https://github.com/${repo}/pull/${prNumber})**`);
+  metaParts.push(`triggered by @${actor}`);
+
+  let md = `# ${emoji} ${label}\n\n`;
+  md += metaParts.join(' · ') + '\n\n';
+  md += `**Baseline:** ${baselineLink}\n`;
+  md += `**Feature:** ${featureLink} ([diff](${diffUrl}))\n`;
+  md += blocksLabel(summary).map(p => `**${p.key}:** ${p.value}`).join(' · ') + '\n\n';
+
+  // Main comparison table
+  const rows = metricRows(summary);
+  md += `| Metric | Baseline | Feature | Change |\n`;
+  md += `|--------|----------|---------|--------|\n`;
+  for (const r of rows) {
+    md += `| ${r.label} | ${r.baseline} | ${r.feature} | ${r.change} |\n`;
+  }
+  md += '\n';
+
+  // Wait time breakdown
+  const wtRows = waitTimeRows(summary);
+  if (wtRows.length > 0) {
+    md += `### Wait Time Breakdown\n\n`;
+    md += `| Metric | Baseline | Feature |\n`;
+    md += `|--------|----------|--------|\n`;
+    for (const r of wtRows) {
+      md += `| ${r.title} | ${r.baseline} | ${r.feature} |\n`;
+    }
+    md += '\n';
+  }
+
+  // Charts
+  if (chartSha) {
+    const prNum = prNumber || '0';
+    const baseUrl = `https://raw.githubusercontent.com/decofe/reth-bench-charts/${chartSha}/pr/${prNum}/${runId}`;
+    const charts = [
+      { file: 'latency_throughput.png', label: 'Latency, Throughput & Diff' },
+      { file: 'wait_breakdown.png', label: 'Wait Time Breakdown' },
+      { file: 'gas_vs_latency.png', label: 'Gas vs Latency' },
+    ];
+    md += `### Charts\n\n`;
+    for (const chart of charts) {
+      md += `<details><summary>${chart.label}</summary>\n\n`;
+      md += `![${chart.label}](${baseUrl}/${chart.file})\n\n`;
+      md += `</details>\n\n`;
+    }
+  }
+
+  // Samply profiles
+  const samplyUrls = loadSamplyUrls(process.env.BENCH_WORK_DIR);
+  const samplyLinks = Object.entries(samplyUrls).map(([run, url]) => `- **${run}**: [Firefox Profiler](${url})`);
+  if (samplyLinks.length > 0) {
+    md += `### Samply Profiles\n\n${samplyLinks.join('\n')}\n\n`;
+  }
+
+  // Grafana
+  if (grafanaUrl) {
+    md += `### Grafana Dashboard\n\n[View real-time metrics](${grafanaUrl})\n\n`;
+  }
+
+  // Node errors
+  try {
+    const errors = fs.readFileSync(process.env.BENCH_WORK_DIR + '/errors.md', 'utf8');
+    if (errors.trim()) md += '\n' + errors + '\n';
+  } catch {}
+
+  await core.summary.addRaw(md).write();
+};

.github/scripts/bench-slack-notify.js

@@ -18,6 +18,7 @@
 
 const fs = require('fs');
 const path = require('path');
+const { fmtChange, fmtMs, verdict, loadSamplyUrls, blocksLabel, metricRows, waitTimeRows } = require('./bench-utils');
 
 const SLACK_API = 'https://slack.com/api/chat.postMessage';
 
@@ -61,41 +62,17 @@ function cell(text) {
   return { type: 'raw_text', text: s || ' ' };
 }
 
+// Slack shortcodes for verdict (Block Kit header doesn't support unicode emoji)
+const SLACK_VERDICT = {
+  '⚠️': ':warning:',
+  '❌': ':x:',
+  '✅': ':white_check_mark:',
+  '⚪': ':white_circle:',
+};
+
 function buildSuccessBlocks({ summary, prNumber, actor, actorSlackId, jobUrl, repo, samplyUrls }) {
-  const b = summary.baseline.stats;
-  const f = summary.feature.stats;
-  const c = summary.changes;
-
-  const sigEmoji = { good: '\u2705', bad: '\u274c', neutral: '\u26aa' };
-
-  function fmtMs(v) { return v.toFixed(2) + 'ms'; }
-  function fmtMgas(v) { return v.toFixed(2); }
-  function fmtS(v) { return v.toFixed(2) + 's'; }
-  function fmtChange(ch) {
-    if (!ch.pct && !ch.ci_pct) return ' ';
-    const pctStr = `${ch.pct >= 0 ? '+' : ''}${ch.pct.toFixed(2)}%`;
-    const ciStr = ch.ci_pct ? ` (\u00b1${ch.ci_pct.toFixed(2)}%)` : '';
-    return `${pctStr}${ciStr} ${sigEmoji[ch.sig]}`;
-  }
-
-  // Overall result for header
-  const vals = Object.values(c);
-  const hasBad = vals.some(v => v.sig === 'bad');
-  const hasGood = vals.some(v => v.sig === 'good');
-  let headerEmoji, headerResult;
-  if (hasBad && hasGood) {
-    headerEmoji = ':warning:';
-    headerResult = 'Mixed Results';
-  } else if (hasBad) {
-    headerEmoji = ':x:';
-    headerResult = 'Regression';
-  } else if (hasGood) {
-    headerEmoji = ':white_check_mark:';
-    headerResult = 'Improvement';
-  } else {
-    headerEmoji = ':white_circle:';
-    headerResult = 'No Difference';
-  }
+  const { emoji, label } = verdict(summary.changes);
+  const headerEmoji = SLACK_VERDICT[emoji] || emoji;
 
   const prUrl = prNumber ? `https://github.com/${repo}/pull/${prNumber}` : '';
   const commitUrl = `https://github.com/${repo}/commit`;
@@ -120,19 +97,7 @@ function buildSuccessBlocks({ summary, prNumber, actor, actorSlackId, jobUrl, re
   if (fl1) featureLine += ` | <${fl1}|Samply 1>`;
   if (fl2) featureLine += ` | <${fl2}|Samply 2>`;
 
-  const cores = process.env.BENCH_CORES || '0';
-  const countsParts = [];
-  if (summary.big_blocks) {
-    const gasRamp = summary.gas_ramp_blocks || 0;
-    if (gasRamp > 0) countsParts.push(`*Gas Ramp:* ${gasRamp}`);
-    countsParts.push(`*Big Blocks:* ${summary.blocks}`);
-  } else {
-    const warmup = summary.warmup_blocks || process.env.BENCH_WARMUP_BLOCKS || '';
-    if (warmup) countsParts.push(`*Warmup:* ${warmup}`);
-    countsParts.push(`*Blocks:* ${summary.blocks}`);
-  }
-  if (cores !== '0') countsParts.push(`*Cores:* ${cores}`);
-  const countsLine = countsParts.join(' | ');
+  const countsLine = blocksLabel(summary).map(p => `*${p.key}:* ${p.value}`).join(' | ');
 
   const baselineArgs = process.env.BENCH_BASELINE_ARGS || '';
   const featureArgs = process.env.BENCH_FEATURE_ARGS || '';
@@ -159,10 +124,17 @@ function buildSuccessBlocks({ summary, prNumber, actor, actorSlackId, jobUrl, re
     },
   ];
 
+  // Build table rows from shared metricRows
+  const rows = metricRows(summary);
+  const tableRows = [
+    [cell('Metric'), cell('Baseline'), cell('Feature'), cell('Change')],
+    ...rows.map(r => [cell(r.label), cell(r.baseline), cell(r.feature), cell(r.change || ' ')]),
+  ];
+
   const blocks = [
     {
       type: 'header',
-      text: { type: 'plain_text', text: `${headerEmoji} ${headerResult}`, emoji: true },
+      text: { type: 'plain_text', text: `${headerEmoji} ${label}`, emoji: true },
     },
     {
       type: 'section',
@@ -176,16 +148,7 @@ function buildSuccessBlocks({ summary, prNumber, actor, actorSlackId, jobUrl, re
         { align: 'right' },
         { align: 'right' },
       ],
-      rows: [
-        [cell('Metric'),  cell('Baseline'), cell('Feature'), cell('Change')],
-        [cell('Mean'),     cell(fmtMs(b.mean_ms)),      cell(fmtMs(f.mean_ms)),      cell(fmtChange(c.mean))],
-        [cell('StdDev'),   cell(fmtMs(b.stddev_ms)),    cell(fmtMs(f.stddev_ms)),    cell(' ')],
-        [cell('P50'),      cell(fmtMs(b.p50_ms)),       cell(fmtMs(f.p50_ms)),       cell(fmtChange(c.p50))],
-        [cell('P90'),      cell(fmtMs(b.p90_ms)),       cell(fmtMs(f.p90_ms)),       cell(fmtChange(c.p90))],
-        [cell('P99'),      cell(fmtMs(b.p99_ms)),       cell(fmtMs(f.p99_ms)),       cell(fmtChange(c.p99))],
-        [cell('Mgas/s'),   cell(fmtMgas(b.mean_mgas_s)), cell(fmtMgas(f.mean_mgas_s)), cell(fmtChange(c.mgas_s))],
-        [cell('Wall Clock'), cell(fmtS(b.wall_clock_s)), cell(fmtS(f.wall_clock_s)), cell(fmtChange(c.wall_clock))],
-      ],
+      rows: tableRows,
     },
     {
       type: 'actions',
@@ -195,16 +158,12 @@ function buildSuccessBlocks({ summary, prNumber, actor, actorSlackId, jobUrl, re
 
   // Wait times as a separate table block (sent as threaded reply due to Slack one-table limit)
   const threadBlocks = [];
-  const waitTimes = summary.wait_times || {};
-  const waitKeys = Object.keys(waitTimes);
-  if (waitKeys.length > 0) {
-    const waitRows = [
+  const wtRows = waitTimeRows(summary);
+  if (wtRows.length > 0) {
+    const waitTableRows = [
       [cell('Wait Time'), cell('Baseline'), cell('Feature')],
+      ...wtRows.map(r => [cell(r.title), cell(r.baseline), cell(r.feature)]),
     ];
-    for (const key of waitKeys) {
-      const wt = waitTimes[key];
-      waitRows.push([cell(wt.title), cell(fmtMs(wt.baseline.mean_ms)), cell(fmtMs(wt.feature.mean_ms))]);
-    }
     threadBlocks.push({
       type: 'table',
       column_settings: [
@@ -212,7 +171,7 @@ function buildSuccessBlocks({ summary, prNumber, actor, actorSlackId, jobUrl, re
         { align: 'right' },
         { align: 'right' },
       ],
-      rows: waitRows,
+      rows: waitTableRows,
     });
   }
 
@@ -274,16 +233,7 @@ async function success({ core, context }) {
   const jobUrl = process.env.BENCH_JOB_URL ||
     `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
 
-  // Load samply profile URLs (files exist when samply profiling was enabled)
-  const samplyUrls = {};
-  for (const run of ['baseline-1', 'baseline-2', 'feature-1', 'feature-2']) {
-    try {
-      const url = fs.readFileSync(
-        path.join(process.env.BENCH_WORK_DIR, run, 'samply-profile-url.txt'), 'utf8'
-      ).trim();
-      if (url) samplyUrls[run] = url;
-    } catch {}
-  }
+  const samplyUrls = loadSamplyUrls(process.env.BENCH_WORK_DIR);
 
   const slackUsers = loadSlackUsers(process.env.GITHUB_WORKSPACE || '.');
   const actorSlackId = slackUsers[actor];

.github/scripts/bench-utils.js

@@ -0,0 +1,96 @@
+// Shared utilities for reth-bench result rendering.
+//
+// Used by bench-job-summary.js and bench-slack-notify.js.
+
+const fs = require('fs');
+const path = require('path');
+
+const SIG_EMOJI = { good: '✅', bad: '❌', neutral: '⚪' };
+
+function fmtMs(v) { return v.toFixed(2) + 'ms'; }
+function fmtMgas(v) { return v.toFixed(2); }
+function fmtS(v) { return v.toFixed(2) + 's'; }
+
+function fmtChange(ch) {
+  if (!ch || (!ch.pct && !ch.ci_pct)) return '';
+  const pctStr = `${ch.pct >= 0 ? '+' : ''}${ch.pct.toFixed(2)}%`;
+  const ciStr = ch.ci_pct ? ` (±${ch.ci_pct.toFixed(2)}%)` : '';
+  return `${pctStr}${ciStr} ${SIG_EMOJI[ch.sig]}`;
+}
+
+function verdict(changes) {
+  const vals = Object.values(changes);
+  const hasBad = vals.some(v => v.sig === 'bad');
+  const hasGood = vals.some(v => v.sig === 'good');
+  if (hasBad && hasGood) return { emoji: '⚠️', label: 'Mixed Results' };
+  if (hasBad) return { emoji: '❌', label: 'Regression' };
+  if (hasGood) return { emoji: '✅', label: 'Improvement' };
+  return { emoji: '⚪', label: 'No Difference' };
+}
+
+function loadSamplyUrls(workDir) {
+  const urls = {};
+  for (const run of ['baseline-1', 'baseline-2', 'feature-1', 'feature-2']) {
+    try {
+      const url = fs.readFileSync(path.join(workDir, run, 'samply-profile-url.txt'), 'utf8').trim();
+      if (url) urls[run] = url;
+    } catch {}
+  }
+  return urls;
+}
+
+function blocksLabel(summary) {
+  const parts = [];
+  if (summary.big_blocks) {
+    if (summary.gas_ramp_blocks) parts.push({ key: 'Gas Ramp', value: summary.gas_ramp_blocks });
+    parts.push({ key: 'Big Blocks', value: summary.blocks });
+  } else {
+    const warmup = summary.warmup_blocks || process.env.BENCH_WARMUP_BLOCKS || '';
+    if (warmup) parts.push({ key: 'Warmup', value: warmup });
+    parts.push({ key: 'Blocks', value: summary.blocks });
+  }
+  const cores = process.env.BENCH_CORES || '0';
+  if (cores !== '0') parts.push({ key: 'Cores', value: cores });
+  return parts;
+}
+
+// The 7 metric rows shared by all renderers.
+// Returns an array of { label, baseline, feature, change } objects.
+function metricRows(summary) {
+  const b = summary.baseline.stats;
+  const f = summary.feature.stats;
+  const c = summary.changes;
+  return [
+    { label: 'Mean',       baseline: fmtMs(b.mean_ms),       feature: fmtMs(f.mean_ms),       change: fmtChange(c.mean) },
+    { label: 'StdDev',     baseline: fmtMs(b.stddev_ms),     feature: fmtMs(f.stddev_ms),     change: '' },
+    { label: 'P50',        baseline: fmtMs(b.p50_ms),        feature: fmtMs(f.p50_ms),        change: fmtChange(c.p50) },
+    { label: 'P90',        baseline: fmtMs(b.p90_ms),        feature: fmtMs(f.p90_ms),        change: fmtChange(c.p90) },
+    { label: 'P99',        baseline: fmtMs(b.p99_ms),        feature: fmtMs(f.p99_ms),        change: fmtChange(c.p99) },
+    { label: 'Mgas/s',     baseline: fmtMgas(b.mean_mgas_s), feature: fmtMgas(f.mean_mgas_s), change: fmtChange(c.mgas_s) },
+    { label: 'Wall Clock', baseline: fmtS(b.wall_clock_s),   feature: fmtS(f.wall_clock_s),   change: fmtChange(c.wall_clock) },
+  ];
+}
+
+// Wait time rows: one row per metric showing mean values.
+function waitTimeRows(summary) {
+  const waitTimes = summary.wait_times || {};
+  const rows = [];
+  for (const key of Object.keys(waitTimes)) {
+    const wt = waitTimes[key];
+    rows.push({ title: wt.title, baseline: fmtMs(wt.baseline.mean_ms), feature: fmtMs(wt.feature.mean_ms) });
+  }
+  return rows;
+}
+
+module.exports = {
+  SIG_EMOJI,
+  fmtMs,
+  fmtMgas,
+  fmtS,
+  fmtChange,
+  verdict,
+  loadSamplyUrls,
+  blocksLabel,
+  metricRows,
+  waitTimeRows,
+};

.github/workflows/bench.yml

@@ -1195,11 +1195,22 @@ jobs:
                 comment_id: parseInt(ackCommentId),
                 body,
               });
-            } else {
-              // No PR — write results to job summary
-              await core.summary.addRaw(body).write();
             }
 
+      - name: Write job summary
+        if: success()
+        uses: actions/github-script@v8
+        with:
+          script: |
+            const jobSummary = require('./.github/scripts/bench-job-summary.js');
+            await jobSummary({
+              core,
+              context,
+              chartSha: '${{ steps.push-charts.outputs.sha }}',
+              grafanaUrl: '${{ steps.metrics.outputs.grafana-url }}',
+              runId: '${{ github.run_id }}',
+            });
+
       - name: Send Slack notification (success)
         if: success() && env.BENCH_NO_SLACK != 'true'
         uses: actions/github-script@v8

Cargo.lock

@@ -9639,6 +9639,7 @@ version = "1.11.3"
 dependencies = [
  "alloy-consensus",
  "alloy-eips",
+ "alloy-genesis",
  "alloy-primitives",
  "alloy-rpc-types-engine",
  "assert_matches",

Cargo.toml

@@ -444,7 +444,6 @@ revm-state = { version = "10.0.0", default-features = false }
 revm-primitives = { version = "22.1.0", default-features = false }
 revm-interpreter = { version = "34.0.0", default-features = false }
 revm-database-interface = { version = "10.0.0", default-features = false }
-op-revm = { version = "17.0.0", default-features = false }
 revm-inspectors = "0.36.0"
 
 # eth

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

@@ -791,15 +791,10 @@ where
         // If the canonical chain is ahead of the new chain,
         // gather all blocks until new head number.
         while current_canonical_number > current_number {
-            if let Some(block) = self.canonical_block_by_hash(old_hash)? {
-                old_hash = block.recovered_block().parent_hash();
-                old_chain.push(block);
-                current_canonical_number -= 1;
-            } else {
-                // This shouldn't happen as we're walking back the canonical chain
-                warn!(target: "engine::tree", current_hash=?old_hash, "Canonical block not found in TreeState");
-                return Ok(None)
-            }
+            let block = self.canonical_block_by_hash(old_hash)?;
+            old_hash = block.recovered_block().parent_hash();
+            old_chain.push(block);
+            current_canonical_number -= 1;
         }
 
         // Both new and old chain pointers are now at the same height.
@@ -808,14 +803,9 @@ where
         // Walk both chains from specified hashes at same height until
         // a common ancestor (fork block) is reached.
         while old_hash != current_hash {
-            if let Some(block) = self.canonical_block_by_hash(old_hash)? {
-                old_hash = block.recovered_block().parent_hash();
-                old_chain.push(block);
-            } else {
-                // This shouldn't happen as we're walking back the canonical chain
-                warn!(target: "engine::tree", current_hash=?old_hash, "Canonical block not found in TreeState");
-                return Ok(None)
-            }
+            let block = self.canonical_block_by_hash(old_hash)?;
+            old_hash = block.recovered_block().parent_hash();
+            old_chain.push(block);
 
             if let Some(block) = self.state.tree_state.executed_block_by_hash(current_hash).cloned()
             {
@@ -942,36 +932,22 @@ where
         let new_head_hash = canonical_header.hash();
         let new_head_number = canonical_header.number();
 
-        // Try to load the canonical ancestor's block
-        match self.canonical_block_by_hash(new_head_hash)? {
-            Some(executed_block) => {
-                // Perform the reorg to properly handle the unwind
-                self.canonical_in_memory_state.update_chain(NewCanonicalChain::Reorg {
-                    new: vec![executed_block],
-                    old: old_blocks,
-                });
+        // Load the canonical ancestor's block
+        let executed_block = self.canonical_block_by_hash(new_head_hash)?;
+        // Perform the reorg to properly handle the unwind
+        self.canonical_in_memory_state
+            .update_chain(NewCanonicalChain::Reorg { new: vec![executed_block], old: old_blocks });
 
-                // CRITICAL: Update the canonical head after the reorg
-                // This ensures get_canonical_head() returns the correct block
-                self.canonical_in_memory_state.set_canonical_head(canonical_header.clone());
+        // CRITICAL: Update the canonical head after the reorg
+        // This ensures get_canonical_head() returns the correct block
+        self.canonical_in_memory_state.set_canonical_head(canonical_header.clone());
 
-                debug!(
-                    target: "engine::tree",
-                    block_number = new_head_number,
-                    block_hash = ?new_head_hash,
-                    "Successfully loaded canonical ancestor into memory via reorg"
-                );
-            }
-            None => {
-                // Fallback: update header only if block cannot be found
-                warn!(
-                    target: "engine::tree",
-                    block_hash = ?new_head_hash,
-                    "Could not find canonical ancestor block, updating header only"
-                );
-                self.canonical_in_memory_state.set_canonical_head(canonical_header.clone());
-            }
-        }
+        debug!(
+            target: "engine::tree",
+            block_number = new_head_number,
+            block_hash = ?new_head_hash,
+            "Successfully loaded canonical ancestor into memory via reorg"
+        );
 
         Ok(())
     }
@@ -997,18 +973,17 @@ where
             return Ok(());
         }
 
-        // Try to load the block from storage
-        if let Some(executed_block) = self.canonical_block_by_hash(block_hash)? {
-            self.canonical_in_memory_state
-                .update_chain(NewCanonicalChain::Commit { new: vec![executed_block] });
+        // Load the block from storage
+        let executed_block = self.canonical_block_by_hash(block_hash)?;
+        self.canonical_in_memory_state
+            .update_chain(NewCanonicalChain::Commit { new: vec![executed_block] });
 
-            debug!(
-                target: "engine::tree",
-                block_number,
-                block_hash = ?block_hash,
-                "Added canonical block to in-memory state"
-            );
-        }
+        debug!(
+            target: "engine::tree",
+            block_number,
+            block_hash = ?block_hash,
+            "Added canonical block to in-memory state"
+        );
 
         Ok(())
     }
@@ -2029,11 +2004,11 @@ where
     /// pruned for a given block, this operation will return an error. On archive nodes, it
     /// can retrieve any block.
     #[instrument(level = "debug", target = "engine::tree", skip(self))]
-    fn canonical_block_by_hash(&self, hash: B256) -> ProviderResult<Option<ExecutedBlock<N>>> {
+    fn canonical_block_by_hash(&self, hash: B256) -> ProviderResult<ExecutedBlock<N>> {
         trace!(target: "engine::tree", ?hash, "Fetching executed block by hash");
         // check memory first
         if let Some(block) = self.state.tree_state.executed_block_by_hash(hash) {
-            return Ok(Some(block.clone()))
+            return Ok(block.clone())
         }
 
         let (block, senders) = self
@@ -2075,11 +2050,22 @@ where
             },
         });
 
-        Ok(Some(ExecutedBlock::new(
+        Ok(ExecutedBlock::new(
             Arc::new(RecoveredBlock::new_sealed(block, senders)),
             execution_output,
             trie_data,
-        )))
+        ))
+    }
+
+    /// Returns `true` if a block with the given hash is known, either in memory or in the
+    /// database. This is a lightweight existence check that avoids constructing a full
+    /// [`SealedHeader`].
+    fn has_block_by_hash(&self, hash: B256) -> ProviderResult<bool> {
+        if self.state.tree_state.contains_hash(&hash) {
+            Ok(true)
+        } else {
+            self.provider.is_known(hash)
+        }
     }
 
     /// Return sealed block header from in-memory state or database by hash.
@@ -2126,7 +2112,7 @@ where
         parent_hash: B256,
     ) -> ProviderResult<Option<B256>> {
         // Check if parent exists in side chain or in canonical chain.
-        if self.sealed_header_by_hash(parent_hash)?.is_some() {
+        if self.has_block_by_hash(parent_hash)? {
             return Ok(Some(parent_hash))
         }
 
@@ -2140,7 +2126,7 @@ where
 
             // If current_header is None, then the current_hash does not have an invalid
             // ancestor in the cache, check its presence in blockchain tree
-            if current_block.is_none() && self.sealed_header_by_hash(current_hash)?.is_some() {
+            if current_block.is_none() && self.has_block_by_hash(current_hash)? {
                 return Ok(Some(current_hash))
             }
         }
@@ -2807,7 +2793,7 @@ where
         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() {
+        if self.state.tree_state.contains_hash(&block_num_hash.hash) {
             convert_to_block(self, input)?;
             return Ok(InsertPayloadOk::AlreadySeen(BlockStatus::Valid));
         }

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

@@ -38,7 +38,10 @@ use reth_trie_parallel::{
     proof_task::{ProofTaskCtx, ProofWorkerHandle},
     root::ParallelStateRootError,
 };
-use reth_trie_sparse::ParallelismThresholds;
+use reth_trie_sparse::{
+    ArenaParallelSparseTrie, ConfigurableSparseTrie, ParallelSparseTrie, ParallelismThresholds,
+    RevealableSparseTrie, SparseStateTrie,
+};
 use std::{
     ops::Not,
     sync::{
@@ -57,10 +60,7 @@ pub mod prewarm;
 pub mod receipt_root_task;
 pub mod sparse_trie;
 
-pub use preserved_sparse_trie::{
-    PayloadSparseTrieCache, PayloadSparseTrieKind, PayloadSparseTrieStoreOutcome,
-    SparseTrieCheckout,
-};
+use preserved_sparse_trie::{PreservedSparseTrie, SharedPreservedSparseTrie};
 
 /// Default parallelism thresholds to use with the [`ParallelSparseTrie`].
 ///
@@ -104,52 +104,6 @@ type IteratorPayloadHandle<Evm, I, N> = PayloadHandle<
     <N as NodePrimitives>::Receipt,
 >;
 
-/// Shared cache handles that can be exported to engine consumers and downstream payload builders.
-#[derive(Debug, Clone)]
-pub struct EngineSharedCaches<Evm: ConfigureEvm> {
-    execution_cache: PayloadExecutionCache,
-    sparse_trie_cache: PayloadSparseTrieCache,
-    precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
-}
-
-impl<Evm> Default for EngineSharedCaches<Evm>
-where
-    Evm: ConfigureEvm,
-{
-    fn default() -> Self {
-        Self::with_sparse_trie_kind(PayloadSparseTrieKind::default())
-    }
-}
-
-impl<Evm> EngineSharedCaches<Evm>
-where
-    Evm: ConfigureEvm,
-{
-    /// Creates shared caches backed by the requested sparse trie implementation.
-    pub fn with_sparse_trie_kind(sparse_trie_kind: PayloadSparseTrieKind) -> Self {
-        Self {
-            execution_cache: Default::default(),
-            sparse_trie_cache: PayloadSparseTrieCache::new(sparse_trie_kind),
-            precompile_cache_map: Default::default(),
-        }
-    }
-
-    /// Returns the shared execution cache handle for engine-internal use.
-    pub(crate) fn execution_cache(&self) -> PayloadExecutionCache {
-        self.execution_cache.clone()
-    }
-
-    /// Returns the shared sparse trie cache handle.
-    pub fn sparse_trie_cache(&self) -> PayloadSparseTrieCache {
-        self.sparse_trie_cache.clone()
-    }
-
-    /// Returns the shared precompile cache map.
-    pub fn precompile_cache_map(&self) -> PrecompileCacheMap<SpecFor<Evm>> {
-        self.precompile_cache_map.clone()
-    }
-}
-
 /// Entrypoint for executing the payload.
 #[derive(Debug)]
 pub struct PayloadProcessor<Evm>
@@ -158,8 +112,8 @@ where
 {
     /// The executor used by to spawn tasks.
     executor: Runtime,
-    /// Shared caches reused across payload processing.
-    shared_caches: EngineSharedCaches<Evm>,
+    /// The most recent cache used for execution.
+    execution_cache: PayloadExecutionCache,
     /// Metrics for trie operations
     trie_metrics: MultiProofTaskMetrics,
     /// Cross-block cache size in bytes.
@@ -172,12 +126,20 @@ where
     evm_config: Evm,
     /// Whether precompile cache should be disabled.
     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,
     /// LFU hot-slot capacity: max storage slots retained across prune cycles.
     sparse_trie_max_hot_slots: usize,
     /// LFU hot-account capacity: max account addresses retained across prune cycles.
     sparse_trie_max_hot_accounts: usize,
     /// Whether sparse trie cache pruning is fully disabled.
     disable_sparse_trie_cache_pruning: bool,
+    /// Whether to use the arena-based sparse trie implementation.
+    enable_arena_sparse_trie: bool,
     /// Whether to disable cache metrics recording.
     disable_cache_metrics: bool,
 }
@@ -197,20 +159,23 @@ where
         executor: Runtime,
         evm_config: Evm,
         config: &TreeConfig,
-        shared_caches: EngineSharedCaches<Evm>,
+        precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
     ) -> Self {
         Self {
             executor,
-            shared_caches,
+            execution_cache: Default::default(),
             trie_metrics: Default::default(),
             cross_block_cache_size: config.cross_block_cache_size(),
             disable_transaction_prewarming: config.disable_prewarming(),
             evm_config,
             disable_state_cache: config.disable_state_cache(),
             precompile_cache_disabled: config.precompile_cache_disabled(),
+            precompile_cache_map,
+            sparse_state_trie: SharedPreservedSparseTrie::default(),
             sparse_trie_max_hot_slots: config.sparse_trie_max_hot_slots(),
             sparse_trie_max_hot_accounts: config.sparse_trie_max_hot_accounts(),
             disable_sparse_trie_cache_pruning: config.disable_sparse_trie_cache_pruning(),
+            enable_arena_sparse_trie: config.enable_arena_sparse_trie(),
             disable_cache_metrics: config.disable_cache_metrics(),
         }
     }
@@ -224,8 +189,8 @@ where
         debug!(target: "engine::tree::payload_processor", "Waiting for execution cache and sparse trie locks");
 
         // Wait for both caches in parallel using std threads
-        let execution_cache = self.shared_caches.execution_cache();
-        let sparse_trie = self.shared_caches.sparse_trie_cache();
+        let execution_cache = self.execution_cache.clone();
+        let sparse_trie = self.sparse_state_trie.clone();
 
         // Use channels and spawn_blocking instead of std::thread::spawn
         let (execution_tx, execution_rx) = std::sync::mpsc::channel();
@@ -539,12 +504,12 @@ where
             terminate_execution: Arc::new(AtomicBool::new(false)),
             executed_tx_index: Arc::clone(&executed_tx_index),
             precompile_cache_disabled: self.precompile_cache_disabled,
-            precompile_cache_map: self.shared_caches.precompile_cache_map(),
+            precompile_cache_map: self.precompile_cache_map.clone(),
         };
 
         let (prewarm_task, to_prewarm_task) = PrewarmCacheTask::new(
             self.executor.clone(),
-            self.shared_caches.execution_cache(),
+            self.execution_cache.clone(),
             prewarm_ctx,
             to_multi_proof,
         );
@@ -572,7 +537,7 @@ where
     /// instance.
     #[instrument(level = "debug", target = "engine::caching", skip(self))]
     fn cache_for(&self, parent_hash: B256) -> SavedCache {
-        if let Some(cache) = self.shared_caches.execution_cache().get_cache_for(parent_hash) {
+        if let Some(cache) = self.execution_cache.get_cache_for(parent_hash) {
             debug!("reusing execution cache");
             cache
         } else {
@@ -597,11 +562,12 @@ where
         parent_state_root: B256,
         chunk_size: usize,
     ) {
-        let sparse_trie_cache = self.shared_caches.sparse_trie_cache();
+        let preserved_sparse_trie = self.sparse_state_trie.clone();
         let trie_metrics = self.trie_metrics.clone();
         let max_hot_slots = self.sparse_trie_max_hot_slots;
         let max_hot_accounts = self.sparse_trie_max_hot_accounts;
         let disable_cache_pruning = self.disable_sparse_trie_cache_pruning;
+        let enable_arena_sparse_trie = self.enable_arena_sparse_trie;
         let executor = self.executor.clone();
 
         let parent_span = Span::current();
@@ -611,19 +577,49 @@ where
             let _enter = debug_span!(target: "engine::tree::payload_processor", parent: parent_span, "sparse_trie_task")
                 .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 start = Instant::now();
-            let mut checkout = sparse_trie_cache.take_or_create_for(parent_state_root);
+            let preserved = preserved_sparse_trie.take();
             trie_metrics
                 .sparse_trie_cache_wait_duration_histogram
                 .record(start.elapsed().as_secs_f64());
-            checkout.set_hot_cache_capacities(max_hot_slots, max_hot_accounts);
 
-            let mut task = SparseTrieCacheTask::new_with_checkout(
+            let mut sparse_state_trie = preserved
+                .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 = if enable_arena_sparse_trie {
+                        RevealableSparseTrie::blind_from(
+                            ConfigurableSparseTrie::Arena(ArenaParallelSparseTrie::default()),
+                        )
+                    } else {
+                        RevealableSparseTrie::blind_from(
+                            ConfigurableSparseTrie::HashMap(
+                                ParallelSparseTrie::default().with_parallelism_thresholds(
+                                    PARALLEL_SPARSE_TRIE_PARALLELISM_THRESHOLDS,
+                                ),
+                            ),
+                        )
+                    };
+                    SparseStateTrie::default()
+                        .with_accounts_trie(default_trie.clone())
+                        .with_default_storage_trie(default_trie)
+                        .with_updates(true)
+                });
+            sparse_state_trie.set_hot_cache_capacities(max_hot_slots, max_hot_accounts);
+
+            let mut task = SparseTrieCacheTask::new_with_trie(
                 &executor,
                 from_multi_proof,
                 proof_worker_handle,
                 trie_metrics.clone(),
-                checkout,
+                sparse_state_trie,
                 chunk_size,
             );
 
@@ -634,7 +630,7 @@ where
             // 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 = sparse_trie_cache.lock();
+            let mut guard = preserved_sparse_trie.lock();
 
             let task_result = result.as_ref().ok().cloned();
             // Send state root computation result - next block may start but will block on take()
@@ -649,7 +645,7 @@ where
                     SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
                     SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
                 );
-                trie.store_prepared_cleared_with_guard(&mut guard);
+                guard.store(PreservedSparseTrie::cleared(trie));
                 drop(guard);
                 executor.spawn_drop(deferred);
                 return;
@@ -678,7 +674,7 @@ where
                 trie_metrics
                     .sparse_trie_retained_storage_tries
                     .set(trie.retained_storage_tries_count() as f64);
-                trie.store_anchored_with_guard(&mut guard, result.state_root);
+                guard.store(PreservedSparseTrie::anchored(trie, result.state_root));
                 deferred
             } else {
                 debug!(
@@ -689,7 +685,7 @@ where
                     SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
                     SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
                 );
-                trie.store_prepared_cleared_with_guard(&mut guard);
+                guard.store(PreservedSparseTrie::cleared(trie));
                 deferred
             };
             drop(guard);
@@ -710,7 +706,7 @@ where
         bundle_state: &BundleState,
     ) {
         let disable_cache_metrics = self.disable_cache_metrics;
-        self.shared_caches.execution_cache().update_with_guard(|cached| {
+        self.execution_cache.update_with_guard(|cached| {
             if cached.as_ref().is_some_and(|c| c.executed_block_hash() != block_with_parent.parent) {
                 debug!(
                     target: "engine::caching",
@@ -1014,7 +1010,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(crate) struct PayloadExecutionCache {
+pub struct PayloadExecutionCache {
     /// Guarded cloneable cache identified by a block hash.
     inner: Arc<RwLock<Option<SavedCache>>>,
     /// Metrics for cache operations.
@@ -1022,11 +1018,11 @@ pub(crate) struct PayloadExecutionCache {
 }
 
 impl PayloadExecutionCache {
-    /// Returns the cache backing store for `parent_hash` if it's available for reuse.
+    /// Returns the cache for `parent_hash` if it's available for use.
     ///
-    /// If the tracked cache is available but keyed to a different parent hash, the cache is
-    /// cleared and returned so callers can reuse the underlying allocations without carrying over
-    /// stale state.
+    /// A cache is considered available when:
+    /// - It exists and matches the requested parent hash
+    /// - No other tasks are currently using it (checked via Arc reference count)
     #[instrument(level = "debug", target = "engine::tree::payload_processor", skip(self))]
     pub(crate) fn get_cache_for(&self, parent_hash: B256) -> Option<SavedCache> {
         let start = Instant::now();
@@ -1065,7 +1061,7 @@ impl PayloadExecutionCache {
                     // and picking up polluted data if the fork block fails.
                     c.clear_with_hash(parent_hash);
                 }
-                return Some(c.clone());
+                return Some(c.clone())
             } else if hash_matches {
                 self.metrics.execution_cache_in_use.increment(1);
             }
@@ -1082,7 +1078,7 @@ impl PayloadExecutionCache {
     /// This is useful for synchronization before starting payload processing.
     ///
     /// Returns the time spent waiting for the lock.
-    pub(crate) fn wait_for_availability(&self) -> Duration {
+    pub fn wait_for_availability(&self) -> Duration {
         let start = Instant::now();
         // Acquire write lock to wait for any current holders to finish
         let _guard = self.inner.write();
@@ -1110,7 +1106,7 @@ impl PayloadExecutionCache {
     ///
     /// Violating this requirement can result in cache corruption, incorrect state data,
     /// and potential consensus failures.
-    pub(crate) fn update_with_guard<F>(&self, update_fn: F)
+    pub fn update_with_guard<F>(&self, update_fn: F)
     where
         F: FnOnce(&mut Option<SavedCache>),
     {
@@ -1179,9 +1175,8 @@ mod tests {
     use super::PayloadExecutionCache;
     use crate::tree::{
         cached_state::{CachedStateMetrics, ExecutionCache, SavedCache},
-        payload_processor::{
-            evm_state_to_hashed_post_state, EngineSharedCaches, ExecutionEnv, PayloadProcessor,
-        },
+        payload_processor::{evm_state_to_hashed_post_state, ExecutionEnv, PayloadProcessor},
+        precompile_cache::PrecompileCacheMap,
         StateProviderBuilder, TreeConfig,
     };
     use alloy_eips::eip1898::{BlockNumHash, BlockWithParent};
@@ -1293,7 +1288,7 @@ mod tests {
             reth_tasks::Runtime::test(),
             EthEvmConfig::new(Arc::new(ChainSpec::default())),
             &TreeConfig::default(),
-            EngineSharedCaches::default(),
+            PrecompileCacheMap::default(),
         );
 
         let parent_hash = B256::from([1u8; 32]);
@@ -1305,17 +1300,13 @@ mod tests {
         let bundle_state = BundleState::default();
 
         // Cache should be empty initially
-        assert!(payload_processor
-            .shared_caches
-            .execution_cache()
-            .get_cache_for(block_hash)
-            .is_none());
+        assert!(payload_processor.execution_cache.get_cache_for(block_hash).is_none());
 
         // Update cache with inserted block
         payload_processor.on_inserted_executed_block(block_with_parent, &bundle_state);
 
         // Cache should now exist for the block hash
-        let cached = payload_processor.shared_caches.execution_cache().get_cache_for(block_hash);
+        let cached = payload_processor.execution_cache.get_cache_for(block_hash);
         assert!(cached.is_some());
         assert_eq!(cached.unwrap().executed_block_hash(), block_hash);
     }
@@ -1326,14 +1317,13 @@ mod tests {
             reth_tasks::Runtime::test(),
             EthEvmConfig::new(Arc::new(ChainSpec::default())),
             &TreeConfig::default(),
-            EngineSharedCaches::default(),
+            PrecompileCacheMap::default(),
         );
 
         // Setup: populate cache with block 1
         let block1_hash = B256::from([1u8; 32]);
         payload_processor
-            .shared_caches
-            .execution_cache()
+            .execution_cache
             .update_with_guard(|slot| *slot = Some(make_saved_cache(block1_hash)));
 
         // Try to insert block 3 with wrong parent (should skip and keep block 1's cache)
@@ -1348,11 +1338,11 @@ mod tests {
         payload_processor.on_inserted_executed_block(block_with_parent, &bundle_state);
 
         // Cache should still be for block 1 (unchanged)
-        let cached = payload_processor.shared_caches.execution_cache().get_cache_for(block1_hash);
+        let cached = payload_processor.execution_cache.get_cache_for(block1_hash);
         assert!(cached.is_some(), "Original cache should be preserved");
 
         // Cache for block 3 should not exist
-        let cached3 = payload_processor.shared_caches.execution_cache().get_cache_for(block3_hash);
+        let cached3 = payload_processor.execution_cache.get_cache_for(block3_hash);
         assert!(cached3.is_none(), "New block cache should not be created on mismatch");
     }
 
@@ -1462,7 +1452,7 @@ mod tests {
             reth_tasks::Runtime::test(),
             EthEvmConfig::new(factory.chain_spec()),
             &TreeConfig::default(),
-            EngineSharedCaches::default(),
+            PrecompileCacheMap::default(),
         );
 
         let provider_factory = BlockchainProvider::new(factory).unwrap();

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

@@ -1,128 +1,44 @@
 //! Preserved sparse trie for reuse across payload validations.
 
-use super::{
-    PARALLEL_SPARSE_TRIE_PARALLELISM_THRESHOLDS, SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
-    SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
-};
 use alloy_primitives::B256;
 use parking_lot::Mutex;
-use reth_trie_sparse::{
-    ArenaParallelSparseTrie, ConfigurableSparseTrie, ParallelSparseTrie, RevealableSparseTrie,
-    SparseStateTrie,
-};
-use std::{
-    ops::{Deref, DerefMut},
-    sync::Arc,
-    time::{Duration, Instant},
-};
+use reth_trie_sparse::{ConfigurableSparseTrie, SparseStateTrie};
+use std::{sync::Arc, time::Instant};
 use tracing::debug;
 
 /// Type alias for the sparse trie type used in preservation.
-type SparseTrie = SparseStateTrie<ConfigurableSparseTrie, ConfigurableSparseTrie>;
+pub(super) type SparseTrie = SparseStateTrie<ConfigurableSparseTrie, ConfigurableSparseTrie>;
 
-/// Sparse trie implementation used by [`PayloadSparseTrieCache`].
-#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
-pub enum PayloadSparseTrieKind {
-    /// Back sparse trie storage with hash maps.
-    #[default]
-    HashMap,
-    /// Back sparse trie storage with arena allocations.
-    Arena,
-}
-
-impl From<bool> for PayloadSparseTrieKind {
-    fn from(enable_arena_sparse_trie: bool) -> Self {
-        if enable_arena_sparse_trie {
-            Self::Arena
-        } else {
-            Self::HashMap
-        }
-    }
-}
-
-#[derive(Debug, Default)]
-struct PayloadSparseTrieState {
-    latest_checkout_id: u64,
-    preserved: Option<PreservedSparseTrie>,
-}
-
-/// Outcome of storing a checked-out sparse trie back into the shared cache.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum PayloadSparseTrieStoreOutcome {
-    /// The checkout was the most recent lease and the trie was stored.
-    Stored,
-    /// A newer checkout had already been issued, so this stale lease was ignored.
-    IgnoredStaleCheckout,
-}
-
-/// Shared sparse trie cache that can be reused across payload validations.
+/// Shared handle to a preserved sparse trie that can be reused across payload validations.
 ///
-/// This is the public sparse-trie SDK surface exposed through
-/// [`EngineSharedCaches`](super::EngineSharedCaches). Callers take or create a trie, use it for
-/// payload work, then store it back either anchored to the resulting state root or cleared for
-/// allocation reuse.
-#[derive(Debug, Clone)]
-pub struct PayloadSparseTrieCache {
-    kind: PayloadSparseTrieKind,
-    state: Arc<Mutex<PayloadSparseTrieState>>,
-}
+/// This is stored in [`PayloadProcessor`](super::PayloadProcessor) and cloned to pass to
+/// [`SparseTrieCacheTask`](super::sparse_trie::SparseTrieCacheTask) for trie reuse.
+#[derive(Debug, Default, Clone)]
+pub(super) struct SharedPreservedSparseTrie(Arc<Mutex<Option<PreservedSparseTrie>>>);
 
-impl Default for PayloadSparseTrieCache {
-    fn default() -> Self {
-        Self::new(PayloadSparseTrieKind::default())
-    }
-}
-
-impl PayloadSparseTrieCache {
-    /// Creates a sparse trie cache backed by the requested trie implementation.
-    pub fn new(kind: PayloadSparseTrieKind) -> Self {
-        Self { kind, state: Arc::new(Mutex::new(PayloadSparseTrieState::default())) }
+impl SharedPreservedSparseTrie {
+    /// Takes the preserved trie if present, leaving `None` in its place.
+    pub(super) fn take(&self) -> Option<PreservedSparseTrie> {
+        self.0.lock().take()
     }
 
-    /// Returns the sparse trie implementation used when the cache needs to create a new trie.
-    pub const fn kind(&self) -> PayloadSparseTrieKind {
-        self.kind
-    }
-
-    /// Takes a preserved trie for `parent_state_root` or creates a new trie if the cache is empty.
-    pub fn take_or_create_for(&self, parent_state_root: B256) -> SparseTrieCheckout {
-        let start = Instant::now();
-        let mut state = self.state.lock();
-        state.latest_checkout_id += 1;
-        let checkout_id = state.latest_checkout_id;
-        let trie = state
-            .preserved
-            .take()
-            .map(|preserved| preserved.into_trie_for(parent_state_root))
-            .unwrap_or_else(|| {
-                debug!(
-                    target: "engine::tree::payload_processor",
-                    %parent_state_root,
-                    kind = ?self.kind,
-                    "Creating new sparse trie - no preserved trie available"
-                );
-                new_sparse_trie(self.kind)
-            });
-        drop(state);
-
-        let elapsed = start.elapsed();
-        if elapsed.as_millis() > 5 {
-            debug!(
-                target: "engine::tree::payload_processor",
-                blocked_for=?elapsed,
-                "Waited for preserved sparse trie checkout"
-            );
-        }
-
-        SparseTrieCheckout { trie: Some(trie), cache: self.clone(), checkout_id }
+    /// 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())
     }
 
     /// Waits until the sparse trie lock becomes available.
     ///
+    /// This acquires and immediately releases the lock, ensuring that any
+    /// ongoing operations complete before returning. Useful for synchronization
+    /// before starting payload processing.
+    ///
     /// Returns the time spent waiting for the lock.
-    pub fn wait_for_availability(&self) -> Duration {
+    pub(super) fn wait_for_availability(&self) -> std::time::Duration {
         let start = Instant::now();
-        let _guard = self.state.lock();
+        let _guard = self.0.lock();
         let elapsed = start.elapsed();
         if elapsed.as_millis() > 5 {
             debug!(
@@ -133,142 +49,27 @@ impl PayloadSparseTrieCache {
         }
         elapsed
     }
-
-    /// Acquires a guard that blocks cache mutation until dropped.
-    ///
-    /// Engine-internal code uses this before making the state-root result visible so the next
-    /// payload cannot observe an empty cache between send and store.
-    pub(super) fn lock(&self) -> PreservedTrieGuard<'_> {
-        PreservedTrieGuard { state: self.state.lock() }
-    }
-}
-
-/// A checked-out sparse trie lease.
-///
-/// This dereferences to [`SparseStateTrie`] so callers can reuse the trie directly. If the lease is
-/// dropped without being stored back, a cleared trie is returned to the shared cache unless a newer
-/// checkout has already superseded it.
-#[derive(Debug)]
-pub struct SparseTrieCheckout {
-    trie: Option<SparseTrie>,
-    cache: PayloadSparseTrieCache,
-    checkout_id: u64,
-}
-
-impl SparseTrieCheckout {
-    /// Stores the trie back into the shared cache anchored to the given state root.
-    pub fn store_anchored(self, state_root: B256) -> PayloadSparseTrieStoreOutcome {
-        let cache = self.cache.clone();
-        let mut guard = cache.lock();
-        self.store_anchored_with_guard(&mut guard, state_root)
-    }
-
-    /// Stores the trie back into the shared cache in a cleared state.
-    pub fn store_cleared(mut self) -> PayloadSparseTrieStoreOutcome {
-        let cache = self.cache.clone();
-        let mut trie = self.take_trie();
-        prepare_cleared_trie(&mut trie);
-        let deferred = trie.take_deferred_drops();
-        let mut guard = cache.lock();
-        let outcome = guard.store(self.checkout_id, PreservedSparseTrie::cleared(trie));
-        drop(guard);
-        drop(deferred);
-        outcome
-    }
-
-    /// Stores the trie back into the shared cache anchored to the given state root while the
-    /// caller is already holding the preservation lock.
-    pub(super) fn store_anchored_with_guard(
-        mut self,
-        guard: &mut PreservedTrieGuard<'_>,
-        state_root: B256,
-    ) -> PayloadSparseTrieStoreOutcome {
-        guard.store(self.checkout_id, PreservedSparseTrie::anchored(self.take_trie(), state_root))
-    }
-
-    /// Stores an already-cleared trie back into the shared cache while the caller is already
-    /// holding the preservation lock.
-    pub(super) fn store_prepared_cleared_with_guard(
-        mut self,
-        guard: &mut PreservedTrieGuard<'_>,
-    ) -> PayloadSparseTrieStoreOutcome {
-        guard.store(self.checkout_id, PreservedSparseTrie::cleared(self.take_trie()))
-    }
-
-    fn take_trie(&mut self) -> SparseTrie {
-        self.trie.take().expect("sparse trie checkout must hold a trie until it is stored")
-    }
-}
-
-impl Deref for SparseTrieCheckout {
-    type Target = SparseTrie;
-
-    fn deref(&self) -> &Self::Target {
-        self.trie.as_ref().expect("sparse trie checkout must hold a trie until it is stored")
-    }
-}
-
-impl DerefMut for SparseTrieCheckout {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        self.trie.as_mut().expect("sparse trie checkout must hold a trie until it is stored")
-    }
-}
-
-impl Drop for SparseTrieCheckout {
-    fn drop(&mut self) {
-        let Some(mut trie) = self.trie.take() else { return };
-
-        debug!(
-            target: "engine::tree::payload_processor",
-            checkout_id = self.checkout_id,
-            "Sparse trie checkout dropped before store, returning cleared trie to cache"
-        );
-
-        prepare_cleared_trie(&mut trie);
-        let deferred = trie.take_deferred_drops();
-        let mut guard = self.cache.lock();
-        let _ = guard.store(self.checkout_id, PreservedSparseTrie::cleared(trie));
-        drop(guard);
-        drop(deferred);
-    }
 }
 
 /// Guard that holds the lock on the preserved trie.
-/// While held, take-or-create calls will block. Call `store()` to save the trie before dropping.
-pub(super) struct PreservedTrieGuard<'a> {
-    state: parking_lot::MutexGuard<'a, PayloadSparseTrieState>,
-}
+/// 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 if the checkout is still current.
-    fn store(
-        &mut self,
-        checkout_id: u64,
-        trie: PreservedSparseTrie,
-    ) -> PayloadSparseTrieStoreOutcome {
-        if checkout_id != self.state.latest_checkout_id {
-            debug!(
-                target: "engine::tree::payload_processor",
-                checkout_id,
-                latest_checkout_id = self.state.latest_checkout_id,
-                "Ignoring stale sparse trie checkout"
-            );
-            return PayloadSparseTrieStoreOutcome::IgnoredStaleCheckout;
-        }
-
-        self.state.preserved.replace(trie);
-        PayloadSparseTrieStoreOutcome::Stored
+    /// 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.
+/// - **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)]
-enum PreservedSparseTrie {
+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).
@@ -286,17 +87,24 @@ enum PreservedSparseTrie {
 
 impl PreservedSparseTrie {
     /// Creates a new anchored preserved trie.
-    const fn anchored(trie: SparseTrie, state_root: B256) -> Self {
+    ///
+    /// 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).
-    const fn cleared(trie: SparseTrie) -> Self {
+    pub(super) const fn cleared(trie: SparseTrie) -> Self {
         Self::Cleared { trie }
     }
 
     /// Consumes self and returns the trie for reuse.
-    fn into_trie_for(self, parent_state_root: B256) -> SparseTrie {
+    ///
+    /// 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!(
@@ -327,111 +135,3 @@ impl PreservedSparseTrie {
         }
     }
 }
-
-fn new_sparse_trie(kind: PayloadSparseTrieKind) -> SparseTrie {
-    let default_trie = match kind {
-        PayloadSparseTrieKind::HashMap => {
-            RevealableSparseTrie::blind_from(ConfigurableSparseTrie::HashMap(
-                ParallelSparseTrie::default()
-                    .with_parallelism_thresholds(PARALLEL_SPARSE_TRIE_PARALLELISM_THRESHOLDS),
-            ))
-        }
-        PayloadSparseTrieKind::Arena => RevealableSparseTrie::blind_from(
-            ConfigurableSparseTrie::Arena(ArenaParallelSparseTrie::default()),
-        ),
-    };
-
-    SparseStateTrie::default()
-        .with_accounts_trie(default_trie.clone())
-        .with_default_storage_trie(default_trie)
-        .with_updates(true)
-}
-
-fn prepare_cleared_trie(trie: &mut SparseTrie) {
-    trie.clear();
-    trie.shrink_to(SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY, SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY);
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn take_or_create_reuses_matching_anchor() {
-        let cache = PayloadSparseTrieCache::default();
-        let state_root = B256::with_last_byte(1);
-
-        assert_eq!(
-            cache.take_or_create_for(state_root).store_anchored(state_root),
-            PayloadSparseTrieStoreOutcome::Stored
-        );
-
-        match cache.state.lock().preserved.as_ref() {
-            Some(PreservedSparseTrie::Anchored { state_root: anchored, .. }) => {
-                assert_eq!(*anchored, state_root);
-            }
-            other => panic!("expected anchored trie, got {other:?}"),
-        }
-    }
-
-    #[test]
-    fn drop_restores_cleared_trie() {
-        let cache = PayloadSparseTrieCache::default();
-        let state_root = B256::with_last_byte(2);
-
-        let mut checkout = cache.take_or_create_for(state_root);
-        checkout.set_updates(true);
-        drop(checkout);
-
-        match cache.state.lock().preserved.as_ref() {
-            Some(PreservedSparseTrie::Cleared { .. }) => {}
-            other => panic!("expected cleared trie, got {other:?}"),
-        }
-    }
-
-    #[test]
-    fn stale_checkout_does_not_overwrite_newer_store() {
-        let cache = PayloadSparseTrieCache::default();
-        let parent_state_root = B256::with_last_byte(3);
-        let anchored_state_root = B256::with_last_byte(4);
-
-        let stale = cache.take_or_create_for(parent_state_root);
-        let fresh = cache.take_or_create_for(parent_state_root);
-
-        assert_eq!(
-            fresh.store_anchored(anchored_state_root),
-            PayloadSparseTrieStoreOutcome::Stored
-        );
-        assert_eq!(stale.store_cleared(), PayloadSparseTrieStoreOutcome::IgnoredStaleCheckout);
-
-        match cache.state.lock().preserved.as_ref() {
-            Some(PreservedSparseTrie::Anchored { state_root, .. }) => {
-                assert_eq!(*state_root, anchored_state_root);
-            }
-            other => panic!("expected anchored trie to survive stale checkout, got {other:?}"),
-        }
-    }
-
-    #[test]
-    fn stale_checkout_drop_does_not_overwrite_newer_store() {
-        let cache = PayloadSparseTrieCache::default();
-        let parent_state_root = B256::with_last_byte(5);
-        let anchored_state_root = B256::with_last_byte(6);
-
-        let stale = cache.take_or_create_for(parent_state_root);
-        let fresh = cache.take_or_create_for(parent_state_root);
-
-        assert_eq!(
-            fresh.store_anchored(anchored_state_root),
-            PayloadSparseTrieStoreOutcome::Stored
-        );
-        drop(stale);
-
-        match cache.state.lock().preserved.as_ref() {
-            Some(PreservedSparseTrie::Anchored { state_root, .. }) => {
-                assert_eq!(*state_root, anchored_state_root);
-            }
-            other => panic!("expected anchored trie to survive stale checkout drop, got {other:?}"),
-        }
-    }
-}

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

@@ -84,7 +84,7 @@ where
     Evm: ConfigureEvm<Primitives = N> + 'static,
 {
     /// Initializes the task with the given transactions pending execution
-    pub(crate) fn new(
+    pub fn new(
         executor: Runtime,
         execution_cache: PayloadExecutionCache,
         ctx: PrewarmContext<N, P, Evm>,

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

@@ -7,7 +7,7 @@ use crate::tree::{
         dispatch_with_chunking, evm_state_to_hashed_post_state, MultiProofMessage,
         DEFAULT_MAX_TARGETS_FOR_CHUNKING,
     },
-    payload_processor::{multiproof::MultiProofTaskMetrics, SparseTrieCheckout},
+    payload_processor::multiproof::MultiProofTaskMetrics,
 };
 use alloy_primitives::B256;
 use alloy_rlp::{Decodable, Encodable};
@@ -30,7 +30,7 @@ use reth_trie_parallel::{
 use reth_trie_sparse::debug_recorder::TrieDebugRecorder;
 use reth_trie_sparse::{
     errors::SparseTrieResult, ConfigurableSparseTrie, DeferredDrops, LeafUpdate,
-    RevealableSparseTrie,
+    RevealableSparseTrie, SparseStateTrie, SparseTrie,
 };
 use revm_primitives::{hash_map::Entry, B256Map};
 use tracing::{debug, debug_span, error, instrument, trace_span};
@@ -39,7 +39,7 @@ use tracing::{debug, debug_span, error, instrument, trace_span};
 const MAX_PENDING_UPDATES: usize = 100;
 
 /// Sparse trie task implementation that uses in-memory sparse trie data to schedule proof fetching.
-pub(super) struct SparseTrieCacheTask {
+pub(super) struct SparseTrieCacheTask<A = ConfigurableSparseTrie, S = ConfigurableSparseTrie> {
     /// Sender for proof results.
     proof_result_tx: CrossbeamSender<ProofResultMessage>,
     /// Receiver for proof results directly from workers.
@@ -47,7 +47,7 @@ pub(super) struct SparseTrieCacheTask {
     /// Receives updates from execution and prewarming.
     updates: CrossbeamReceiver<SparseTrieTaskMessage>,
     /// `SparseStateTrie` used for computing the state root.
-    trie: SparseTrieCheckout,
+    trie: SparseStateTrie<A, S>,
     /// Handle to the proof worker pools (storage and account).
     proof_worker_handle: ProofWorkerHandle,
 
@@ -110,14 +110,18 @@ pub(super) struct SparseTrieCacheTask {
     metrics: MultiProofTaskMetrics,
 }
 
-impl SparseTrieCacheTask {
+impl<A, S> SparseTrieCacheTask<A, S>
+where
+    A: SparseTrie + Default,
+    S: SparseTrie + Default + Clone,
+{
     /// Creates a new sparse trie, pre-populating with an existing [`SparseStateTrie`].
-    pub(super) fn new_with_checkout(
+    pub(super) fn new_with_trie(
         executor: &Runtime,
         updates: CrossbeamReceiver<MultiProofMessage>,
         proof_worker_handle: ProofWorkerHandle,
         metrics: MultiProofTaskMetrics,
-        trie: SparseTrieCheckout,
+        trie: SparseStateTrie<A, S>,
         chunk_size: usize,
     ) -> Self {
         let (proof_result_tx, proof_result_rx) = crossbeam_channel::unbounded();
@@ -201,7 +205,7 @@ impl SparseTrieCacheTask {
         max_values_capacity: usize,
         disable_pruning: bool,
         updates: &TrieUpdates,
-    ) -> (SparseTrieCheckout, DeferredDrops) {
+    ) -> (SparseStateTrie<A, S>, DeferredDrops) {
         let Self { mut trie, .. } = self;
         trie.commit_updates(updates);
         if !disable_pruning {
@@ -220,7 +224,7 @@ impl SparseTrieCacheTask {
         self,
         max_nodes_capacity: usize,
         max_values_capacity: usize,
-    ) -> (SparseTrieCheckout, DeferredDrops) {
+    ) -> (SparseStateTrie<A, S>, DeferredDrops) {
         let Self { mut trie, .. } = self;
         trie.clear();
         trie.shrink_to(max_nodes_capacity, max_values_capacity);
@@ -302,14 +306,20 @@ impl SparseTrieCacheTask {
                 self.promote_pending_account_updates()?;
                 self.metrics.sparse_trie_process_updates_duration_histogram.record(t.elapsed());
 
-                if self.finished_state_updates
-                    && self.account_updates.is_empty()
-                    && self.storage_updates.iter().all(|(_, updates)| updates.is_empty())
+                if self.finished_state_updates &&
+                    self.account_updates.is_empty() &&
+                    self.storage_updates.iter().all(|(_, updates)| updates.is_empty())
                 {
                     break;
                 }
 
                 self.dispatch_pending_targets();
+
+                // If there's still no pending updates spend some time pre-computing the account
+                // trie upper hashes
+                if self.proof_result_rx.is_empty() {
+                    self.trie.calculate_subtries();
+                }
             } else if self.updates.is_empty() || self.pending_updates > MAX_PENDING_UPDATES {
                 // If we don't have any pending updates OR we've accumulated a lot already, apply
                 // them to the trie,
@@ -596,6 +606,7 @@ impl SparseTrieCacheTask {
 
         Ok(updates_len_after < updates_len_before)
     }
+
     /// Computes storage roots for accounts whose storage updates are fully drained.
     ///
     /// For each storage trie T that:
@@ -616,16 +627,16 @@ impl SparseTrieCacheTask {
             .filter_map(|(address, updates)| updates.is_empty().then_some(*address))
             .collect();
 
-        struct SendStorageTriePtr(*mut RevealableSparseTrie<ConfigurableSparseTrie>);
+        struct SendStorageTriePtr<S>(*mut RevealableSparseTrie<S>);
         // SAFETY: this wrapper only forwards the pointer across rayon; deref invariants are
         // documented at the use site below.
-        unsafe impl Send for SendStorageTriePtr {}
+        unsafe impl<S: Send> Send for SendStorageTriePtr<S> {}
 
-        let mut tries_to_compute_roots: Vec<(B256, SendStorageTriePtr)> =
+        let mut tries_to_compute_roots: Vec<(B256, SendStorageTriePtr<S>)> =
             Vec::with_capacity(addresses_to_compute_roots.len());
         for address in addresses_to_compute_roots {
-            if let Some(trie) = self.trie.storage_tries_mut().get_mut(&address)
-                && !trie.is_root_cached()
+            if let Some(trie) = self.trie.storage_tries_mut().get_mut(&address) &&
+                !trie.is_root_cached()
             {
                 tries_to_compute_roots.push((address, SendStorageTriePtr(trie)));
             }
@@ -724,7 +735,7 @@ impl SparseTrieCacheTask {
             // We need to keep iterating if any updates are being drained because that might
             // indicate that more pending account updates can be promoted.
             if num_promoted == 0 || !self.process_account_leaf_updates(false)? {
-                break;
+                break
             }
         }
 
@@ -845,6 +856,7 @@ pub struct StateRootComputeOutcome {
 mod tests {
     use super::*;
     use alloy_primitives::{keccak256, Address, B256, U256};
+    use reth_trie_sparse::ArenaParallelSparseTrie;
 
     #[test]
     fn test_run_hashing_task_hashed_state_update_forwards() {
@@ -867,7 +879,10 @@ mod tests {
         let expected_state = hashed_state.clone();
 
         let handle = std::thread::spawn(move || {
-            SparseTrieCacheTask::run_hashing_task(updates_rx, hashed_state_tx);
+            SparseTrieCacheTask::<ArenaParallelSparseTrie, ArenaParallelSparseTrie>::run_hashing_task(
+                updates_rx,
+                hashed_state_tx,
+            );
         });
 
         updates_tx.send(MultiProofMessage::HashedStateUpdate(hashed_state)).unwrap();

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

@@ -4,7 +4,7 @@ use crate::tree::{
     cached_state::{CacheStats, CachedStateProvider},
     error::{InsertBlockError, InsertBlockErrorKind, InsertPayloadError},
     instrumented_state::{InstrumentedStateProvider, StateProviderStats},
-    payload_processor::{EngineSharedCaches, PayloadProcessor},
+    payload_processor::PayloadProcessor,
     precompile_cache::{CachedPrecompile, CachedPrecompileMetrics, PrecompileCacheMap},
     sparse_trie::StateRootComputeOutcome,
     CacheWaitDurations, EngineApiMetrics, EngineApiTreeState, ExecutionEnv, PayloadHandle,
@@ -190,13 +190,16 @@ where
         validator: V,
         config: TreeConfig,
         invalid_block_hook: Box<dyn InvalidBlockHook<N>>,
-        shared_caches: EngineSharedCaches<Evm>,
         changeset_cache: ChangesetCache,
         runtime: reth_tasks::Runtime,
     ) -> Self {
-        let precompile_cache_map = shared_caches.precompile_cache_map();
-        let payload_processor =
-            PayloadProcessor::new(runtime.clone(), evm_config.clone(), &config, shared_caches);
+        let precompile_cache_map = PrecompileCacheMap::default();
+        let payload_processor = PayloadProcessor::new(
+            runtime.clone(),
+            evm_config.clone(),
+            &config,
+            precompile_cache_map.clone(),
+        );
         Self {
             provider,
             consensus,
@@ -310,7 +313,7 @@ where
         // Validate block consensus rules which includes header validation
         if let Err(consensus_err) = self.validate_block_inner(&block, None) {
             // Header validation error takes precedence over execution error
-            return Err(InsertBlockError::new(block, consensus_err.into()).into());
+            return Err(InsertBlockError::new(block, consensus_err.into()).into())
         }
 
         // Also validate against the parent
@@ -318,7 +321,7 @@ where
             self.consensus.validate_header_against_parent(block.sealed_header(), parent_block)
         {
             // Parent validation error takes precedence over execution error
-            return Err(InsertBlockError::new(block, consensus_err.into()).into());
+            return Err(InsertBlockError::new(block, consensus_err.into()).into())
         }
 
         // No header validation errors, return the original execution error
@@ -393,7 +396,7 @@ where
                     Ok(val) => val,
                     Err(e) => {
                         let block = convert_to_block(input)?;
-                        return Err(InsertBlockError::new(block, e.into()).into());
+                        return Err(InsertBlockError::new(block, e.into()).into())
                     }
                 }
             };
@@ -426,7 +429,7 @@ where
                 convert_to_block(input)?,
                 ProviderError::HeaderNotFound(parent_hash.into()).into(),
             )
-            .into());
+            .into())
         };
         let mut state_provider = ensure_ok!(provider_builder.build());
         drop(_enter);
@@ -439,7 +442,7 @@ where
                 convert_to_block(input)?,
                 ProviderError::HeaderNotFound(parent_hash.into()).into(),
             )
-            .into());
+            .into())
         };
 
         let evm_env = debug_span!(target: "engine::tree::payload_validator", "evm_env")
@@ -759,7 +762,7 @@ where
                 )
                 .into(),
             )
-            .into());
+            .into())
         }
 
         let timing_stats = state_provider_stats.map(|stats| {
@@ -821,14 +824,14 @@ where
     ) -> Result<(), ConsensusError> {
         if let Err(e) = self.consensus.validate_header(block.sealed_header()) {
             error!(target: "engine::tree::payload_validator", ?block, "Failed to validate header {}: {e}", block.hash());
-            return Err(e);
+            return Err(e)
         }
 
         if let Err(e) =
             self.consensus.validate_block_pre_execution_with_tx_root(block, transaction_root)
         {
             error!(target: "engine::tree::payload_validator", ?block, "Failed to validate block {}: {e}", block.hash());
-            return Err(e);
+            return Err(e)
         }
 
         Ok(())
@@ -1320,7 +1323,7 @@ where
         trace!(target: "engine::tree::payload_validator", block=?block.num_hash(), "Validating block consensus");
         // validate block consensus rules
         if let Err(e) = self.validate_block_inner(block, transaction_root) {
-            return Err(e.into());
+            return Err(e.into())
         }
 
         // now validate against the parent
@@ -1329,7 +1332,7 @@ where
             self.consensus.validate_header_against_parent(block.sealed_header(), parent_block)
         {
             warn!(target: "engine::tree::payload_validator", ?block, "Failed to validate header {} against parent: {e}", block.hash());
-            return Err(e.into());
+            return Err(e.into())
         }
         drop(_enter);
 
@@ -1342,7 +1345,7 @@ where
         {
             // call post-block hook
             self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
-            return Err(err.into());
+            return Err(err.into())
         }
         drop(_enter);
 
@@ -1358,7 +1361,7 @@ where
         {
             // call post-block hook
             self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
-            return Err(err.into());
+            return Err(err.into())
         }
 
         // record post-execution validation duration
@@ -1466,7 +1469,7 @@ where
                 self.provider.clone(),
                 historical,
                 Some(blocks),
-            )));
+            )))
         }
 
         // Check if the block is persisted
@@ -1474,7 +1477,7 @@ where
             debug!(target: "engine::tree::payload_validator", %hash, number = %header.number(), "found canonical state for block in database, creating provider builder");
             // For persisted blocks, we create a builder that will fetch state directly from the
             // database
-            return Ok(Some(StateProviderBuilder::new(self.provider.clone(), hash, None)));
+            return Ok(Some(StateProviderBuilder::new(self.provider.clone(), hash, None)))
         }
 
         debug!(target: "engine::tree::payload_validator", %hash, "no canonical state found for block");
@@ -1506,7 +1509,7 @@ where
     ) {
         if state.invalid_headers.get(&block.hash()).is_some() {
             // we already marked this block as invalid
-            return;
+            return
         }
         self.invalid_block_hook.on_invalid_block(parent_header, block, output, trie_updates);
     }

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

@@ -74,6 +74,11 @@ impl<N: NodePrimitives> TreeState<N> {
         self.blocks_by_hash.get(&hash)
     }
 
+    /// Returns `true` if a block with the given hash exists in memory.
+    pub fn contains_hash(&self, hash: &B256) -> bool {
+        self.blocks_by_hash.contains_key(hash)
+    }
+
     /// Returns the sealed block header by hash.
     pub 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())

crates/engine/tree/src/tree/tests.rs

@@ -203,7 +203,6 @@ impl TestHarness {
             payload_validator,
             TreeConfig::default(),
             Box::new(NoopInvalidBlockHook::default()),
-            EngineSharedCaches::default(),
             changeset_cache.clone(),
             reth_tasks::Runtime::test(),
         );
@@ -408,7 +407,6 @@ impl ValidatorTestHarness {
             payload_validator,
             TreeConfig::default(),
             Box::new(NoopInvalidBlockHook::default()),
-            EngineSharedCaches::default(),
             changeset_cache,
             reth_tasks::Runtime::test(),
         );

crates/engine/tree/tests/shared_caches_sdk.rs

@@ -1,27 +0,0 @@
-//! SDK smoke tests for `EngineSharedCaches`.
-
-use alloy_primitives::B256;
-use reth_engine_tree::tree::{
-    EngineSharedCaches, PayloadSparseTrieKind, PayloadSparseTrieStoreOutcome,
-};
-use reth_evm_ethereum::EthEvmConfig;
-
-#[test]
-fn engine_shared_caches_exposes_public_sparse_trie_sdk() {
-    let caches =
-        EngineSharedCaches::<EthEvmConfig>::with_sparse_trie_kind(PayloadSparseTrieKind::Arena);
-
-    let _precompile_cache_map = caches.precompile_cache_map();
-
-    let sparse_trie_cache = caches.sparse_trie_cache();
-    assert_eq!(sparse_trie_cache.kind(), PayloadSparseTrieKind::Arena);
-    let state_root = B256::with_last_byte(1);
-
-    assert_eq!(
-        sparse_trie_cache.take_or_create_for(state_root).store_anchored(state_root),
-        PayloadSparseTrieStoreOutcome::Stored
-    );
-
-    let checkout = sparse_trie_cache.take_or_create_for(state_root);
-    assert!(checkout.memory_size() > 0 || checkout.retained_storage_tries_count() == 0);
-}

crates/era-downloader/src/client.rs

@@ -4,7 +4,6 @@ use eyre::{eyre, OptionExt};
 use futures_util::{stream::StreamExt, Stream, TryStreamExt};
 use reqwest::{Client, IntoUrl, Url};
 use reth_era::common::file_ops::EraFileType;
-use reth_fs_util::FsPathError;
 use sha2::{Digest, Sha256};
 use std::{future::Future, path::Path, str::FromStr};
 use tokio::{
@@ -137,7 +136,7 @@ impl<Http: HttpClient + Clone> EraClient<Http> {
                     let Some(number) = self.file_name_to_number(name) &&
                     (number < index || number >= last)
                 {
-                    remove_file_ignore_not_found(entry.path())?;
+                    reth_fs_util::remove_file_if_exists(entry.path())?;
                 }
             }
         }
@@ -322,16 +321,6 @@ impl<Http: HttpClient + Clone> EraClient<Http> {
     }
 }
 
-fn remove_file_ignore_not_found(path: impl AsRef<Path>) -> eyre::Result<()> {
-    match reth_fs_util::remove_file(path) {
-        Ok(()) => Ok(()),
-        Err(FsPathError::RemoveFile { source, .. }) if source.kind() == io::ErrorKind::NotFound => {
-            Ok(())
-        }
-        Err(err) => Err(err.into()),
-    }
-}
-
 async fn checksum(mut reader: impl AsyncRead + Unpin) -> eyre::Result<Vec<u8>> {
     let mut hasher = Sha256::new();
 
@@ -378,25 +367,4 @@ mod tests {
 
         assert_eq!(actual_number, expected_number);
     }
-
-    #[test]
-    fn test_remove_file_ignore_not_found() {
-        let temp_dir = tempfile::tempdir().unwrap();
-        let path = temp_dir.path().join("missing.era1");
-
-        assert!(remove_file_ignore_not_found(&path).is_ok());
-    }
-
-    #[test]
-    fn test_remove_file_ignore_not_found_preserves_other_errors() {
-        let temp_dir = tempfile::tempdir().unwrap();
-        let path = temp_dir.path().join("dir");
-        std::fs::create_dir_all(&path).unwrap();
-
-        let err = remove_file_ignore_not_found(&path).unwrap_err();
-        assert!(matches!(
-            err.downcast_ref::<FsPathError>(),
-            Some(FsPathError::RemoveFile { source, .. }) if source.kind() != io::ErrorKind::NotFound
-        ));
-    }
 }

crates/fs-util/src/lib.rs

@@ -260,6 +260,17 @@ pub fn remove_file(path: impl AsRef<Path>) -> Result<()> {
     fs::remove_file(path).map_err(|err| FsPathError::remove_file(err, path))
 }
 
+/// Removes a file at the given path, ignoring the error if the file does not exist
+/// (`ErrorKind::NotFound`).
+pub fn remove_file_if_exists(path: impl AsRef<Path>) -> Result<()> {
+    match remove_file(&path) {
+        Err(FsPathError::RemoveFile { source, .. }) if source.kind() == io::ErrorKind::NotFound => {
+            Ok(())
+        }
+        result => result,
+    }
+}
+
 /// Wrapper for `std::fs::create_dir_all`
 pub fn create_dir_all(path: impl AsRef<Path>) -> Result<()> {
     let path = path.as_ref();

crates/net/network/src/transactions/mod.rs

@@ -473,8 +473,14 @@ impl<Pool: TransactionPool, N: NetworkPrimitives> TransactionsManager<Pool, N> {
         self.transactions_by_peers.remove(&hash);
     }
 
-    /// Penalize the peers that intentionally sent the bad transaction, and cache it to avoid
-    /// fetching or importing it again.
+    /// Handles a failed transaction import.
+    ///
+    /// Blob sidecar errors (e.g. invalid proof, missing sidecar) are penalized via
+    /// `report_peer_bad_transactions` but NOT cached in `bad_imports` — the transaction itself
+    /// may be valid when fetched from another peer with correct sidecar data.
+    ///
+    /// Other bad transactions are penalized and cached in `bad_imports` to avoid fetching or
+    /// importing them again.
     ///
     /// Errors that count as bad transactions are:
     ///
@@ -499,6 +505,18 @@ impl<Pool: TransactionPool, N: NetworkPrimitives> TransactionsManager<Pool, N> {
     fn on_bad_import(&mut self, err: PoolError) {
         let peers = self.transactions_by_peers.remove(&err.hash);
 
+        if err.is_bad_blob_sidecar() {
+            // Blob sidecar errors: penalize but do NOT cache the hash as bad.
+            // The transaction may be valid — only the sidecar from this peer was wrong.
+            // Using regular penalties means repeated offenders still get disconnected.
+            if let Some(peers) = peers {
+                for peer_id in peers {
+                    self.report_peer_bad_transactions(peer_id);
+                }
+            }
+            return
+        }
+
         // if we're _currently_ syncing, we ignore a bad transaction
         if !err.is_bad_transaction() || self.network.is_syncing() {
             return
@@ -2169,6 +2187,7 @@ mod tests {
         NetworkConfigBuilder, NetworkManager,
     };
     use alloy_consensus::{TxEip1559, TxLegacy};
+    use alloy_eips::eip4844::BlobTransactionValidationError;
     use alloy_primitives::{hex, Signature, TxKind, B256, U256};
     use alloy_rlp::Decodable;
     use futures::FutureExt;
@@ -2181,11 +2200,13 @@ mod tests {
     };
     use reth_storage_api::noop::NoopProvider;
     use reth_tasks::Runtime;
-    use reth_transaction_pool::test_utils::{
-        testing_pool, MockTransaction, MockTransactionFactory, TestPool,
+    use reth_transaction_pool::{
+        error::{Eip4844PoolTransactionError, InvalidPoolTransactionError, PoolError},
+        test_utils::{testing_pool, MockTransaction, MockTransactionFactory, TestPool},
     };
     use secp256k1::SecretKey;
     use std::{
+        collections::HashSet,
         future::poll_fn,
         net::{IpAddr, Ipv4Addr, SocketAddr},
         str::FromStr,
@@ -2553,6 +2574,67 @@ mod tests {
         );
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_bad_blob_sidecar_not_cached_as_bad_import() {
+        let (mut tx_manager, _network) = new_tx_manager().await;
+        let peer_id = PeerId::new([1; 64]);
+        let hash = B256::from_slice(&[1; 32]);
+
+        tx_manager.network.update_sync_state(SyncState::Idle);
+        tx_manager.transactions_by_peers.insert(hash, HashSet::from([peer_id]));
+
+        let err = PoolError::new(
+            hash,
+            InvalidPoolTransactionError::Eip4844(Eip4844PoolTransactionError::InvalidEip4844Blob(
+                BlobTransactionValidationError::InvalidProof,
+            )),
+        );
+
+        tx_manager.on_bad_import(err);
+
+        assert!(!tx_manager.bad_imports.contains(&hash));
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_missing_blob_sidecar_not_cached_as_bad_import() {
+        let (mut tx_manager, _network) = new_tx_manager().await;
+        let peer_id = PeerId::new([1; 64]);
+        let hash = B256::from_slice(&[3; 32]);
+
+        tx_manager.network.update_sync_state(SyncState::Idle);
+        tx_manager.transactions_by_peers.insert(hash, HashSet::from([peer_id]));
+
+        let err = PoolError::new(
+            hash,
+            InvalidPoolTransactionError::Eip4844(
+                Eip4844PoolTransactionError::MissingEip4844BlobSidecar,
+            ),
+        );
+
+        tx_manager.on_bad_import(err);
+
+        assert!(!tx_manager.bad_imports.contains(&hash));
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_non_blob_sidecar_error_still_cached_as_bad_import() {
+        let (mut tx_manager, _network) = new_tx_manager().await;
+        let peer_id = PeerId::new([1; 64]);
+        let hash = B256::from_slice(&[2; 32]);
+
+        tx_manager.network.update_sync_state(SyncState::Idle);
+        tx_manager.transactions_by_peers.insert(hash, HashSet::from([peer_id]));
+
+        let err = PoolError::new(
+            hash,
+            InvalidPoolTransactionError::Eip4844(Eip4844PoolTransactionError::NoEip4844Blobs),
+        );
+
+        tx_manager.on_bad_import(err);
+
+        assert!(tx_manager.bad_imports.contains(&hash));
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn test_on_get_pooled_transactions_network() {
         reth_tracing::init_test_tracing();

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

@@ -15,7 +15,7 @@ use reth_engine_tree::{
     chain::{ChainEvent, FromOrchestrator},
     engine::{EngineApiKind, EngineApiRequest, EngineRequestHandler},
     launch::build_engine_orchestrator,
-    tree::{EngineSharedCaches, PayloadSparseTrieKind, TreeConfig},
+    tree::TreeConfig,
 };
 use reth_engine_util::EngineMessageStreamExt;
 use reth_exex::ExExManagerHandle;
@@ -90,10 +90,6 @@ impl EngineNodeLauncher {
 
         // Create changeset cache that will be shared across the engine
         let changeset_cache = ChangesetCache::new();
-        let main_shared_caches =
-            EngineSharedCaches::<<CB::Components as NodeComponents<T>>::Evm>::with_sparse_trie_kind(
-                PayloadSparseTrieKind::from(engine_tree_config.enable_arena_sparse_trie()),
-            );
 
         // setup the launch context
         let ctx = ctx
@@ -127,8 +123,7 @@ impl EngineNodeLauncher {
             .with_blockchain_db::<T, _>(move |provider_factory| {
                 Ok(BlockchainProvider::new(provider_factory)?)
             })?
-            .with_components(components_builder, on_component_initialized)
-            .await?;
+            .with_components(components_builder, on_component_initialized).await?;
 
         // spawn exexs if any
         let maybe_exex_manager_handle = ctx.launch_exex(installed_exex).await?;
@@ -199,12 +194,7 @@ impl EngineNodeLauncher {
         // Build the engine validator with all required components
         let engine_validator = validator_builder
             .clone()
-            .build_tree_validator_with_caches(
-                &add_ons_ctx,
-                engine_tree_config.clone(),
-                changeset_cache.clone(),
-                main_shared_caches.clone(),
-            )
+            .build_tree_validator(&add_ons_ctx, engine_tree_config.clone(), changeset_cache.clone())
             .await?;
 
         // Create the consensus engine stream with optional reorg
@@ -217,18 +207,8 @@ impl EngineNodeLauncher {
                 || async {
                     // Create a separate cache for reorg validator (not shared with main engine)
                     let reorg_cache = ChangesetCache::new();
-                    let reorg_shared_caches = EngineSharedCaches::<
-                        <CB::Components as NodeComponents<T>>::Evm,
-                    >::with_sparse_trie_kind(
-                        PayloadSparseTrieKind::from(engine_tree_config.enable_arena_sparse_trie()),
-                    );
                     validator_builder
-                        .build_tree_validator_with_caches(
-                            &add_ons_ctx,
-                            engine_tree_config.clone(),
-                            reorg_cache,
-                            reorg_shared_caches,
-                        )
+                        .build_tree_validator(&add_ons_ctx, engine_tree_config.clone(), reorg_cache)
                         .await
                 },
                 node_config.debug.reorg_frequency,

crates/node/builder/src/rpc.rs

@@ -1,8 +1,8 @@
 //! Builder support for rpc components.
 
 pub use jsonrpsee::server::middleware::rpc::{RpcService, RpcServiceBuilder};
+use reth_engine_tree::tree::WaitForCaches;
 pub use reth_engine_tree::tree::{BasicEngineValidator, EngineValidator};
-use reth_engine_tree::tree::{EngineSharedCaches, PayloadSparseTrieKind, WaitForCaches};
 pub use reth_rpc_builder::{middleware::RethRpcMiddleware, Identity, Stack};
 pub use reth_trie_db::ChangesetCache;
 
@@ -981,8 +981,7 @@ where
         let Self { eth_api_builder, engine_api_builder, hooks, .. } = self;
 
         let engine_api = engine_api_builder.build_engine_api(&ctx).await?;
-        let AddOnsContext { node, config, beacon_engine_handle, jwt_secret, engine_events, .. } =
-            ctx;
+        let AddOnsContext { node, config, beacon_engine_handle, jwt_secret, engine_events } = ctx;
 
         info!(target: "reth::cli", "Engine API handler initialized");
 
@@ -1295,25 +1294,6 @@ pub trait EngineValidatorBuilder<Node: FullNodeComponents>: Send + Sync + Clone
         tree_config: TreeConfig,
         changeset_cache: ChangesetCache,
     ) -> impl Future<Output = eyre::Result<Self::EngineValidator>> + Send;
-
-    /// Builds the tree validator using the shared cache handles exported by the launcher.
-    ///
-    /// The default implementation preserves the legacy behavior and ignores the provided caches.
-    fn build_tree_validator_with_caches(
-        self,
-        ctx: &AddOnsContext<'_, Node>,
-        tree_config: TreeConfig,
-        changeset_cache: ChangesetCache,
-        shared_caches: EngineSharedCaches<Node::Evm>,
-    ) -> impl Future<Output = eyre::Result<Self::EngineValidator>> + Send
-    where
-        Self: Sized,
-    {
-        async move {
-            let _ = shared_caches;
-            self.build_tree_validator(ctx, tree_config, changeset_cache).await
-        }
-    }
 }
 
 /// Basic implementation of [`EngineValidatorBuilder`].
@@ -1361,20 +1341,6 @@ where
         ctx: &AddOnsContext<'_, Node>,
         tree_config: TreeConfig,
         changeset_cache: ChangesetCache,
-    ) -> eyre::Result<Self::EngineValidator> {
-        let shared_caches = EngineSharedCaches::with_sparse_trie_kind(PayloadSparseTrieKind::from(
-            tree_config.enable_arena_sparse_trie(),
-        ));
-        self.build_tree_validator_with_caches(ctx, tree_config, changeset_cache, shared_caches)
-            .await
-    }
-
-    async fn build_tree_validator_with_caches(
-        self,
-        ctx: &AddOnsContext<'_, Node>,
-        tree_config: TreeConfig,
-        changeset_cache: ChangesetCache,
-        shared_caches: EngineSharedCaches<Node::Evm>,
     ) -> eyre::Result<Self::EngineValidator> {
         let validator = self.payload_validator_builder.build(ctx).await?;
         let data_dir = ctx.config.datadir.clone().resolve_datadir(ctx.config.chain.chain());
@@ -1387,7 +1353,6 @@ where
             validator,
             tree_config,
             invalid_block_hook,
-            shared_caches,
             changeset_cache,
             ctx.node.task_executor().clone(),
         ))

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

@@ -81,7 +81,7 @@ pub mod clients {
         otterscan::OtterscanClient,
         reth::RethApiClient,
         reth_engine::RethEngineApiClient,
-        rpc::RpcApiServer,
+        rpc::RpcApiClient,
         testing::TestingApiClient,
         trace::TraceApiClient,
         txpool::TxPoolApiClient,
@@ -90,6 +90,6 @@ pub mod clients {
     };
     pub use reth_rpc_eth_api::{
         EthApiClient, EthBundleApiClient, EthCallBundleApiClient, EthFilterApiClient,
-        L2EthApiExtServer,
+        L2EthApiExtClient,
     };
 }

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

@@ -394,20 +394,13 @@ where
     where
         EthApi: FullEthApiServer<Provider = Provider, Pool = Pool>,
     {
-        let mut modules = TransportRpcModules::default();
-
-        if !module_config.is_empty() {
-            let TransportRpcModuleConfig { http, ws, ipc, config } = module_config.clone();
-
-            let mut registry = self.into_registry(config.unwrap_or_default(), eth, engine_events);
-
-            modules.config = module_config;
-            modules.http = registry.maybe_module(http.as_ref());
-            modules.ws = registry.maybe_module(ws.as_ref());
-            modules.ipc = registry.maybe_module(ipc.as_ref());
+        if module_config.is_empty() {
+            TransportRpcModules::default()
+        } else {
+            let config = module_config.config.clone().unwrap_or_default();
+            let mut registry = self.into_registry(config, eth, engine_events);
+            registry.create_transport_rpc_modules(module_config)
         }
-
-        modules
     }
 }
 

crates/rpc/rpc-eth-api/src/helpers/call.rs

@@ -127,9 +127,11 @@ pub trait EthCall: EstimateCall + Call + LoadPendingBlock + LoadBlock + FullEthA
                         .into());
                     }
 
+                    let attributes = this.next_env_attributes(&parent)?;
+
                     let mut evm_env = this
                         .evm_config()
-                        .next_evm_env(&parent, &this.next_env_attributes(&parent)?)
+                        .next_evm_env(&parent, &attributes)
                         .map_err(RethError::other)
                         .map_err(Self::Error::from_eth_err)?;
 
@@ -205,7 +207,7 @@ pub trait EthCall: EstimateCall + Call + LoadPendingBlock + LoadBlock + FullEthA
 
                     let ctx = this
                         .evm_config()
-                        .context_for_next_block(&parent, this.next_env_attributes(&parent)?)
+                        .context_for_next_block(&parent, attributes)
                         .map_err(RethError::other)
                         .map_err(Self::Error::from_eth_err)?;
                     let map_err = |e: EthApiError| -> Self::Error {

crates/rpc/rpc-eth-api/src/helpers/signer.rs

@@ -31,4 +31,46 @@ pub trait EthSigner<T, TxReq = TransactionRequest>: Send + Sync + DynClone {
     fn sign_typed_data(&self, address: Address, payload: &TypedData) -> Result<Signature>;
 }
 
-dyn_clone::clone_trait_object!(<T> EthSigner<T>);
+dyn_clone::clone_trait_object!(<T, TxReq> EthSigner<T, TxReq>);
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[derive(Clone)]
+    struct MockSigner;
+
+    struct MockSignedTx;
+    struct MockTxReq;
+
+    #[async_trait::async_trait]
+    impl EthSigner<MockSignedTx, MockTxReq> for MockSigner {
+        fn accounts(&self) -> Vec<Address> {
+            Vec::new()
+        }
+
+        async fn sign(&self, _address: Address, _message: &[u8]) -> Result<Signature> {
+            Err(SignError::NoAccount)
+        }
+
+        async fn sign_transaction(
+            &self,
+            _request: MockTxReq,
+            _address: &Address,
+        ) -> Result<MockSignedTx> {
+            Err(SignError::NoAccount)
+        }
+
+        fn sign_typed_data(&self, _address: Address, _payload: &TypedData) -> Result<Signature> {
+            Err(SignError::NoAccount)
+        }
+    }
+
+    #[test]
+    fn clones_trait_object_with_custom_tx_request_type() {
+        let signer: Box<dyn EthSigner<MockSignedTx, MockTxReq>> = Box::new(MockSigner);
+        let cloned: Box<dyn EthSigner<MockSignedTx, MockTxReq>> = dyn_clone::clone_box(&*signer);
+
+        assert!(cloned.accounts().is_empty());
+    }
+}

crates/storage/db-common/src/init.rs

@@ -10,12 +10,7 @@ use alloy_primitives::{
 use reth_chainspec::EthChainSpec;
 use reth_codecs::Compact;
 use reth_config::config::EtlConfig;
-use reth_db_api::{
-    models::{storage_sharded_key::StorageShardedKey, ShardedKey},
-    tables,
-    transaction::DbTxMut,
-    BlockNumberList, DatabaseError,
-};
+use reth_db_api::{tables, transaction::DbTxMut, DatabaseError};
 use reth_etl::Collector;
 use reth_execution_errors::StateRootError;
 use reth_primitives_traits::{
@@ -23,9 +18,9 @@ use reth_primitives_traits::{
 };
 use reth_provider::{
     errors::provider::ProviderResult, providers::StaticFileWriter, BlockHashReader, BlockNumReader,
-    BundleStateInit, ChainSpecProvider, DBProvider, DatabaseProviderFactory, EitherWriter,
-    ExecutionOutcome, HashingWriter, HeaderProvider, HistoryWriter, MetadataProvider,
-    MetadataWriter, NodePrimitivesProvider, OriginalValuesKnown, ProviderError, RevertsInit,
+    BundleStateInit, ChainSpecProvider, DBProvider, DatabaseProviderFactory, ExecutionOutcome,
+    HashingWriter, HeaderProvider, HistoryWriter, MetadataProvider, MetadataWriter,
+    NodePrimitivesProvider, OriginalValuesKnown, ProviderError, RevertsInit,
     RocksDBProviderFactory, StageCheckpointReader, StageCheckpointWriter, StateWriteConfig,
     StateWriter, StaticFileProviderFactory, StorageSettings, StorageSettingsCache, TrieWriter,
 };
@@ -46,6 +41,11 @@ use serde::{Deserialize, Serialize};
 use std::io::BufRead;
 use tracing::{debug, error, info, trace, warn};
 
+pub use reth_provider::init::{
+    insert_account_history, insert_genesis_account_history, insert_genesis_history,
+    insert_genesis_storage_history, insert_history, insert_storage_history,
+};
+
 /// Default soft limit for number of bytes to read from state dump file, before inserting into
 /// database.
 ///
@@ -415,137 +415,6 @@ where
     Ok(())
 }
 
-/// Inserts history indices for genesis accounts and storage.
-///
-/// Writes to either MDBX or `RocksDB` based on storage settings configuration,
-/// using [`EitherWriter`] to abstract over the storage backend.
-pub fn insert_genesis_history<'a, 'b, Provider>(
-    provider: &Provider,
-    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)> + Clone,
-) -> ProviderResult<()>
-where
-    Provider: DBProvider<Tx: DbTxMut>
-        + HistoryWriter
-        + ChainSpecProvider
-        + StorageSettingsCache
-        + RocksDBProviderFactory
-        + NodePrimitivesProvider,
-{
-    let genesis_block_number = provider.chain_spec().genesis_header().number();
-    insert_history(provider, alloc, genesis_block_number)
-}
-
-/// Inserts account history indices for genesis accounts.
-pub fn insert_genesis_account_history<'a, 'b, Provider>(
-    provider: &Provider,
-    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)>,
-) -> ProviderResult<()>
-where
-    Provider: DBProvider<Tx: DbTxMut>
-        + HistoryWriter
-        + ChainSpecProvider
-        + StorageSettingsCache
-        + RocksDBProviderFactory
-        + NodePrimitivesProvider,
-{
-    let genesis_block_number = provider.chain_spec().genesis_header().number();
-    insert_account_history(provider, alloc, genesis_block_number)
-}
-
-/// Inserts storage history indices for genesis accounts.
-pub fn insert_genesis_storage_history<'a, 'b, Provider>(
-    provider: &Provider,
-    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)>,
-) -> ProviderResult<()>
-where
-    Provider: DBProvider<Tx: DbTxMut>
-        + HistoryWriter
-        + ChainSpecProvider
-        + StorageSettingsCache
-        + RocksDBProviderFactory
-        + NodePrimitivesProvider,
-{
-    let genesis_block_number = provider.chain_spec().genesis_header().number();
-    insert_storage_history(provider, alloc, genesis_block_number)
-}
-
-/// Inserts history indices for genesis accounts and storage.
-///
-/// Writes to either MDBX or `RocksDB` based on storage settings configuration,
-/// using [`EitherWriter`] to abstract over the storage backend.
-pub fn insert_history<'a, 'b, Provider>(
-    provider: &Provider,
-    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)> + Clone,
-    block: u64,
-) -> ProviderResult<()>
-where
-    Provider: DBProvider<Tx: DbTxMut>
-        + HistoryWriter
-        + StorageSettingsCache
-        + RocksDBProviderFactory
-        + NodePrimitivesProvider,
-{
-    insert_account_history(provider, alloc.clone(), block)?;
-    insert_storage_history(provider, alloc, block)?;
-    Ok(())
-}
-
-/// Inserts account history indices at the given block.
-pub fn insert_account_history<'a, 'b, Provider>(
-    provider: &Provider,
-    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)>,
-    block: u64,
-) -> ProviderResult<()>
-where
-    Provider: DBProvider<Tx: DbTxMut>
-        + HistoryWriter
-        + StorageSettingsCache
-        + RocksDBProviderFactory
-        + NodePrimitivesProvider,
-{
-    provider.with_rocksdb_batch(|batch| {
-        let mut writer = EitherWriter::new_accounts_history(provider, batch)?;
-        let list = BlockNumberList::new([block]).expect("single block always fits");
-        for (addr, _) in alloc {
-            writer.upsert_account_history(ShardedKey::last(*addr), &list)?;
-        }
-        trace!(target: "reth::cli", "Inserted account history");
-        Ok(((), writer.into_raw_rocksdb_batch()))
-    })?;
-
-    Ok(())
-}
-
-/// Inserts storage history indices at the given block.
-pub fn insert_storage_history<'a, 'b, Provider>(
-    provider: &Provider,
-    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)>,
-    block: u64,
-) -> ProviderResult<()>
-where
-    Provider: DBProvider<Tx: DbTxMut>
-        + HistoryWriter
-        + StorageSettingsCache
-        + RocksDBProviderFactory
-        + NodePrimitivesProvider,
-{
-    provider.with_rocksdb_batch(|batch| {
-        let mut writer = EitherWriter::new_storages_history(provider, batch)?;
-        let list = BlockNumberList::new([block]).expect("single block always fits");
-        for (addr, account) in alloc {
-            if let Some(storage) = &account.storage {
-                for key in storage.keys() {
-                    writer.upsert_storage_history(StorageShardedKey::last(*addr, *key), &list)?;
-                }
-            }
-        }
-        trace!(target: "reth::cli", "Inserted storage history");
-        Ok(((), writer.into_raw_rocksdb_batch()))
-    })?;
-
-    Ok(())
-}
-
 /// Inserts header for the genesis state.
 pub fn insert_genesis_header<Provider, Spec>(
     provider: &Provider,

crates/storage/provider/Cargo.toml

@@ -36,6 +36,7 @@ reth-fs-util.workspace = true
 
 # ethereum
 alloy-eips.workspace = true
+alloy-genesis.workspace = true
 alloy-primitives.workspace = true
 alloy-rpc-types-engine.workspace = true
 alloy-consensus.workspace = true

crates/storage/provider/src/either_writer.rs

@@ -68,11 +68,11 @@ pub type RocksBatchArg<'a> = crate::providers::rocksdb::RocksDBBatch<'a>;
 /// The raw `RocksDB` batch type returned by [`EitherWriter::into_raw_rocksdb_batch`].
 pub type RawRocksDBBatch = rocksdb::WriteBatchWithTransaction<true>;
 
-/// Helper type for `RocksDB` transaction reference argument in reader constructors.
+/// Helper type for `RocksDB` snapshot argument in reader constructors.
 ///
-/// The `Option` allows callers to skip transaction creation when `RocksDB` isn't needed
+/// The `Option` allows callers to skip `RocksDB` access when it isn't needed
 /// (e.g., on legacy MDBX-only nodes).
-pub type RocksTxRefArg<'a> = Option<&'a crate::providers::rocksdb::RocksTx<'a>>;
+pub type RocksDBRefArg<'a> = Option<crate::providers::rocksdb::RocksReadSnapshot<'a>>;
 
 /// Represents a destination for writing data, either to database, static files, or `RocksDB`.
 #[derive(Debug, Display)]
@@ -672,8 +672,8 @@ pub enum EitherReader<'a, CURSOR, N> {
     Database(CURSOR, PhantomData<&'a ()>),
     /// Read from static file
     StaticFile(StaticFileProvider<N>, PhantomData<&'a ()>),
-    /// Read from `RocksDB` transaction
-    RocksDB(&'a crate::providers::rocksdb::RocksTx<'a>),
+    /// Read from `RocksDB` snapshot (works in both read-only and read-write modes)
+    RocksDB(crate::providers::rocksdb::RocksReadSnapshot<'a>),
 }
 
 impl<'a> EitherReader<'a, (), ()> {
@@ -698,7 +698,7 @@ impl<'a> EitherReader<'a, (), ()> {
     /// Creates a new [`EitherReader`] for storages history based on storage settings.
     pub fn new_storages_history<P>(
         provider: &P,
-        _rocksdb_tx: RocksTxRefArg<'a>,
+        rocksdb: RocksDBRefArg<'a>,
     ) -> ProviderResult<EitherReaderTy<'a, P, tables::StoragesHistory>>
     where
         P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
@@ -706,7 +706,7 @@ impl<'a> EitherReader<'a, (), ()> {
     {
         if provider.cached_storage_settings().storage_v2 {
             return Ok(EitherReader::RocksDB(
-                _rocksdb_tx.expect("storages_history_in_rocksdb requires rocksdb tx"),
+                rocksdb.expect("storages_history_in_rocksdb requires rocksdb snapshot"),
             ));
         }
 
@@ -719,7 +719,7 @@ impl<'a> EitherReader<'a, (), ()> {
     /// Creates a new [`EitherReader`] for transaction hash numbers based on storage settings.
     pub fn new_transaction_hash_numbers<P>(
         provider: &P,
-        _rocksdb_tx: RocksTxRefArg<'a>,
+        rocksdb: RocksDBRefArg<'a>,
     ) -> ProviderResult<EitherReaderTy<'a, P, tables::TransactionHashNumbers>>
     where
         P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
@@ -727,7 +727,7 @@ impl<'a> EitherReader<'a, (), ()> {
     {
         if provider.cached_storage_settings().storage_v2 {
             return Ok(EitherReader::RocksDB(
-                _rocksdb_tx.expect("transaction_hash_numbers_in_rocksdb requires rocksdb tx"),
+                rocksdb.expect("transaction_hash_numbers_in_rocksdb requires rocksdb snapshot"),
             ));
         }
 
@@ -740,7 +740,7 @@ impl<'a> EitherReader<'a, (), ()> {
     /// Creates a new [`EitherReader`] for account history based on storage settings.
     pub fn new_accounts_history<P>(
         provider: &P,
-        _rocksdb_tx: RocksTxRefArg<'a>,
+        rocksdb: RocksDBRefArg<'a>,
     ) -> ProviderResult<EitherReaderTy<'a, P, tables::AccountsHistory>>
     where
         P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
@@ -748,7 +748,7 @@ impl<'a> EitherReader<'a, (), ()> {
     {
         if provider.cached_storage_settings().storage_v2 {
             return Ok(EitherReader::RocksDB(
-                _rocksdb_tx.expect("account_history_in_rocksdb requires rocksdb tx"),
+                rocksdb.expect("account_history_in_rocksdb requires rocksdb snapshot"),
             ));
         }
 
@@ -820,7 +820,7 @@ where
         match self {
             Self::Database(cursor, _) => Ok(cursor.seek_exact(hash)?.map(|(_, v)| v)),
             Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
-            Self::RocksDB(tx) => tx.get::<tables::TransactionHashNumbers>(hash),
+            Self::RocksDB(snapshot) => snapshot.get::<tables::TransactionHashNumbers>(hash),
         }
     }
 }
@@ -837,7 +837,7 @@ where
         match self {
             Self::Database(cursor, _) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
             Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
-            Self::RocksDB(tx) => tx.get::<tables::StoragesHistory>(key),
+            Self::RocksDB(snapshot) => snapshot.get::<tables::StoragesHistory>(key),
         }
     }
 
@@ -861,7 +861,7 @@ where
                 )
             }
             Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
-            Self::RocksDB(tx) => tx.storage_history_info(
+            Self::RocksDB(snapshot) => snapshot.storage_history_info(
                 address,
                 storage_key,
                 block_number,
@@ -883,7 +883,7 @@ where
         match self {
             Self::Database(cursor, _) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
             Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
-            Self::RocksDB(tx) => tx.get::<tables::AccountsHistory>(key),
+            Self::RocksDB(snapshot) => snapshot.get::<tables::AccountsHistory>(key),
         }
     }
 
@@ -906,8 +906,8 @@ where
                 )
             }
             Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
-            Self::RocksDB(tx) => {
-                tx.account_history_info(address, block_number, lowest_available_block_number)
+            Self::RocksDB(snapshot) => {
+                snapshot.account_history_info(address, block_number, lowest_available_block_number)
             }
         }
     }
@@ -1428,7 +1428,7 @@ mod rocksdb_tests {
 
         // Run queries against both backends using EitherReader
         let mdbx_ro = factory.database_provider_ro().unwrap();
-        let rocks_tx = rocks_provider.tx();
+        let rocks_snapshot = rocks_provider.snapshot();
 
         for (i, query) in queries.iter().enumerate() {
             // MDBX query via EitherReader
@@ -1441,10 +1441,8 @@ mod rocksdb_tests {
                 .account_history_info(address, query.block_number, query.lowest_available)
                 .unwrap();
 
-            // RocksDB query via EitherReader
-            let mut rocks_reader: EitherReader<'_, AccountsHistoryReadCursor, EthPrimitives> =
-                EitherReader::RocksDB(&rocks_tx);
-            let rocks_result = rocks_reader
+            // RocksDB query via EitherReader — reuse snapshot for consistent view
+            let rocks_result = rocks_snapshot
                 .account_history_info(address, query.block_number, query.lowest_available)
                 .unwrap();
 
@@ -1477,7 +1475,6 @@ mod rocksdb_tests {
             );
         }
 
-        rocks_tx.rollback().unwrap();
         drop(temp_dir);
     }
 
@@ -1520,7 +1517,7 @@ mod rocksdb_tests {
 
         // Run queries against both backends using EitherReader
         let mdbx_ro = factory.database_provider_ro().unwrap();
-        let rocks_tx = rocks_provider.tx();
+        let rocks_snapshot = rocks_provider.snapshot();
 
         for (i, query) in queries.iter().enumerate() {
             // MDBX query via EitherReader
@@ -1538,10 +1535,8 @@ mod rocksdb_tests {
                 )
                 .unwrap();
 
-            // RocksDB query via EitherReader
-            let mut rocks_reader: EitherReader<'_, StoragesHistoryReadCursor, EthPrimitives> =
-                EitherReader::RocksDB(&rocks_tx);
-            let rocks_result = rocks_reader
+            // RocksDB query via snapshot — reuse for consistent view
+            let rocks_result = rocks_snapshot
                 .storage_history_info(
                     address,
                     storage_key,
@@ -1579,7 +1574,6 @@ mod rocksdb_tests {
             );
         }
 
-        rocks_tx.rollback().unwrap();
         drop(temp_dir);
     }
 
@@ -1809,10 +1803,10 @@ mod rocksdb_tests {
     }
 
     /// Test that `EitherReader::new_accounts_history` panics when settings require
-    /// `RocksDB` but no tx is provided (`None`). This is an invariant violation that
-    /// indicates a bug - `with_rocksdb_tx` should always provide a tx when needed.
+    /// `RocksDB` but no snapshot is given (`None`). This is an invariant violation that
+    /// indicates a bug - `with_rocksdb_snapshot` should always provide a snapshot when needed.
     #[test]
-    #[should_panic(expected = "account_history_in_rocksdb requires rocksdb tx")]
+    #[should_panic(expected = "account_history_in_rocksdb requires rocksdb snapshot")]
     fn test_settings_mismatch_panics() {
         let factory = create_test_provider_factory();
 

crates/storage/provider/src/init.rs

@@ -0,0 +1,145 @@
+use crate::{
+    ChainSpecProvider, DBProvider, EitherWriter, HistoryWriter, NodePrimitivesProvider,
+    ProviderResult, RocksDBProviderFactory, StorageSettingsCache,
+};
+use alloy_consensus::BlockHeader;
+use alloy_genesis::GenesisAccount;
+use alloy_primitives::Address;
+use reth_chainspec::EthChainSpec;
+use reth_db::{
+    models::{storage_sharded_key::StorageShardedKey, ShardedKey},
+    transaction::DbTxMut,
+    BlockNumberList,
+};
+use tracing::trace;
+
+/// Inserts history indices for genesis accounts and storage.
+///
+/// Writes to either MDBX or `RocksDB` based on storage settings configuration,
+/// using [`EitherWriter`] to abstract over the storage backend.
+pub fn insert_genesis_history<'a, 'b, Provider>(
+    provider: &Provider,
+    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)> + Clone,
+) -> ProviderResult<()>
+where
+    Provider: DBProvider<Tx: DbTxMut>
+        + HistoryWriter
+        + ChainSpecProvider
+        + StorageSettingsCache
+        + RocksDBProviderFactory
+        + NodePrimitivesProvider,
+{
+    let genesis_block_number = provider.chain_spec().genesis_header().number();
+    insert_history(provider, alloc, genesis_block_number)
+}
+
+/// Inserts account history indices for genesis accounts.
+pub fn insert_genesis_account_history<'a, 'b, Provider>(
+    provider: &Provider,
+    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)>,
+) -> ProviderResult<()>
+where
+    Provider: DBProvider<Tx: DbTxMut>
+        + HistoryWriter
+        + ChainSpecProvider
+        + StorageSettingsCache
+        + RocksDBProviderFactory
+        + NodePrimitivesProvider,
+{
+    let genesis_block_number = provider.chain_spec().genesis_header().number();
+    insert_account_history(provider, alloc, genesis_block_number)
+}
+
+/// Inserts storage history indices for genesis accounts.
+pub fn insert_genesis_storage_history<'a, 'b, Provider>(
+    provider: &Provider,
+    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)>,
+) -> ProviderResult<()>
+where
+    Provider: DBProvider<Tx: DbTxMut>
+        + HistoryWriter
+        + ChainSpecProvider
+        + StorageSettingsCache
+        + RocksDBProviderFactory
+        + NodePrimitivesProvider,
+{
+    let genesis_block_number = provider.chain_spec().genesis_header().number();
+    insert_storage_history(provider, alloc, genesis_block_number)
+}
+
+/// Inserts history indices for genesis accounts and storage.
+///
+/// Writes to either MDBX or `RocksDB` based on storage settings configuration,
+/// using [`EitherWriter`] to abstract over the storage backend.
+pub fn insert_history<'a, 'b, Provider>(
+    provider: &Provider,
+    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)> + Clone,
+    block: u64,
+) -> ProviderResult<()>
+where
+    Provider: DBProvider<Tx: DbTxMut>
+        + HistoryWriter
+        + StorageSettingsCache
+        + RocksDBProviderFactory
+        + NodePrimitivesProvider,
+{
+    insert_account_history(provider, alloc.clone(), block)?;
+    insert_storage_history(provider, alloc, block)?;
+    Ok(())
+}
+
+/// Inserts account history indices at the given block.
+pub fn insert_account_history<'a, 'b, Provider>(
+    provider: &Provider,
+    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)>,
+    block: u64,
+) -> ProviderResult<()>
+where
+    Provider: DBProvider<Tx: DbTxMut>
+        + HistoryWriter
+        + StorageSettingsCache
+        + RocksDBProviderFactory
+        + NodePrimitivesProvider,
+{
+    provider.with_rocksdb_batch(|batch| {
+        let mut writer = EitherWriter::new_accounts_history(provider, batch)?;
+        let list = BlockNumberList::new([block]).expect("single block always fits");
+        for (addr, _) in alloc {
+            writer.upsert_account_history(ShardedKey::last(*addr), &list)?;
+        }
+        trace!(target: "reth::provider", "Inserted account history");
+        Ok(((), writer.into_raw_rocksdb_batch()))
+    })?;
+
+    Ok(())
+}
+
+/// Inserts storage history indices at the given block.
+pub fn insert_storage_history<'a, 'b, Provider>(
+    provider: &Provider,
+    alloc: impl Iterator<Item = (&'a Address, &'b GenesisAccount)>,
+    block: u64,
+) -> ProviderResult<()>
+where
+    Provider: DBProvider<Tx: DbTxMut>
+        + HistoryWriter
+        + StorageSettingsCache
+        + RocksDBProviderFactory
+        + NodePrimitivesProvider,
+{
+    provider.with_rocksdb_batch(|batch| {
+        let mut writer = EitherWriter::new_storages_history(provider, batch)?;
+        let list = BlockNumberList::new([block]).expect("single block always fits");
+        for (addr, account) in alloc {
+            if let Some(storage) = &account.storage {
+                for key in storage.keys() {
+                    writer.upsert_storage_history(StorageShardedKey::last(*addr, *key), &list)?;
+                }
+            }
+        }
+        trace!(target: "reth::cli", "Inserted storage history");
+        Ok(((), writer.into_raw_rocksdb_batch()))
+    })?;
+
+    Ok(())
+}

crates/storage/provider/src/lib.rs

@@ -12,6 +12,9 @@
 #![cfg_attr(not(test), warn(unused_crate_dependencies))]
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
+/// Utility functions for initializing the database.
+pub mod init;
+
 /// Various provider traits.
 mod traits;
 pub use traits::*;

crates/storage/provider/src/providers/blockchain_provider.rs

@@ -555,7 +555,6 @@ impl<N: ProviderNodeTypes> StateProviderFactory for BlockchainProvider<N> {
     ) -> ProviderResult<StateProviderBox> {
         trace!(target: "providers::blockchain", ?block_number, "Getting history by block number");
         let provider = self.consistent_provider()?;
-        provider.ensure_canonical_block(block_number)?;
         let hash = provider
             .block_hash(block_number)?
             .ok_or_else(|| ProviderError::HeaderNotFound(block_number.into()))?;

crates/storage/provider/src/providers/consistent.rs

@@ -600,21 +600,23 @@ impl<N: ProviderNodeTypes> ConsistentProvider<N> {
         self,
         block_hash: BlockHash,
     ) -> ProviderResult<StateProviderBox> {
+        // Resolve block number and verify it's canonical before destructuring self
+        let block_number =
+            self.block_number(block_hash)?.ok_or(ProviderError::BlockHashNotFound(block_hash))?;
+        self.ensure_canonical_block(block_number)?;
+
         let Self { storage_provider, head_block, .. } = self;
-        let into_history_at_block_hash = |block_hash| -> ProviderResult<StateProviderBox> {
-            let block_number = storage_provider
-                .block_number(block_hash)?
-                .ok_or(ProviderError::BlockHashNotFound(block_hash))?;
-            storage_provider.try_into_history_at_block(block_number)
-        };
         if let Some(Some(block_state)) =
             head_block.as_ref().map(|b| b.block_on_chain(block_hash.into()))
         {
             let anchor_hash = block_state.anchor().hash;
-            let latest_historical = into_history_at_block_hash(anchor_hash)?;
+            let block_number = storage_provider
+                .block_number(anchor_hash)?
+                .ok_or(ProviderError::BlockHashNotFound(anchor_hash))?;
+            let latest_historical = storage_provider.try_into_history_at_block(block_number)?;
             return Ok(Box::new(block_state.state_provider(latest_historical)));
         }
-        into_history_at_block_hash(block_hash)
+        storage_provider.try_into_history_at_block(block_number)
     }
 }
 

crates/storage/provider/src/providers/database/provider.rs

@@ -1925,8 +1925,8 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> TransactionsProvider for Datab
     type Transaction = TxTy<N>;
 
     fn transaction_id(&self, tx_hash: TxHash) -> ProviderResult<Option<TxNumber>> {
-        self.with_rocksdb_tx(|tx_ref| {
-            let mut reader = EitherReader::new_transaction_hash_numbers(self, tx_ref)?;
+        self.with_rocksdb_snapshot(|rocksdb_ref| {
+            let mut reader = EitherReader::new_transaction_hash_numbers(self, rocksdb_ref)?;
             reader.get_transaction_hash_number(tx_hash)
         })
     }

crates/storage/provider/src/providers/mod.rs

@@ -36,7 +36,7 @@ pub(crate) mod rocksdb;
 
 pub use rocksdb::{
     PruneShardOutcome, PrunedIndices, RocksDBBatch, RocksDBBuilder, RocksDBIter, RocksDBProvider,
-    RocksDBRawIter, RocksDBStats, RocksDBTableStats, RocksTx,
+    RocksDBRawIter, RocksDBStats, RocksDBTableStats, RocksReadSnapshot, RocksTx,
 };
 
 /// Helper trait to bound [`NodeTypes`] so that combined with database they satisfy

crates/storage/provider/src/providers/rocksdb/invariants.rs

@@ -9,7 +9,7 @@ use crate::StaticFileProviderFactory;
 use alloy_eips::eip2718::Encodable2718;
 use alloy_primitives::BlockNumber;
 use rayon::prelude::*;
-use reth_db_api::tables;
+use reth_db_api::{table::Table, tables};
 use reth_stages_types::StageId;
 use reth_static_file_types::StaticFileSegment;
 use reth_storage_api::{
@@ -334,6 +334,7 @@ impl RocksDBProvider {
 
                 let batch = self.unwind_storage_history_indices(&indices)?;
                 self.commit_batch(batch)?;
+                self.flush(&[tables::StoragesHistory::NAME])?;
             }
 
             batch_start = batch_end + 1;

crates/storage/provider/src/providers/rocksdb/mod.rs

@@ -7,5 +7,5 @@ mod provider;
 pub(crate) use provider::{PendingRocksDBBatches, RocksDBWriteCtx};
 pub use provider::{
     PruneShardOutcome, PrunedIndices, RocksDBBatch, RocksDBBuilder, RocksDBIter, RocksDBProvider,
-    RocksDBRawIter, RocksDBStats, RocksDBTableStats, RocksTx,
+    RocksDBRawIter, RocksDBStats, RocksDBTableStats, RocksReadSnapshot, RocksTx,
 };

crates/storage/provider/src/providers/rocksdb/provider.rs

@@ -27,8 +27,8 @@ use reth_storage_errors::{
 use rocksdb::{
     BlockBasedOptions, Cache, ColumnFamilyDescriptor, CompactionPri, DBCompressionType,
     DBRawIteratorWithThreadMode, IteratorMode, OptimisticTransactionDB,
-    OptimisticTransactionOptions, Options, Transaction, WriteBatchWithTransaction, WriteOptions,
-    DB,
+    OptimisticTransactionOptions, Options, SnapshotWithThreadMode, Transaction,
+    WriteBatchWithTransaction, WriteOptions, DB,
 };
 use std::{
     collections::BTreeMap,
@@ -430,16 +430,6 @@ impl RocksDBProviderInner {
         cf.ok_or_else(|| DatabaseError::Other(format!("Column family '{}' not found", T::NAME)))
     }
 
-    /// Gets the column family handle for a table from the read-write database.
-    ///
-    /// # Panics
-    /// Panics if in read-only mode.
-    fn cf_handle_rw(&self, name: &str) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
-        self.db_rw()
-            .cf_handle(name)
-            .ok_or_else(|| DatabaseError::Other(format!("Column family '{}' not found", name)))
-    }
-
     /// Gets a value from a column family.
     fn get_cf(
         &self,
@@ -504,6 +494,14 @@ impl RocksDBProviderInner {
         }
     }
 
+    /// Returns a read-only, point-in-time snapshot of the database.
+    fn snapshot(&self) -> RocksReadSnapshotInner<'_> {
+        match self {
+            Self::ReadWrite { db, .. } => RocksReadSnapshotInner::ReadWrite(db.snapshot()),
+            Self::ReadOnly { db, .. } => RocksReadSnapshotInner::ReadOnly(db.snapshot()),
+        }
+    }
+
     /// Returns the path to the database directory.
     fn path(&self) -> &Path {
         match self {
@@ -703,6 +701,13 @@ impl RocksDBProvider {
         matches!(self.0.as_ref(), RocksDBProviderInner::ReadOnly { .. })
     }
 
+    /// Returns a read-only, point-in-time snapshot of the database.
+    ///
+    /// Lighter weight than [`RocksTx`] — no write-conflict tracking, and `Send + Sync`.
+    pub fn snapshot(&self) -> RocksReadSnapshot<'_> {
+        RocksReadSnapshot { inner: self.0.snapshot(), provider: self }
+    }
+
     /// Creates a new transaction with MDBX-like semantics (read-your-writes, rollback).
     ///
     /// Note: With `OptimisticTransactionDB`, commits may fail if there are conflicts.
@@ -1366,6 +1371,185 @@ impl RocksDBProvider {
     }
 }
 
+/// A point-in-time read snapshot of the `RocksDB` database.
+///
+/// All reads through this snapshot see a consistent view of the database at the point
+/// the snapshot was created, regardless of concurrent writes. This is the primary reader
+/// used by [`EitherReader::RocksDB`](crate::either_writer::EitherReader) for history lookups.
+///
+/// Lighter weight than [`RocksTx`] — no transaction overhead, no write support.
+pub struct RocksReadSnapshot<'db> {
+    inner: RocksReadSnapshotInner<'db>,
+    provider: &'db RocksDBProvider,
+}
+
+/// Inner enum to hold the snapshot for either read-write or read-only mode.
+enum RocksReadSnapshotInner<'db> {
+    /// Snapshot from read-write `OptimisticTransactionDB`.
+    ReadWrite(SnapshotWithThreadMode<'db, OptimisticTransactionDB>),
+    /// Snapshot from read-only `DB`.
+    ReadOnly(SnapshotWithThreadMode<'db, DB>),
+}
+
+impl<'db> RocksReadSnapshotInner<'db> {
+    /// Returns a raw iterator over a column family.
+    fn raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'_> {
+        match self {
+            Self::ReadWrite(snap) => RocksDBRawIterEnum::ReadWrite(snap.raw_iterator_cf(cf)),
+            Self::ReadOnly(snap) => RocksDBRawIterEnum::ReadOnly(snap.raw_iterator_cf(cf)),
+        }
+    }
+}
+
+impl fmt::Debug for RocksReadSnapshot<'_> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("RocksReadSnapshot")
+            .field("provider", &self.provider)
+            .finish_non_exhaustive()
+    }
+}
+
+impl<'db> RocksReadSnapshot<'db> {
+    /// Gets the column family handle for a table.
+    fn cf_handle<T: Table>(&self) -> Result<&'db rocksdb::ColumnFamily, DatabaseError> {
+        self.provider.get_cf_handle::<T>()
+    }
+
+    /// Gets a value from the specified table.
+    pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
+        let encoded_key = key.encode();
+        let cf = self.cf_handle::<T>()?;
+        let result = match &self.inner {
+            RocksReadSnapshotInner::ReadWrite(snap) => snap.get_cf(cf, encoded_key.as_ref()),
+            RocksReadSnapshotInner::ReadOnly(snap) => snap.get_cf(cf, encoded_key.as_ref()),
+        }
+        .map_err(|e| {
+            ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
+                message: e.to_string().into(),
+                code: -1,
+            }))
+        })?;
+
+        Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
+    }
+
+    /// Lookup account history and return [`HistoryInfo`] directly.
+    pub fn account_history_info(
+        &self,
+        address: Address,
+        block_number: BlockNumber,
+        lowest_available_block_number: Option<BlockNumber>,
+    ) -> ProviderResult<HistoryInfo> {
+        let key = ShardedKey::new(address, block_number);
+        self.history_info::<tables::AccountsHistory>(
+            key.encode().as_ref(),
+            block_number,
+            lowest_available_block_number,
+            |key_bytes| Ok(<ShardedKey<Address> as Decode>::decode(key_bytes)?.key == address),
+            |prev_bytes| {
+                <ShardedKey<Address> as Decode>::decode(prev_bytes)
+                    .map(|k| k.key == address)
+                    .unwrap_or(false)
+            },
+        )
+    }
+
+    /// Lookup storage history and return [`HistoryInfo`] directly.
+    pub fn storage_history_info(
+        &self,
+        address: Address,
+        storage_key: B256,
+        block_number: BlockNumber,
+        lowest_available_block_number: Option<BlockNumber>,
+    ) -> ProviderResult<HistoryInfo> {
+        let key = StorageShardedKey::new(address, storage_key, block_number);
+        self.history_info::<tables::StoragesHistory>(
+            key.encode().as_ref(),
+            block_number,
+            lowest_available_block_number,
+            |key_bytes| {
+                let k = <StorageShardedKey as Decode>::decode(key_bytes)?;
+                Ok(k.address == address && k.sharded_key.key == storage_key)
+            },
+            |prev_bytes| {
+                <StorageShardedKey as Decode>::decode(prev_bytes)
+                    .map(|k| k.address == address && k.sharded_key.key == storage_key)
+                    .unwrap_or(false)
+            },
+        )
+    }
+
+    /// Generic history lookup using the snapshot's raw iterator.
+    fn history_info<T>(
+        &self,
+        encoded_key: &[u8],
+        block_number: BlockNumber,
+        lowest_available_block_number: Option<BlockNumber>,
+        key_matches: impl FnOnce(&[u8]) -> Result<bool, reth_db_api::DatabaseError>,
+        prev_key_matches: impl Fn(&[u8]) -> bool,
+    ) -> ProviderResult<HistoryInfo>
+    where
+        T: Table<Value = BlockNumberList>,
+    {
+        let is_maybe_pruned = lowest_available_block_number.is_some();
+        let fallback = || {
+            Ok(if is_maybe_pruned {
+                HistoryInfo::MaybeInPlainState
+            } else {
+                HistoryInfo::NotYetWritten
+            })
+        };
+
+        let cf = self.cf_handle::<T>()?;
+        let mut iter = self.inner.raw_iterator_cf(cf);
+
+        iter.seek(encoded_key);
+        iter.status().map_err(|e| {
+            ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
+                message: e.to_string().into(),
+                code: -1,
+            }))
+        })?;
+
+        if !iter.valid() {
+            return fallback();
+        }
+
+        let Some(key_bytes) = iter.key() else {
+            return fallback();
+        };
+        if !key_matches(key_bytes)? {
+            return fallback();
+        }
+
+        let Some(value_bytes) = iter.value() else {
+            return fallback();
+        };
+        let chunk = BlockNumberList::decompress(value_bytes)?;
+        let (rank, found_block) = compute_history_rank(&chunk, block_number);
+
+        let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) {
+            iter.prev();
+            iter.status().map_err(|e| {
+                ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
+                    message: e.to_string().into(),
+                    code: -1,
+                }))
+            })?;
+            let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches);
+            !has_prev
+        } else {
+            false
+        };
+
+        Ok(HistoryInfo::from_lookup(
+            found_block,
+            is_before_first_write,
+            lowest_available_block_number,
+        ))
+    }
+}
+
 /// Outcome of pruning a history shard in `RocksDB`.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum PruneShardOutcome {
@@ -2270,151 +2454,6 @@ impl<'db> RocksTx<'db> {
             ProviderError::Database(DatabaseError::Other(format!("rollback failed: {e}")))
         })
     }
-
-    /// Lookup account history and return [`HistoryInfo`] directly.
-    ///
-    /// This is a thin wrapper around `history_info` that:
-    /// - Builds the `ShardedKey` for the address + target block.
-    /// - Validates that the found shard belongs to the same address.
-    pub fn account_history_info(
-        &self,
-        address: Address,
-        block_number: BlockNumber,
-        lowest_available_block_number: Option<BlockNumber>,
-    ) -> ProviderResult<HistoryInfo> {
-        let key = ShardedKey::new(address, block_number);
-        self.history_info::<tables::AccountsHistory>(
-            key.encode().as_ref(),
-            block_number,
-            lowest_available_block_number,
-            |key_bytes| Ok(<ShardedKey<Address> as Decode>::decode(key_bytes)?.key == address),
-            |prev_bytes| {
-                <ShardedKey<Address> as Decode>::decode(prev_bytes)
-                    .map(|k| k.key == address)
-                    .unwrap_or(false)
-            },
-        )
-    }
-
-    /// Lookup storage history and return [`HistoryInfo`] directly.
-    ///
-    /// This is a thin wrapper around `history_info` that:
-    /// - Builds the `StorageShardedKey` for address + storage key + target block.
-    /// - Validates that the found shard belongs to the same address and storage slot.
-    pub fn storage_history_info(
-        &self,
-        address: Address,
-        storage_key: B256,
-        block_number: BlockNumber,
-        lowest_available_block_number: Option<BlockNumber>,
-    ) -> ProviderResult<HistoryInfo> {
-        let key = StorageShardedKey::new(address, storage_key, block_number);
-        self.history_info::<tables::StoragesHistory>(
-            key.encode().as_ref(),
-            block_number,
-            lowest_available_block_number,
-            |key_bytes| {
-                let k = <StorageShardedKey as Decode>::decode(key_bytes)?;
-                Ok(k.address == address && k.sharded_key.key == storage_key)
-            },
-            |prev_bytes| {
-                <StorageShardedKey as Decode>::decode(prev_bytes)
-                    .map(|k| k.address == address && k.sharded_key.key == storage_key)
-                    .unwrap_or(false)
-            },
-        )
-    }
-
-    /// Generic history lookup for sharded history tables.
-    ///
-    /// Seeks to the shard containing `block_number`, checks if the key matches via `key_matches`,
-    /// and uses `prev_key_matches` to detect if a previous shard exists for the same key.
-    fn history_info<T>(
-        &self,
-        encoded_key: &[u8],
-        block_number: BlockNumber,
-        lowest_available_block_number: Option<BlockNumber>,
-        key_matches: impl FnOnce(&[u8]) -> Result<bool, reth_db_api::DatabaseError>,
-        prev_key_matches: impl Fn(&[u8]) -> bool,
-    ) -> ProviderResult<HistoryInfo>
-    where
-        T: Table<Value = BlockNumberList>,
-    {
-        // History may be pruned if a lowest available block is set.
-        let is_maybe_pruned = lowest_available_block_number.is_some();
-        let fallback = || {
-            Ok(if is_maybe_pruned {
-                HistoryInfo::MaybeInPlainState
-            } else {
-                HistoryInfo::NotYetWritten
-            })
-        };
-
-        let cf = self.provider.0.cf_handle_rw(T::NAME)?;
-
-        // Create a raw iterator to access key bytes directly.
-        let mut iter: DBRawIteratorWithThreadMode<'_, Transaction<'_, OptimisticTransactionDB>> =
-            self.inner.raw_iterator_cf(&cf);
-
-        // Seek to the smallest key >= encoded_key.
-        iter.seek(encoded_key);
-        Self::raw_iter_status_ok(&iter)?;
-
-        if !iter.valid() {
-            // No shard found at or after target block.
-            //
-            // (MaybeInPlainState) The key may have been written, but due to pruning we may not have
-            // changesets and history, so we need to make a plain state lookup.
-            // (HistoryInfo::NotYetWritten) The key has not been written to at all.
-            return fallback();
-        }
-
-        // Check if the found key matches our target entity.
-        let Some(key_bytes) = iter.key() else {
-            return fallback();
-        };
-        if !key_matches(key_bytes)? {
-            // The found key is for a different entity.
-            return fallback();
-        }
-
-        // Decompress the block list for this shard.
-        let Some(value_bytes) = iter.value() else {
-            return fallback();
-        };
-        let chunk = BlockNumberList::decompress(value_bytes)?;
-        let (rank, found_block) = compute_history_rank(&chunk, block_number);
-
-        // Lazy check for previous shard - only called when needed.
-        // If we can step to a previous shard for this same key, history already exists,
-        // so the target block is not before the first write.
-        let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) {
-            iter.prev();
-            Self::raw_iter_status_ok(&iter)?;
-            let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches);
-            !has_prev
-        } else {
-            false
-        };
-
-        Ok(HistoryInfo::from_lookup(
-            found_block,
-            is_before_first_write,
-            lowest_available_block_number,
-        ))
-    }
-
-    /// Returns an error if the raw iterator is in an invalid state due to an I/O error.
-    fn raw_iter_status_ok(
-        iter: &DBRawIteratorWithThreadMode<'_, Transaction<'_, OptimisticTransactionDB>>,
-    ) -> ProviderResult<()> {
-        iter.status().map_err(|e| {
-            ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
-                message: e.to_string().into(),
-                code: -1,
-            }))
-        })
-    }
 }
 
 /// Wrapper enum for `RocksDB` iterators that works in both read-write and read-only modes.
@@ -2488,6 +2527,14 @@ impl RocksDBRawIterEnum<'_> {
         }
     }
 
+    /// Moves the iterator to the previous key.
+    fn prev(&mut self) {
+        match self {
+            Self::ReadWrite(iter) => iter.prev(),
+            Self::ReadOnly(iter) => iter.prev(),
+        }
+    }
+
     /// Returns the status of the iterator.
     fn status(&self) -> Result<(), rocksdb::Error> {
         match self {
@@ -2948,16 +2995,45 @@ mod tests {
         let shard_key = ShardedKey::new(address, u64::MAX);
         provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
 
-        let tx = provider.tx();
-
         // Query for block 50 with lowest_available_block_number = 100
         // This simulates a pruned state where data before block 100 is not available.
         // Since we're before the first write AND pruning boundary is set, we need to
         // check the changeset at the first write block.
-        let result = tx.account_history_info(address, 50, Some(100)).unwrap();
+        let result = provider.snapshot().account_history_info(address, 50, Some(100)).unwrap();
         assert_eq!(result, HistoryInfo::InChangeset(100));
+    }
 
-        tx.rollback().unwrap();
+    /// Verifies that history lookups work on a read-only `RocksDB` provider.
+    /// This was the original bug — read-only mode panicked on `account_history_info`.
+    #[test]
+    fn test_account_history_info_read_only() {
+        let temp_dir = TempDir::new().unwrap();
+        let address = Address::from([0x42; 20]);
+        let chunk = IntegerList::new([100, 200, 300]).unwrap();
+        let shard_key = ShardedKey::new(address, u64::MAX);
+
+        // Write data with a read-write provider, then drop it.
+        {
+            let provider =
+                RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
+            provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
+        }
+
+        // Reopen in read-only mode and verify the lookup succeeds (no panic).
+        let ro_provider = RocksDBBuilder::new(temp_dir.path())
+            .with_default_tables()
+            .with_read_only(true)
+            .build()
+            .unwrap();
+
+        let result = ro_provider.snapshot().account_history_info(address, 200, None).unwrap();
+        assert_eq!(result, HistoryInfo::InChangeset(200));
+
+        let result = ro_provider.snapshot().account_history_info(address, 50, None).unwrap();
+        assert_eq!(result, HistoryInfo::NotYetWritten);
+
+        let result = ro_provider.snapshot().account_history_info(address, 400, None).unwrap();
+        assert_eq!(result, HistoryInfo::InPlainState);
     }
 
     #[test]

crates/storage/provider/src/providers/state/historical.rs

@@ -156,8 +156,8 @@ impl<'b, Provider: DBProvider + ChangeSetReader + StorageChangeSetReader + Block
             return Err(ProviderError::StateAtBlockPruned(self.block_number))
         }
 
-        self.provider.with_rocksdb_tx(|rocks_tx_ref| {
-            let mut reader = EitherReader::new_accounts_history(self.provider, rocks_tx_ref)?;
+        self.provider.with_rocksdb_snapshot(|rocksdb_ref| {
+            let mut reader = EitherReader::new_accounts_history(self.provider, rocksdb_ref)?;
             reader.account_history_info(
                 address,
                 self.block_number,
@@ -181,8 +181,8 @@ impl<'b, Provider: DBProvider + ChangeSetReader + StorageChangeSetReader + Block
             return Err(ProviderError::StateAtBlockPruned(self.block_number))
         }
 
-        self.provider.with_rocksdb_tx(|rocks_tx_ref| {
-            let mut reader = EitherReader::new_storages_history(self.provider, rocks_tx_ref)?;
+        self.provider.with_rocksdb_snapshot(|rocksdb_ref| {
+            let mut reader = EitherReader::new_storages_history(self.provider, rocksdb_ref)?;
             reader.storage_history_info(
                 address,
                 lookup_key,

crates/storage/provider/src/traits/rocksdb_provider.rs

@@ -1,5 +1,5 @@
 use crate::{
-    either_writer::{RawRocksDBBatch, RocksBatchArg, RocksTxRefArg},
+    either_writer::{RawRocksDBBatch, RocksBatchArg, RocksDBRefArg},
     providers::RocksDBProvider,
 };
 use reth_storage_api::StorageSettingsCache;
@@ -25,20 +25,24 @@ pub trait RocksDBProviderFactory {
     /// full commit path.
     fn commit_pending_rocksdb_batches(&self) -> ProviderResult<()>;
 
-    /// Executes a closure with a `RocksDB` transaction for reading.
+    /// Executes a closure with a `RocksDB` point-in-time snapshot for consistent reads.
     ///
-    /// This helper encapsulates all the cfg-gated `RocksDB` transaction handling for reads.
-    /// On legacy MDBX-only nodes (where `any_in_rocksdb()` is false), this skips creating
-    /// the `RocksDB` transaction entirely, avoiding unnecessary overhead.
-    fn with_rocksdb_tx<F, R>(&self, f: F) -> ProviderResult<R>
+    /// This helper encapsulates `RocksDB` access for read operations.
+    /// On legacy MDBX-only nodes (where `storage_v2` is false), this skips creating
+    /// the `RocksDB` snapshot entirely, avoiding unnecessary overhead.
+    ///
+    /// Unlike a transaction-based approach, this works in both read-only and read-write
+    /// modes since the snapshot provides a consistent view of the data at the time it
+    /// was created.
+    fn with_rocksdb_snapshot<F, R>(&self, f: F) -> ProviderResult<R>
     where
         Self: StorageSettingsCache,
-        F: FnOnce(RocksTxRefArg<'_>) -> ProviderResult<R>,
+        F: FnOnce(RocksDBRefArg<'_>) -> ProviderResult<R>,
     {
         if self.cached_storage_settings().storage_v2 {
             let rocksdb = self.rocksdb_provider();
-            let tx = rocksdb.tx();
-            return f(Some(&tx));
+            let snapshot = rocksdb.snapshot();
+            return f(Some(snapshot));
         }
         f(None)
     }
@@ -84,7 +88,7 @@ mod tests {
     use reth_db_api::models::StorageSettings;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
-    /// Mock `RocksDB` provider that tracks `tx()` calls.
+    /// Mock `RocksDB` provider that tracks snapshot creation calls.
     struct MockRocksDBProvider {
         tx_call_count: AtomicUsize,
     }
@@ -146,25 +150,29 @@ mod tests {
     }
 
     #[test]
-    fn test_legacy_settings_skip_rocksdb_tx_creation() {
+    fn test_legacy_settings_skip_rocksdb_snapshot() {
         let provider = TestProvider::new(StorageSettings::v1());
 
-        let result = provider.with_rocksdb_tx(|tx| {
-            assert!(tx.is_none(), "legacy settings should pass None tx");
+        let result = provider.with_rocksdb_snapshot(|rocksdb| {
+            assert!(rocksdb.is_none(), "legacy settings should pass None");
             Ok(42)
         });
 
         assert_eq!(result.unwrap(), 42);
-        assert_eq!(provider.tx_call_count(), 0, "should not create RocksDB tx for legacy settings");
+        assert_eq!(
+            provider.tx_call_count(),
+            0,
+            "should not create RocksDB provider for legacy settings"
+        );
     }
 
     #[test]
-    fn test_rocksdb_settings_create_tx() {
+    fn test_rocksdb_settings_create_snapshot() {
         let settings = StorageSettings::v2();
         let provider = TestProvider::new(settings);
 
-        let result = provider.with_rocksdb_tx(|tx| {
-            assert!(tx.is_some(), "rocksdb settings should pass Some tx");
+        let result = provider.with_rocksdb_snapshot(|rocksdb| {
+            assert!(rocksdb.is_some(), "rocksdb settings should pass Some snapshot");
             Ok(42)
         });
 
@@ -172,7 +180,7 @@ mod tests {
         assert_eq!(
             provider.tx_call_count(),
             1,
-            "should create RocksDB tx when any_in_rocksdb is true"
+            "should create RocksDB provider when storage_v2 is true"
         );
     }
 }

crates/transaction-pool/src/error.rs

@@ -145,6 +145,18 @@ impl PoolError {
             }
         }
     }
+
+    /// Returns `true` if this is a blob sidecar error that should NOT be cached as a bad import.
+    ///
+    /// The transaction hash may be valid — the issue is peer-specific (e.g. malformed sidecar
+    /// data), so we penalize the peer but allow re-fetching from other peers.
+    #[inline]
+    pub const fn is_bad_blob_sidecar(&self) -> bool {
+        match &self.kind {
+            PoolErrorKind::InvalidTransaction(err) => err.is_bad_blob_sidecar(),
+            _ => false,
+        }
+    }
 }
 
 /// Represents all errors that can happen when validating transactions for the pool for EIP-4844
@@ -400,6 +412,24 @@ impl InvalidPoolTransactionError {
         }
     }
 
+    /// Returns `true` if this is a blob sidecar error (e.g. invalid proof, missing sidecar).
+    ///
+    /// These errors indicate the sidecar data from a specific peer was bad, but the transaction
+    /// hash itself may be valid when fetched from another peer.
+    #[inline]
+    pub const fn is_bad_blob_sidecar(&self) -> bool {
+        matches!(
+            self,
+            Self::Eip4844(
+                Eip4844PoolTransactionError::MissingEip4844BlobSidecar |
+                    Eip4844PoolTransactionError::InvalidEip4844Blob(_) |
+                    Eip4844PoolTransactionError::UnexpectedEip7594SidecarBeforeOsaka |
+                    Eip4844PoolTransactionError::UnexpectedEip4844SidecarAfterOsaka |
+                    Eip4844PoolTransactionError::Eip7594SidecarDisallowed
+            )
+        )
+    }
+
     /// Returns true if this is a [`Self::Consensus`] variant.
     pub const fn as_consensus(&self) -> Option<&InvalidTransactionError> {
         match self {
@@ -475,4 +505,32 @@ mod tests {
 
         assert!(err.downcast_other_ref::<E>().is_some());
     }
+
+    #[test]
+    fn bad_blob_sidecar_detection() {
+        let err = PoolError::new(
+            TxHash::ZERO,
+            InvalidPoolTransactionError::Eip4844(Eip4844PoolTransactionError::InvalidEip4844Blob(
+                BlobTransactionValidationError::InvalidProof,
+            )),
+        );
+
+        assert!(err.is_bad_blob_sidecar());
+
+        let err = PoolError::new(
+            TxHash::ZERO,
+            InvalidPoolTransactionError::Eip4844(
+                Eip4844PoolTransactionError::MissingEip4844BlobSidecar,
+            ),
+        );
+
+        assert!(err.is_bad_blob_sidecar());
+
+        let err = PoolError::new(
+            TxHash::ZERO,
+            InvalidPoolTransactionError::Eip4844(Eip4844PoolTransactionError::NoEip4844Blobs),
+        );
+
+        assert!(!err.is_bad_blob_sidecar());
+    }
 }

crates/trie/common/src/prefix_set.rs

@@ -1,6 +1,7 @@
 use crate::Nibbles;
 use alloc::{sync::Arc, vec::Vec};
 use alloy_primitives::map::{B256Map, B256Set};
+use core::ops::Range;
 
 /// Collection of mutable prefix sets.
 #[derive(Clone, Default, Debug)]
@@ -225,6 +226,36 @@ impl PrefixSet {
         false
     }
 
+    /// Returns `true` if any key in the set falls within the given half-open range
+    /// `[start, end)`.
+    ///
+    /// Like [`Self::contains`], this method maintains the internal index for sequential access
+    /// optimization.
+    #[inline]
+    pub fn contains_range(&mut self, range: Range<&Nibbles>) -> bool {
+        if self.all {
+            return true
+        }
+
+        while self.index > 0 && &self.keys[self.index] >= range.end {
+            self.index -= 1;
+        }
+
+        for (idx, key) in self.keys[self.index..].iter().enumerate() {
+            if key >= range.start && key < range.end {
+                self.index += idx;
+                return true
+            }
+
+            if key >= range.end {
+                self.index += idx;
+                return false
+            }
+        }
+
+        false
+    }
+
     /// Returns an iterator over reference to _all_ nibbles regardless of cursor position.
     pub fn iter(&self) -> core::slice::Iter<'_, Nibbles> {
         self.keys.iter()

crates/trie/sparse/src/arena/mod.rs

@@ -2817,7 +2817,24 @@ impl SparseTrie for ArenaParallelSparseTrie {
 
                     let num_subtrie_updates = update_idx - subtrie_start;
 
-                    if num_subtrie_updates >= threshold {
+                    // If all updates are removals and could empty the subtrie,
+                    // force inline processing so the upper-arena collapse logic
+                    // can detect blinded siblings and request proofs.
+                    let all_removals = subtrie_updates
+                        .iter()
+                        // Filter out Touched, as they don't affect the structure of the trie. So an
+                        // update set with 2 removals and one Touched could still result in an empty
+                        // sub trie.
+                        .filter(|(_, _, u)| matches!(u, LeafUpdate::Changed(_)))
+                        .all(|(_, _, u)| matches!(u, LeafUpdate::Changed(v) if v.is_empty()));
+                    let subtrie_num_leaves = match &self.upper_arena[child_idx] {
+                        ArenaSparseNode::Subtrie(s) => s.num_leaves,
+                        _ => 0,
+                    };
+                    let might_empty_subtrie =
+                        all_removals && num_subtrie_updates as u64 >= subtrie_num_leaves;
+
+                    if num_subtrie_updates >= threshold && !might_empty_subtrie {
                         // Take subtrie for parallel update.
                         trace!(target: TRACE_TARGET, ?subtrie_root_path, num_subtrie_updates, "Taking subtrie for parallel update");
                         let ArenaSparseNode::Subtrie(subtrie) = mem::replace(
@@ -2978,7 +2995,10 @@ impl SparseTrie for ArenaParallelSparseTrie {
         }
 
         // Navigate to each taken subtrie via seek to propagate dirty state
-        // through intermediate branches, and collapse any EmptyRoot subtries.
+        // through intermediate branches. Taken subtries are guaranteed not to
+        // become EmptyRoot (the would-empty-subtrie check above forces those
+        // inline), so we only need to handle sibling collapses that may have
+        // occurred during inline processing while this subtrie was taken.
         {
             let mut cursor = mem::take(&mut self.buffers.cursor);
             cursor.reset(&self.upper_arena, self.root, Nibbles::default());
@@ -2987,13 +3007,17 @@ impl SparseTrie for ArenaParallelSparseTrie {
                 let find_result = cursor.seek(&mut self.upper_arena, path);
                 match find_result {
                     SeekResult::RevealedSubtrie => {
-                        let head_idx = cursor.head().expect("cursor is non-empty").index;
-                        if let ArenaSparseNode::Subtrie(s) = &self.upper_arena[head_idx] &&
-                            matches!(s.arena[s.root], ArenaSparseNode::EmptyRoot)
-                        {
-                            self.maybe_unwrap_subtrie(&mut cursor);
-                            continue;
-                        }
+                        debug_assert!(
+                            {
+                                let head_idx = cursor.head().expect("cursor is non-empty").index;
+                                !matches!(
+                                    &self.upper_arena[head_idx],
+                                    ArenaSparseNode::Subtrie(s) if matches!(s.arena[s.root], ArenaSparseNode::EmptyRoot)
+                                )
+                            },
+                            "taken subtrie became EmptyRoot — should have been forced inline"
+                        );
+
                         cursor.pop(&mut self.upper_arena);
 
                         // The parent branch (now at cursor top) may have had a sibling

crates/trie/sparse/tests/suite/commit_updates.rs

@@ -5,7 +5,7 @@ use super::*;
 /// After calling `commit_updates` with taken updates, a subsequent mutation + `root()` +
 /// `take_updates()` should only report the delta from the new baseline — it must NOT
 /// re-report branch nodes from the first round.
-pub(super) fn test_commit_updates_syncs_branch_masks<T: SparseTrie + Default>() {
+pub(super) fn test_commit_updates_syncs_branch_masks<T: SparseTrie>(new_trie: fn() -> T) {
     // 5 leaves spread across different subtrie regions.
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
@@ -27,7 +27,7 @@ pub(super) fn test_commit_updates_syncs_branch_masks<T: SparseTrie + Default>()
     ]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial hashes.
     let _ = trie.root();
@@ -79,7 +79,7 @@ pub(super) fn test_commit_updates_syncs_branch_masks<T: SparseTrie + Default>()
 /// Committing empty updated/removed sets should not change trie behavior.
 ///
 /// Build a trie, compute root, then commit empty updates. Root should be unchanged.
-pub(super) fn test_commit_updates_empty_is_noop<T: SparseTrie + Default>() {
+pub(super) fn test_commit_updates_empty_is_noop<T: SparseTrie>(new_trie: fn() -> T) {
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
     let mut key_b = B256::ZERO;
@@ -90,7 +90,7 @@ pub(super) fn test_commit_updates_empty_is_noop<T: SparseTrie + Default>() {
         BTreeMap::from([(key_a, U256::from(1)), (key_b, U256::from(2)), (key_c, U256::from(3))]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     let hash1 = trie.root();
     assert_eq!(hash1, harness.original_root());

crates/trie/sparse/tests/suite/find_leaf.rs

@@ -1,6 +1,6 @@
 use super::*;
 
-pub(super) fn test_find_leaf_exists<T: SparseTrie + Default>() {
+pub(super) fn test_find_leaf_exists<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -9,7 +9,7 @@ pub(super) fn test_find_leaf_exists<T: SparseTrie + Default>() {
         [(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))].into_iter().collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let trie: T = harness.init_trie_fully_revealed(false);
+    let trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let key2_nibbles = Nibbles::unpack(key2);
 
@@ -29,7 +29,7 @@ pub(super) fn test_find_leaf_exists<T: SparseTrie + Default>() {
     );
 }
 
-pub(super) fn test_find_leaf_nonexistent<T: SparseTrie + Default>() {
+pub(super) fn test_find_leaf_nonexistent<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -38,7 +38,7 @@ pub(super) fn test_find_leaf_nonexistent<T: SparseTrie + Default>() {
         [(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))].into_iter().collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let trie: T = harness.init_trie_fully_revealed(false);
+    let trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let nonexistent_key = B256::with_last_byte(0x99);
     let nonexistent_nibbles = Nibbles::unpack(nonexistent_key);
@@ -52,7 +52,7 @@ pub(super) fn test_find_leaf_nonexistent<T: SparseTrie + Default>() {
 
 /// `find_leaf` on a path that traverses a blinded node returns
 /// `Err(LeafLookupError::BlindedNode)`.
-pub(super) fn test_find_leaf_blinded<T: SparseTrie + Default>() {
+pub(super) fn test_find_leaf_blinded<T: SparseTrie>(new_trie: fn() -> T) {
     // Use ≥16 keys per nibble group so branch children become hash nodes (>32 bytes RLP),
     // ensuring partial reveal leaves blinded subtries.
     let mut base_storage = BTreeMap::new();
@@ -78,7 +78,7 @@ pub(super) fn test_find_leaf_blinded<T: SparseTrie + Default>() {
     let harness = SuiteTestHarness::new(base_storage);
 
     // Reveal only group_a keys, leaving group_b's subtrie blinded.
-    let trie: T = harness.init_trie_with_targets(&group_a_keys, false);
+    let trie: T = harness.init_trie_with_targets(&group_a_keys, false, new_trie);
 
     // Look up a key in group_b — should hit a blinded node.
     let blinded_key = group_b_keys[0];
@@ -93,7 +93,7 @@ pub(super) fn test_find_leaf_blinded<T: SparseTrie + Default>() {
 
 /// `find_leaf` with an expected value that doesn't match the actual leaf value
 /// returns `Err(LeafLookupError::ValueMismatch)`.
-pub(super) fn test_find_leaf_value_mismatch<T: SparseTrie + Default>() {
+pub(super) fn test_find_leaf_value_mismatch<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -102,7 +102,7 @@ pub(super) fn test_find_leaf_value_mismatch<T: SparseTrie + Default>() {
         [(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))].into_iter().collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let trie: T = harness.init_trie_fully_revealed(false);
+    let trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let key2_nibbles = Nibbles::unpack(key2);
     let wrong_value = encode_fixed_size(&U256::from(999)).to_vec();
@@ -119,7 +119,7 @@ pub(super) fn test_find_leaf_value_mismatch<T: SparseTrie + Default>() {
 ///
 /// Two leaves sharing prefix 0x12 create a branch with children at nibbles 3 and 5.
 /// Searching for a key with nibble 7 at that branch position should return `NonExistent`.
-pub(super) fn test_find_leaf_nonexistent_branch_divergence<T: SparseTrie + Default>() {
+pub(super) fn test_find_leaf_nonexistent_branch_divergence<T: SparseTrie>(new_trie: fn() -> T) {
     // key1 nibbles: 1,2,3,4,0,0,...  →  B256 = 0x12340000...
     let mut key1 = B256::ZERO;
     key1.0[0] = 0x12;
@@ -134,7 +134,7 @@ pub(super) fn test_find_leaf_nonexistent_branch_divergence<T: SparseTrie + Defau
         [(key1, U256::from(1)), (key2, U256::from(2))].into_iter().collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let trie: T = harness.init_trie_fully_revealed(false);
+    let trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // search_path nibbles: 1,2,7,8,0,0,...  →  B256 = 0x12780000...
     // Nibble 7 is unset at the branch (only 3 and 5 are set).
@@ -155,7 +155,7 @@ pub(super) fn test_find_leaf_nonexistent_branch_divergence<T: SparseTrie + Defau
 ///
 /// A single leaf at nibbles [1,2,3,4,5,6,...] creates an extension root with key 0x12.
 /// Searching for a key at nibbles [1,2,7,8,...] diverges from that extension.
-pub(super) fn test_find_leaf_nonexistent_extension_divergence<T: SparseTrie + Default>() {
+pub(super) fn test_find_leaf_nonexistent_extension_divergence<T: SparseTrie>(new_trie: fn() -> T) {
     // Single leaf: nibbles [1,2,3,4,5,6,0,0,...] → B256 = 0x12345600...
     let mut key1 = B256::ZERO;
     key1.0[0] = 0x12;
@@ -165,7 +165,7 @@ pub(super) fn test_find_leaf_nonexistent_extension_divergence<T: SparseTrie + De
     let base_storage: BTreeMap<B256, U256> = once((key1, U256::from(1))).collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let trie: T = harness.init_trie_fully_revealed(false);
+    let trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Search path diverges from the extension: nibbles [1,2,7,8,0,0,...] → B256 = 0x12780000...
     let mut search_key = B256::ZERO;
@@ -185,7 +185,7 @@ pub(super) fn test_find_leaf_nonexistent_extension_divergence<T: SparseTrie + De
 ///
 /// A single leaf at nibbles [1,2,3,4,0,0,...] exists. Searching for a key at nibbles
 /// [1,2,3,4,5,6,0,0,...] extends past the existing leaf — it should return `NonExistent`.
-pub(super) fn test_find_leaf_nonexistent_leaf_divergence<T: SparseTrie + Default>() {
+pub(super) fn test_find_leaf_nonexistent_leaf_divergence<T: SparseTrie>(new_trie: fn() -> T) {
     // Existing leaf at short path: nibbles [1,2,3,4,0,0,...] → B256 = 0x12340000...
     let mut existing_key = B256::ZERO;
     existing_key.0[0] = 0x12;
@@ -194,7 +194,7 @@ pub(super) fn test_find_leaf_nonexistent_leaf_divergence<T: SparseTrie + Default
     let base_storage: BTreeMap<B256, U256> = once((existing_key, U256::from(1))).collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let trie: T = harness.init_trie_fully_revealed(false);
+    let trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Search path extends past the existing leaf: nibbles [1,2,3,4,5,6,0,0,...]
     // → B256 = 0x12345600...

crates/trie/sparse/tests/suite/get_leaf_value.rs

@@ -2,7 +2,7 @@ use super::*;
 
 /// After inserting or updating a leaf via `update_leaves`, `get_leaf_value`
 /// should return the new value.
-pub(super) fn test_get_leaf_value_after_update<T: SparseTrie + Default>() {
+pub(super) fn test_get_leaf_value_after_update<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -11,7 +11,7 @@ pub(super) fn test_get_leaf_value_after_update<T: SparseTrie + Default>() {
         [(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))].into_iter().collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Insert a new leaf (key4) with value 42.
     let key4 = B256::with_last_byte(0x40);
@@ -42,7 +42,7 @@ pub(super) fn test_get_leaf_value_after_update<T: SparseTrie + Default>() {
 }
 
 /// After removing a leaf via `update_leaves`, `get_leaf_value` should return `None`.
-pub(super) fn test_get_leaf_value_after_removal<T: SparseTrie + Default>() {
+pub(super) fn test_get_leaf_value_after_removal<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -51,7 +51,7 @@ pub(super) fn test_get_leaf_value_after_removal<T: SparseTrie + Default>() {
         [(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))].into_iter().collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let key2_nibbles = Nibbles::unpack(key2);
     let expected_value_rlp = encode_fixed_size(&U256::from(2)).to_vec();

crates/trie/sparse/tests/suite/lifecycle.rs

@@ -5,7 +5,7 @@ use super::*;
 /// Build a trie with enough leaves to produce hashed branch children (≥16 per subtrie),
 /// insert 1 new leaf + modify 1 existing, compute root, take updates, commit, then verify
 /// root is unchanged (cache hit) and updates are non-empty.
-pub(super) fn test_full_lifecycle_update_root_take_commit<T: SparseTrie + Default>() {
+pub(super) fn test_full_lifecycle_update_root_take_commit<T: SparseTrie>(new_trie: fn() -> T) {
     // 16 leaves sharing prefix [1,0] to produce hashed branch children.
     let mut storage: BTreeMap<B256, U256> = BTreeMap::new();
     for i in 0u8..16 {
@@ -20,7 +20,7 @@ pub(super) fn test_full_lifecycle_update_root_take_commit<T: SparseTrie + Defaul
     storage.insert(key_extra, U256::from(100));
 
     let harness = SuiteTestHarness::new(storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial hashes.
     let _ = trie.root();
@@ -66,7 +66,7 @@ pub(super) fn test_full_lifecycle_update_root_take_commit<T: SparseTrie + Defaul
 
 /// Multiple rounds of (update → root → `take_updates` → `commit_updates`), followed by
 /// a prune, simulating block processing.
-pub(super) fn test_multi_round_update_commit_prune_cycle<T: SparseTrie + Default>() {
+pub(super) fn test_multi_round_update_commit_prune_cycle<T: SparseTrie>(new_trie: fn() -> T) {
     // Build a trie with 10 leaves.
     let mut storage: BTreeMap<B256, U256> = BTreeMap::new();
     let mut keys = Vec::new();
@@ -78,7 +78,7 @@ pub(super) fn test_multi_round_update_commit_prune_cycle<T: SparseTrie + Default
     }
 
     let mut harness = SuiteTestHarness::new(storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial hashes.
     let _ = trie.root();
@@ -146,7 +146,7 @@ pub(super) fn test_multi_round_update_commit_prune_cycle<T: SparseTrie + Default
 ///
 /// Build a 5-leaf trie, reveal all proofs, apply 2 modifications + 1 removal,
 /// and verify root matches the reference trie with the same mutations.
-pub(super) fn test_reveal_update_root_basic_lifecycle<T: SparseTrie + Default>() {
+pub(super) fn test_reveal_update_root_basic_lifecycle<T: SparseTrie>(new_trie: fn() -> T) {
     let mut keys = Vec::new();
     let mut storage: BTreeMap<B256, U256> = BTreeMap::new();
     for i in 0u8..5 {
@@ -157,7 +157,7 @@ pub(super) fn test_reveal_update_root_basic_lifecycle<T: SparseTrie + Default>()
     }
 
     let harness = SuiteTestHarness::new(storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Apply 2 modifications and 1 removal.
     let mut changeset: BTreeMap<B256, U256> = BTreeMap::new();
@@ -185,7 +185,7 @@ pub(super) fn test_reveal_update_root_basic_lifecycle<T: SparseTrie + Default>()
 
 /// Incremental reveal and update with retry loop.
 /// Partial proof → `update_leaves` hits blinded nodes → reveal more → retry succeeds.
-pub(super) fn test_incremental_reveal_and_update_with_retry<T: SparseTrie + Default>() {
+pub(super) fn test_incremental_reveal_and_update_with_retry<T: SparseTrie>(new_trie: fn() -> T) {
     // Build 10 leaves across multiple subtries so partial reveal leaves some blinded.
     // Use 16 keys per group so branch children become hash nodes (>32 bytes RLP).
     let mut base_storage = BTreeMap::new();
@@ -211,7 +211,7 @@ pub(super) fn test_incremental_reveal_and_update_with_retry<T: SparseTrie + Defa
     let harness = SuiteTestHarness::new(base_storage.clone());
 
     // Reveal only group A keys, leaving group B's subtrie blinded.
-    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, true);
+    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, true, new_trie);
 
     // Prepare updates for 5 keys: 3 from group A (covered) + 2 from group B (blinded).
     let mut changeset: BTreeMap<B256, U256> = BTreeMap::new();
@@ -269,7 +269,7 @@ pub(super) fn test_incremental_reveal_and_update_with_retry<T: SparseTrie + Defa
 /// Simulates a complete block processing cycle: receive state updates → apply to storage
 /// tries → compute storage roots → promote to account trie → compute state root → take
 /// updates → commit → prune for next block.
-pub(super) fn test_full_block_processing_lifecycle<T: SparseTrie + Default>() {
+pub(super) fn test_full_block_processing_lifecycle<T: SparseTrie>(new_trie: fn() -> T) {
     // --- Setup: Build account trie with 5 accounts ---
     // A1 storage: 5 slots, A2 storage: 3 slots, A3-A5: empty storage.
     // Account trie leaf values = RLP-encoded storage roots.
@@ -318,9 +318,9 @@ pub(super) fn test_full_block_processing_lifecycle<T: SparseTrie + Default>() {
     let mut acct_harness = SuiteTestHarness::new(acct_storage.clone());
 
     // Initialize all tries fully revealed with update tracking.
-    let mut a1_trie: T = a1_harness.init_trie_fully_revealed(true);
-    let mut a2_trie: T = a2_harness.init_trie_fully_revealed(true);
-    let mut acct_trie: T = acct_harness.init_trie_fully_revealed(true);
+    let mut a1_trie: T = a1_harness.init_trie_fully_revealed(true, new_trie);
+    let mut a2_trie: T = a2_harness.init_trie_fully_revealed(true, new_trie);
+    let mut acct_trie: T = acct_harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial hashes for all tries.
     let _ = a1_trie.root();
@@ -406,7 +406,7 @@ pub(super) fn test_full_block_processing_lifecycle<T: SparseTrie + Default>() {
 /// `Touched` is used to prewarm accounts/slots before actual state changes arrive.
 /// When the real `Changed` update arrives, it overwrites the `Touched` entry.
 /// This test verifies that prewarming followed by mutation works correctly.
-pub(super) fn test_touched_prewarm_then_changed_update<T: SparseTrie + Default>() {
+pub(super) fn test_touched_prewarm_then_changed_update<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -424,7 +424,7 @@ pub(super) fn test_touched_prewarm_then_changed_update<T: SparseTrie + Default>(
     .collect();
 
     let mut harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Step 1: Prewarm 3 keys with Touched — all should be drained (paths are revealed).
     let mut leaf_updates: B256Map<LeafUpdate> =
@@ -468,9 +468,9 @@ pub(super) fn test_touched_prewarm_then_changed_update<T: SparseTrie + Default>(
 /// A `Touched` update hits a blinded node, triggering a proof request. After the proof is
 /// revealed, a `Changed` update for the same key succeeds. This is the prewarm-miss →
 /// reveal → update sequence.
-pub(super) fn test_touched_on_blinded_triggers_proof_then_changed_succeeds<
-    T: SparseTrie + Default,
->() {
+pub(super) fn test_touched_on_blinded_triggers_proof_then_changed_succeeds<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Two groups of 16 keys each to create blinded subtries.
     let mut base_storage = BTreeMap::new();
 
@@ -493,7 +493,7 @@ pub(super) fn test_touched_on_blinded_triggers_proof_then_changed_succeeds<
     let mut harness = SuiteTestHarness::new(base_storage);
 
     // Reveal only group A — group B's subtrie is blinded.
-    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false);
+    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false, new_trie);
 
     // Step 1: Touched on a key in group B's blinded subtrie → callback fires.
     let target_key = group_b_keys[0];
@@ -540,7 +540,7 @@ pub(super) fn test_touched_on_blinded_triggers_proof_then_changed_succeeds<
 /// Simulates the `SparseStateTrie::update_account` pattern: read existing leaf via
 /// `get_leaf_value`, decode, modify (change one field while preserving another), re-encode,
 /// and update. Verifies that root matches reference.
-pub(super) fn test_get_leaf_value_for_storage_root_lookup<T: SparseTrie + Default>() {
+pub(super) fn test_get_leaf_value_for_storage_root_lookup<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -551,7 +551,7 @@ pub(super) fn test_get_leaf_value_for_storage_root_lookup<T: SparseTrie + Defaul
             .collect();
 
     let mut harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Step 1: Read existing leaf value via get_leaf_value.
     let key1_nibbles = Nibbles::unpack(key1);
@@ -589,7 +589,7 @@ pub(super) fn test_get_leaf_value_for_storage_root_lookup<T: SparseTrie + Defaul
 
 /// Before updating, check existence via `find_leaf`, then
 /// insert/modify, and verify `find_leaf` reflects the new state.
-pub(super) fn test_find_leaf_before_update_to_check_existence<T: SparseTrie + Default>() {
+pub(super) fn test_find_leaf_before_update_to_check_existence<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -599,7 +599,7 @@ pub(super) fn test_find_leaf_before_update_to_check_existence<T: SparseTrie + De
         [(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))].into_iter().collect();
 
     let mut harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let key2_nibbles = Nibbles::unpack(key2);
     let nonexistent_nibbles = Nibbles::unpack(nonexistent_key);
@@ -650,7 +650,7 @@ pub(super) fn test_find_leaf_before_update_to_check_existence<T: SparseTrie + De
 ///
 /// Build 10-leaf trie, do Block 1 (update K1,K2,K3 → commit → prune retaining K1,K2),
 /// then Block 2: update K1 (hot, works immediately), update K5 (cold, needs re-reveal).
-pub(super) fn test_prune_then_reuse_for_next_block<T: SparseTrie + Default>() {
+pub(super) fn test_prune_then_reuse_for_next_block<T: SparseTrie>(new_trie: fn() -> T) {
     // Build a trie with 10 leaves.
     let mut storage: BTreeMap<B256, U256> = BTreeMap::new();
     let mut keys = Vec::new();
@@ -662,7 +662,7 @@ pub(super) fn test_prune_then_reuse_for_next_block<T: SparseTrie + Default>() {
     }
 
     let mut harness = SuiteTestHarness::new(storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial hashes.
     let _ = trie.root();

crates/trie/sparse/tests/suite/main.rs

@@ -1,7 +1,7 @@
 //! Generic `SparseTrie` test suite.
 //!
-//! Tests are written as generic functions `test_foo<T: SparseTrie + Default>()` and stamped out
-//! for every concrete implementation via the [`sparse_trie_tests`] macro.
+//! Tests are written as generic functions `test_foo<T: SparseTrie>(new_trie: fn() -> T)` and
+//! stamped out for every concrete implementation via the [`sparse_trie_tests`] macro.
 //!
 //! Tests are organized into modules by which `SparseTrie` method is the most likely root cause
 //! of failure for each test case:
@@ -117,13 +117,14 @@ impl SuiteTestHarness {
 
     /// Initializes a trie with the harness root node and reveals all proof nodes for the
     /// given target keys. Returns the initialized trie.
-    fn init_trie_with_targets<T: SparseTrie + Default>(
+    fn init_trie_with_targets<T: SparseTrie>(
         &self,
         target_keys: &[B256],
         retain_updates: bool,
+        new_trie: fn() -> T,
     ) -> T {
         let root_node = self.root_node();
-        let mut trie = T::default();
+        let mut trie = (new_trie)();
         trie.set_root(root_node.node, root_node.masks, retain_updates)
             .expect("set_root should succeed");
 
@@ -138,9 +139,13 @@ impl SuiteTestHarness {
     }
 
     /// Initializes a trie and reveals proofs for all keys in the base storage.
-    fn init_trie_fully_revealed<T: SparseTrie + Default>(&self, retain_updates: bool) -> T {
+    fn init_trie_fully_revealed<T: SparseTrie>(
+        &self,
+        retain_updates: bool,
+        new_trie: fn() -> T,
+    ) -> T {
         let keys: Vec<B256> = self.storage().keys().copied().collect();
-        self.init_trie_with_targets(&keys, retain_updates)
+        self.init_trie_with_targets(&keys, retain_updates, new_trie)
     }
 }
 
@@ -158,7 +163,7 @@ macro_rules! sparse_trie_tests {
             $(
                 #[test]
                 fn $test_fn() {
-                    super::$test_fn::<ParallelSparseTrie>();
+                    super::$test_fn(ParallelSparseTrie::default);
                 }
             )*
         }
@@ -169,7 +174,27 @@ macro_rules! sparse_trie_tests {
             $(
                 #[test]
                 fn $test_fn() {
-                    super::$test_fn::<ArenaParallelSparseTrie>();
+                    super::$test_fn(ArenaParallelSparseTrie::default);
+                }
+            )*
+        }
+
+        mod arena_parallel_sparse_trie_always_parallel {
+            use reth_trie_sparse::{ArenaParallelSparseTrie, ArenaParallelismThresholds};
+
+            $(
+                #[test]
+                fn $test_fn() {
+                    super::$test_fn(|| {
+                        ArenaParallelSparseTrie::default().with_parallelism_thresholds(
+                            ArenaParallelismThresholds {
+                                min_dirty_leaves: 1,
+                                min_revealed_nodes: 1,
+                                min_updates: 1,
+                                min_leaves_for_prune: 1,
+                            },
+                        )
+                    });
                 }
             )*
         }
@@ -245,6 +270,8 @@ sparse_trie_tests! {
     test_orphaned_value_update_falls_through_to_full_insertion,
     test_branch_collapse_updates_leaf_key_len_across_subtries,
     test_remove_leaf_does_not_reveal_blind_subtries,
+    test_branch_collapse_multi_empty_subtries_blinded_remaining,
+    test_subtrie_emptied_by_deletes_with_touched,
 
     // root
     test_root_empty_trie,

crates/trie/sparse/tests/suite/prune.rs

@@ -1,6 +1,6 @@
 use super::*;
 
-pub(super) fn test_prune_retains_specified_leaves<T: SparseTrie + Default>() {
+pub(super) fn test_prune_retains_specified_leaves<T: SparseTrie>(new_trie: fn() -> T) {
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
     let mut key_b = B256::ZERO;
@@ -21,7 +21,7 @@ pub(super) fn test_prune_retains_specified_leaves<T: SparseTrie + Default>() {
     ]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Compute root before prune.
     let hash1 = trie.root();
@@ -47,7 +47,7 @@ pub(super) fn test_prune_retains_specified_leaves<T: SparseTrie + Default>() {
 /// Build a trie with 10+ leaves spread across multiple subtries, fully reveal
 /// and compute root. Then prune retaining only 1 leaf. `size_hint()` must
 /// decrease and `prune` must return > 0.
-pub(super) fn test_prune_reduces_node_count<T: SparseTrie + Default>() {
+pub(super) fn test_prune_reduces_node_count<T: SparseTrie>(new_trie: fn() -> T) {
     // Create 16 keys with different first nibbles to spread across subtries.
     let keys: Vec<B256> = (0u8..16)
         .map(|i| {
@@ -61,7 +61,7 @@ pub(super) fn test_prune_reduces_node_count<T: SparseTrie + Default>() {
         keys.iter().enumerate().map(|(i, k)| (*k, U256::from(i + 1))).collect();
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Compute root to cache hashes (required for pruning).
     let _root = trie.root();
@@ -83,7 +83,7 @@ pub(super) fn test_prune_reduces_node_count<T: SparseTrie + Default>() {
 
 /// Pruning with an empty retained set should convert all subtrees to
 /// hash stubs (maximum pruning). Root hash must be unchanged.
-pub(super) fn test_prune_empty_retained_set<T: SparseTrie + Default>() {
+pub(super) fn test_prune_empty_retained_set<T: SparseTrie>(new_trie: fn() -> T) {
     let keys: Vec<B256> = (0u8..16)
         .map(|i| {
             let mut k = B256::ZERO;
@@ -96,7 +96,7 @@ pub(super) fn test_prune_empty_retained_set<T: SparseTrie + Default>() {
         keys.iter().enumerate().map(|(i, k)| (*k, U256::from(i + 1))).collect();
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let hash_before = trie.root();
 
@@ -116,7 +116,7 @@ pub(super) fn test_prune_empty_retained_set<T: SparseTrie + Default>() {
     );
 }
 
-pub(super) fn test_prune_requires_computed_hashes<T: SparseTrie + Default>() {
+pub(super) fn test_prune_requires_computed_hashes<T: SparseTrie>(new_trie: fn() -> T) {
     let keys: Vec<B256> = (0u8..5)
         .map(|i| {
             let mut k = B256::ZERO;
@@ -129,7 +129,7 @@ pub(super) fn test_prune_requires_computed_hashes<T: SparseTrie + Default>() {
         keys.iter().enumerate().map(|(i, k)| (*k, U256::from(i + 1))).collect();
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Dirty the trie by updating a leaf — do NOT call root() to compute hashes.
     let mut leaf_updates: B256Map<LeafUpdate> = B256Map::default();
@@ -142,7 +142,7 @@ pub(super) fn test_prune_requires_computed_hashes<T: SparseTrie + Default>() {
 
     // Compare against pruning after root() is called (clean state).
     // With dirty nodes, pruning is limited because dirty subtrees lack cached hashes.
-    let mut trie_clean: T = harness.init_trie_fully_revealed(false);
+    let mut trie_clean: T = harness.init_trie_fully_revealed(false, new_trie);
     trie_clean.root();
     let clean_pruned = trie_clean.prune(&retained);
 
@@ -152,7 +152,7 @@ pub(super) fn test_prune_requires_computed_hashes<T: SparseTrie + Default>() {
     );
 }
 
-pub(super) fn test_prune_then_update_and_recompute_root<T: SparseTrie + Default>() {
+pub(super) fn test_prune_then_update_and_recompute_root<T: SparseTrie>(new_trie: fn() -> T) {
     let keys: Vec<B256> = (0u8..5)
         .map(|i| {
             let mut k = B256::ZERO;
@@ -165,7 +165,7 @@ pub(super) fn test_prune_then_update_and_recompute_root<T: SparseTrie + Default>
         keys.iter().enumerate().map(|(i, k)| (*k, U256::from(i + 1))).collect();
 
     let harness = SuiteTestHarness::new(storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     trie.root();
 
@@ -187,7 +187,7 @@ pub(super) fn test_prune_then_update_and_recompute_root<T: SparseTrie + Default>
     assert_eq!(root_after, expected_root, "root after prune + update should match reference trie");
 }
 
-pub(super) fn test_prune_then_reveal_pruned_subtree<T: SparseTrie + Default>() {
+pub(super) fn test_prune_then_reveal_pruned_subtree<T: SparseTrie>(new_trie: fn() -> T) {
     let keys: Vec<B256> = (0u8..5)
         .map(|i| {
             let mut k = B256::ZERO;
@@ -200,7 +200,7 @@ pub(super) fn test_prune_then_reveal_pruned_subtree<T: SparseTrie + Default>() {
         keys.iter().enumerate().map(|(i, k)| (*k, U256::from(i + 1))).collect();
 
     let harness = SuiteTestHarness::new(storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     trie.root();
 
@@ -227,7 +227,7 @@ pub(super) fn test_prune_then_reveal_pruned_subtree<T: SparseTrie + Default>() {
 
 /// Pruning a trie with both large (hashed) and small (embedded) node values
 /// should preserve the root hash.
-pub(super) fn test_prune_mixed_embedded_and_hashed_nodes<T: SparseTrie + Default>() {
+pub(super) fn test_prune_mixed_embedded_and_hashed_nodes<T: SparseTrie>(new_trie: fn() -> T) {
     let mut storage = BTreeMap::new();
 
     // 4 keys with large values (produce hashed nodes: RLP ≥ 32 bytes)
@@ -243,7 +243,7 @@ pub(super) fn test_prune_mixed_embedded_and_hashed_nodes<T: SparseTrie + Default
         storage.insert(key, U256::from(1));
     }
 
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     let mut leaf_updates = SuiteTestHarness::leaf_updates(&storage);
     trie.update_leaves(&mut leaf_updates, |_, _| {
         panic!("no proof callback expected on empty trie");
@@ -259,7 +259,7 @@ pub(super) fn test_prune_mixed_embedded_and_hashed_nodes<T: SparseTrie + Default
 
 /// After pruning, inserting a new leaf at a
 /// previously-unrevealed path should not panic.
-pub(super) fn test_prune_then_update_no_panic<T: SparseTrie + Default>() {
+pub(super) fn test_prune_then_update_no_panic<T: SparseTrie>(new_trie: fn() -> T) {
     // Build a trie with 64 leaves (16 keys × 4 first-nibble groups).
     let mut storage = BTreeMap::new();
     for group in 0..4u8 {
@@ -271,7 +271,7 @@ pub(super) fn test_prune_then_update_no_panic<T: SparseTrie + Default>() {
     }
 
     let harness = SuiteTestHarness::new(storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let root_before_prune = trie.root();
 
@@ -297,19 +297,19 @@ pub(super) fn test_prune_then_update_no_panic<T: SparseTrie + Default>() {
 
 /// When the root is not a branch (e.g., a single
 /// leaf or empty root), `prune` should immediately return 0 without walking.
-pub(super) fn test_prune_only_descends_into_branch_root<T: SparseTrie + Default>() {
+pub(super) fn test_prune_only_descends_into_branch_root<T: SparseTrie>(new_trie: fn() -> T) {
     // Single-leaf trie: root is a leaf node, not a branch.
     let storage: BTreeMap<B256, U256> =
         BTreeMap::from([(B256::with_last_byte(0x10), U256::from(1))]);
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let _root = trie.root();
     let pruned = trie.prune(&[]);
     assert_eq!(pruned, 0, "non-branch root should not prune any nodes");
 
     // Empty root: also not a branch.
-    let mut empty_trie = T::default();
+    let mut empty_trie = (new_trie)();
     let pruned_empty = empty_trie.prune(&[]);
     assert_eq!(pruned_empty, 0, "empty root should not prune any nodes");
 }
@@ -317,7 +317,7 @@ pub(super) fn test_prune_only_descends_into_branch_root<T: SparseTrie + Default>
 /// Small subtrie root nodes (RLP < 32 bytes) are
 /// handled correctly during prune. After `root()` + `prune()`, a subsequent `root()`
 /// still returns the same hash.
-pub(super) fn test_prune_handles_small_subtrie_root_nodes<T: SparseTrie + Default>() {
+pub(super) fn test_prune_handles_small_subtrie_root_nodes<T: SparseTrie>(new_trie: fn() -> T) {
     // Build a trie with two groups of leaves to create a branch root with mixed
     // subtrie sizes:
     // - Group A (nibble 0x1): 16 leaves with large values → hashable subtrie root (RLP ≥ 32 bytes)
@@ -335,7 +335,7 @@ pub(super) fn test_prune_handles_small_subtrie_root_nodes<T: SparseTrie + Defaul
     storage.insert(small_key, U256::from(1));
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let root_before = trie.root();
     assert_eq!(root_before, harness.original_root());

crates/trie/sparse/tests/suite/reveal_nodes.rs

@@ -4,7 +4,7 @@ use super::*;
 ///
 /// Calling `reveal_nodes` with an empty slice should return `Ok(())` and leave
 /// the trie state unchanged.
-pub(super) fn test_reveal_nodes_empty_slice<T: SparseTrie + Default>() {
+pub(super) fn test_reveal_nodes_empty_slice<T: SparseTrie>(new_trie: fn() -> T) {
     // Set up a trie with a root node.
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
@@ -15,7 +15,7 @@ pub(super) fn test_reveal_nodes_empty_slice<T: SparseTrie + Default>() {
 
     let harness = SuiteTestHarness::new(storage);
     let root_node = harness.root_node();
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     trie.set_root(root_node.node, root_node.masks, true).expect("set_root should succeed");
 
     let root_before = trie.root();
@@ -31,7 +31,7 @@ pub(super) fn test_reveal_nodes_empty_slice<T: SparseTrie + Default>() {
 ///
 /// Revealing a single leaf node within a branch should make it accessible and
 /// produce correct root hashes.
-pub(super) fn test_reveal_nodes_single_leaf<T: SparseTrie + Default>() {
+pub(super) fn test_reveal_nodes_single_leaf<T: SparseTrie>(new_trie: fn() -> T) {
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
     let mut key_b = B256::ZERO;
@@ -44,7 +44,7 @@ pub(super) fn test_reveal_nodes_single_leaf<T: SparseTrie + Default>() {
     let harness = SuiteTestHarness::new(storage);
 
     // Set root and reveal only one leaf's proof.
-    let mut trie: T = harness.init_trie_with_targets(&[key_a], true);
+    let mut trie: T = harness.init_trie_with_targets(&[key_a], true, new_trie);
     let root = trie.root();
     assert_eq!(root, harness.original_root());
 }
@@ -53,7 +53,7 @@ pub(super) fn test_reveal_nodes_single_leaf<T: SparseTrie + Default>() {
 ///
 /// Revealing the same proof nodes twice should not corrupt the trie or change
 /// the root hash. The second reveal is a no-op.
-pub(super) fn test_reveal_nodes_idempotent<T: SparseTrie + Default>() {
+pub(super) fn test_reveal_nodes_idempotent<T: SparseTrie>(new_trie: fn() -> T) {
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
     let mut key_b = B256::ZERO;
@@ -66,7 +66,7 @@ pub(super) fn test_reveal_nodes_idempotent<T: SparseTrie + Default>() {
     let harness = SuiteTestHarness::new(storage);
 
     // First reveal: set root and reveal all proof nodes.
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
     let root_first = trie.root();
     assert_eq!(root_first, harness.original_root());
 
@@ -85,7 +85,7 @@ pub(super) fn test_reveal_nodes_idempotent<T: SparseTrie + Default>() {
 /// Branch node masks provided during reveal should be stored and used for update tracking.
 /// After modifying a leaf and computing the root, `take_updates()` should contain entries
 /// reflecting which branch nodes were updated vs removed, guided by the stored masks.
-pub(super) fn test_reveal_nodes_with_branch_masks<T: SparseTrie + Default>() {
+pub(super) fn test_reveal_nodes_with_branch_masks<T: SparseTrie>(new_trie: fn() -> T) {
     // Build a trie with 16 leaves sharing first nibble 0x1 to produce non-root branch nodes
     // with hashed children (needed for masks to produce InsertUpdated actions).
     let mut storage: BTreeMap<B256, U256> = BTreeMap::new();
@@ -99,7 +99,7 @@ pub(super) fn test_reveal_nodes_with_branch_masks<T: SparseTrie + Default>() {
     let harness = SuiteTestHarness::new(storage);
 
     // Initialize trie with masks (from proofs) and retain_updates=true.
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Compute root to cache initial branch hashes.
     let _ = trie.root();
@@ -129,7 +129,7 @@ pub(super) fn test_reveal_nodes_with_branch_masks<T: SparseTrie + Default>() {
 ///
 /// Calling `reveal_nodes` when the root is `EmptyRoot` should return `Ok(())` without
 /// modifying trie state, even when non-empty proof nodes are provided.
-pub(super) fn test_reveal_nodes_skips_on_empty_root<T: SparseTrie + Default>() {
+pub(super) fn test_reveal_nodes_skips_on_empty_root<T: SparseTrie>(new_trie: fn() -> T) {
     // Build a harness with real data so we can obtain non-trivial proof nodes.
     let storage: BTreeMap<B256, U256> = BTreeMap::from([
         (B256::with_last_byte(1), U256::from(10)),
@@ -142,7 +142,7 @@ pub(super) fn test_reveal_nodes_skips_on_empty_root<T: SparseTrie + Default>() {
     let (mut proof_nodes, _) = harness.proof_v2(&mut targets);
 
     // Create a trie with an empty root.
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     trie.set_root(TrieNodeV2::EmptyRoot, None, true).expect("set_root EmptyRoot should succeed");
 
     // Reveal non-empty proof nodes — should be a no-op on an empty root.
@@ -161,7 +161,9 @@ pub(super) fn test_reveal_nodes_skips_on_empty_root<T: SparseTrie + Default>() {
 /// When `reveal_nodes` receives proof nodes that include entries not reachable from the
 /// current trie root (e.g., boundary leaves for unrelated subtries), those nodes should
 /// be silently skipped without corrupting state.
-pub(super) fn test_reveal_nodes_filters_unreachable_boundary_leaves<T: SparseTrie + Default>() {
+pub(super) fn test_reveal_nodes_filters_unreachable_boundary_leaves<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Create a trie with two groups of keys under different first nibbles.
     // Group A: 3 keys under nibble 0x1
     // Group B: 3 keys under nibble 0x2
@@ -192,7 +194,7 @@ pub(super) fn test_reveal_nodes_filters_unreachable_boundary_leaves<T: SparseTri
 
     // Initialize trie with root and reveal ONLY group A keys.
     let root_node = harness.root_node();
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     trie.set_root(root_node.node, root_node.masks, false).expect("set_root should succeed");
 
     let mut targets_a: Vec<ProofV2Target> = keys_a.iter().map(|k| ProofV2Target::new(*k)).collect();
@@ -237,7 +239,7 @@ pub(super) fn test_reveal_nodes_filters_unreachable_boundary_leaves<T: SparseTri
 /// When proofs from a 2-leaf trie are revealed, then a 3rd leaf is inserted, then another
 /// proof from the original 2-leaf trie is revealed, the branch node should not be overwritten
 /// by the stale proof. The root must match a reference trie with all 3 keys.
-pub(super) fn test_reveal_insert_reveal_preserves_branch_state<T: SparseTrie + Default>() {
+pub(super) fn test_reveal_insert_reveal_preserves_branch_state<T: SparseTrie>(new_trie: fn() -> T) {
     // Two original keys and one to insert.
     let key_a = B256::with_last_byte(0x00);
     let key_b = B256::with_last_byte(0x01);
@@ -249,7 +251,7 @@ pub(super) fn test_reveal_insert_reveal_preserves_branch_state<T: SparseTrie + D
     let harness = SuiteTestHarness::new(original_storage);
 
     // Initialize trie with root, reveal proof for key_a only.
-    let mut trie: T = harness.init_trie_with_targets(&[key_a], false);
+    let mut trie: T = harness.init_trie_with_targets(&[key_a], false, new_trie);
 
     // Insert key_b via update_leaves.
     let insert_value = U256::from(2);
@@ -277,7 +279,9 @@ pub(super) fn test_reveal_insert_reveal_preserves_branch_state<T: SparseTrie + D
 /// After removing a leaf that collapses a branch into an
 /// extension, revealing a stale proof (which had a branch at root) should not overwrite the
 /// extension node.
-pub(super) fn test_remove_then_reveal_does_not_overwrite_collapsed_node<T: SparseTrie + Default>() {
+pub(super) fn test_remove_then_reveal_does_not_overwrite_collapsed_node<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Nibbles [0,0,..], [1,1,..], [1,2,..] — root branch has children at nibbles 0 and 1.
     // Packed into B256 keys: byte 0x00 → nibbles [0,0], byte 0x11 → nibbles [1,1], etc.
     let key_a = {
@@ -302,7 +306,7 @@ pub(super) fn test_remove_then_reveal_does_not_overwrite_collapsed_node<T: Spars
     let harness = SuiteTestHarness::new(original_storage);
 
     // Initialize trie with root and reveal proofs for all keys.
-    let mut trie: T = harness.init_trie_with_targets(&[key_a, key_b, key_c], false);
+    let mut trie: T = harness.init_trie_with_targets(&[key_a, key_b, key_c], false, new_trie);
 
     // Remove key_a (0x0000..) — should collapse root branch into extension (shared prefix 0x01).
     let removals: BTreeMap<B256, U256> = BTreeMap::from([(key_a, U256::ZERO)]);
@@ -329,7 +333,9 @@ pub(super) fn test_remove_then_reveal_does_not_overwrite_collapsed_node<T: Spars
 /// After inserting a leaf that converts an extension root into
 /// a branch, revealing a stale proof from the original trie (which has an extension at root)
 /// should not overwrite the branch.
-pub(super) fn test_insert_then_reveal_does_not_overwrite_branch<T: SparseTrie + Default>() {
+pub(super) fn test_insert_then_reveal_does_not_overwrite_branch<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Original trie: keys 0x0001.. and 0x0002.. share prefix 0x00 → extension root.
     let key_a = {
         let mut k = B256::ZERO;
@@ -350,7 +356,7 @@ pub(super) fn test_insert_then_reveal_does_not_overwrite_branch<T: SparseTrie +
     let harness = SuiteTestHarness::new(original_storage);
 
     // Initialize trie with root, reveal all proofs.
-    let mut trie: T = harness.init_trie_with_targets(&[key_a, key_b], false);
+    let mut trie: T = harness.init_trie_with_targets(&[key_a, key_b], false, new_trie);
 
     // Insert key_c at 0x0100.. — different first nibble, forces extension→branch conversion.
     let key_c = {

crates/trie/sparse/tests/suite/root.rs

@@ -1,8 +1,8 @@
 use super::*;
 
 /// Calling `root()` on a fresh, empty trie returns `EMPTY_ROOT_HASH`.
-pub(super) fn test_root_empty_trie<T: SparseTrie + Default>() {
-    let mut trie = T::default();
+pub(super) fn test_root_empty_trie<T: SparseTrie>(new_trie: fn() -> T) {
+    let mut trie = (new_trie)();
     assert_eq!(trie.root(), EMPTY_ROOT_HASH, "empty trie should return EMPTY_ROOT_HASH");
 }
 
@@ -10,7 +10,7 @@ pub(super) fn test_root_empty_trie<T: SparseTrie + Default>() {
 ///
 /// After fully revealing and computing root once, calling `root()` again without
 /// mutations should return the same hash and `is_root_cached()` should be true.
-pub(super) fn test_root_cached_returns_without_recomputation<T: SparseTrie + Default>() {
+pub(super) fn test_root_cached_returns_without_recomputation<T: SparseTrie>(new_trie: fn() -> T) {
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
     let mut key_b = B256::ZERO;
@@ -21,7 +21,7 @@ pub(super) fn test_root_cached_returns_without_recomputation<T: SparseTrie + Def
         BTreeMap::from([(key_a, U256::from(1)), (key_b, U256::from(2)), (key_c, U256::from(3))]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     let root1 = trie.root();
     assert_eq!(root1, harness.original_root(), "first root should match reference");
@@ -36,7 +36,7 @@ pub(super) fn test_root_cached_returns_without_recomputation<T: SparseTrie + Def
 ///
 /// After modifying one leaf's value, `root()` should differ from the original and
 /// match the reference trie with the updated value.
-pub(super) fn test_root_after_single_leaf_update<T: SparseTrie + Default>() {
+pub(super) fn test_root_after_single_leaf_update<T: SparseTrie>(new_trie: fn() -> T) {
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
     let mut key_b = B256::ZERO;
@@ -47,7 +47,7 @@ pub(super) fn test_root_after_single_leaf_update<T: SparseTrie + Default>() {
         BTreeMap::from([(key_a, U256::from(1)), (key_b, U256::from(2)), (key_c, U256::from(3))]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let original_root = trie.root();
     assert_eq!(original_root, harness.original_root(), "initial root should match reference");
@@ -75,7 +75,7 @@ pub(super) fn test_root_after_single_leaf_update<T: SparseTrie + Default>() {
 ///
 /// Two tries built from the same 5 key-value pairs inserted in different orders
 /// must produce the same root hash.
-pub(super) fn test_root_deterministic_across_update_orders<T: SparseTrie + Default>() {
+pub(super) fn test_root_deterministic_across_update_orders<T: SparseTrie>(new_trie: fn() -> T) {
     // Define 5 key-value pairs spread across different subtrie regions.
     let mut k1 = B256::ZERO;
     k1.0[0] = 0x10;
@@ -99,7 +99,7 @@ pub(super) fn test_root_deterministic_across_update_orders<T: SparseTrie + Defau
     // Build trie A: insert keys in order 1,2,3,4,5.
     let order_a = [pairs[0], pairs[1], pairs[2], pairs[3], pairs[4]];
     let root_a = {
-        let mut trie = T::default();
+        let mut trie = (new_trie)();
         for (key, value) in &order_a {
             let mut leaf_updates =
                 SuiteTestHarness::leaf_updates(&BTreeMap::from([(*key, *value)]));
@@ -111,7 +111,7 @@ pub(super) fn test_root_deterministic_across_update_orders<T: SparseTrie + Defau
     // Build trie B: insert keys in order 5,3,1,4,2.
     let order_b = [pairs[4], pairs[2], pairs[0], pairs[3], pairs[1]];
     let root_b = {
-        let mut trie = T::default();
+        let mut trie = (new_trie)();
         for (key, value) in &order_b {
             let mut leaf_updates =
                 SuiteTestHarness::leaf_updates(&BTreeMap::from([(*key, *value)]));
@@ -132,7 +132,7 @@ pub(super) fn test_root_deterministic_across_update_orders<T: SparseTrie + Defau
 ///
 /// When the root node's RLP encoding is smaller than 32 bytes, `root()` must still return the
 /// correct hash. A second `root()` call should use the cached result without panic.
-pub(super) fn test_root_handles_small_root_node_without_hash<T: SparseTrie + Default>() {
+pub(super) fn test_root_handles_small_root_node_without_hash<T: SparseTrie>(new_trie: fn() -> T) {
     // A single small leaf produces a root node whose RLP is < 32 bytes.
     let key = B256::with_last_byte(1);
     let value = U256::from(1);
@@ -140,7 +140,7 @@ pub(super) fn test_root_handles_small_root_node_without_hash<T: SparseTrie + Def
     let storage: BTreeMap<B256, U256> = BTreeMap::from([(key, value)]);
     let harness = SuiteTestHarness::new(storage);
 
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let root1 = trie.root();
     assert_eq!(root1, harness.original_root(), "first root() should match reference trie");

crates/trie/sparse/tests/suite/set_root.rs

@@ -4,7 +4,7 @@ use super::*;
 ///
 /// Two leaves whose first nibbles differ produce a branch root node.
 /// After `set_root` + `reveal_nodes`, `root()` must match the reference hash.
-pub(super) fn test_set_root_with_branch_node<T: SparseTrie + Default>() {
+pub(super) fn test_set_root_with_branch_node<T: SparseTrie>(new_trie: fn() -> T) {
     // Keys whose first nibbles differ → branch at root.
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10; // first nibble = 1
@@ -14,7 +14,7 @@ pub(super) fn test_set_root_with_branch_node<T: SparseTrie + Default>() {
         BTreeMap::from([(key_a, U256::from(100)), (key_b, U256::from(200))]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
     let root = trie.root();
     assert_eq!(root, harness.original_root());
 }
@@ -23,12 +23,12 @@ pub(super) fn test_set_root_with_branch_node<T: SparseTrie + Default>() {
 ///
 /// A single key-value pair produces a leaf root node. After `set_root` + `root()`,
 /// the hash must match the reference trie.
-pub(super) fn test_set_root_with_leaf_node<T: SparseTrie + Default>() {
+pub(super) fn test_set_root_with_leaf_node<T: SparseTrie>(new_trie: fn() -> T) {
     let storage: BTreeMap<B256, U256> = BTreeMap::from([(B256::ZERO, U256::from(42))]);
 
     let harness = SuiteTestHarness::new(storage);
     let root_node = harness.root_node();
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     trie.set_root(root_node.node, root_node.masks, true).expect("set_root should succeed");
     let root = trie.root();
     assert_eq!(root, harness.original_root());
@@ -38,7 +38,7 @@ pub(super) fn test_set_root_with_leaf_node<T: SparseTrie + Default>() {
 ///
 /// Two keys sharing a long common prefix produce an extension root node.
 /// After `set_root` + `reveal_nodes`, `root()` must match the reference hash.
-pub(super) fn test_set_root_with_extension_node<T: SparseTrie + Default>() {
+pub(super) fn test_set_root_with_extension_node<T: SparseTrie>(new_trie: fn() -> T) {
     // Keys that share first byte 0xAB → extension root.
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0xAB;
@@ -49,7 +49,7 @@ pub(super) fn test_set_root_with_extension_node<T: SparseTrie + Default>() {
         BTreeMap::from([(key_a, U256::from(100)), (key_b, U256::from(200))]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
     let root = trie.root();
     assert_eq!(root, harness.original_root());
 }
@@ -58,7 +58,7 @@ pub(super) fn test_set_root_with_extension_node<T: SparseTrie + Default>() {
 ///
 /// When `set_root` is called with `retain_updates = true`, subsequent mutations
 /// should be tracked and `take_updates()` should return non-empty results.
-pub(super) fn test_set_root_retains_updates_when_requested<T: SparseTrie + Default>() {
+pub(super) fn test_set_root_retains_updates_when_requested<T: SparseTrie>(new_trie: fn() -> T) {
     // Build a trie with enough leaves to produce non-root branch nodes with hash children.
     // We need leaves sharing a prefix nibble so that intermediate branch nodes are created,
     // and enough entries that children are hashed (RLP ≥ 32 bytes).
@@ -71,7 +71,7 @@ pub(super) fn test_set_root_retains_updates_when_requested<T: SparseTrie + Defau
     }
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Compute root once so branch hashes are cached.
     let _ = trie.root();
@@ -101,7 +101,9 @@ pub(super) fn test_set_root_retains_updates_when_requested<T: SparseTrie + Defau
 ///
 /// When `set_root` is called with `retain_updates = false`, `take_updates()` should
 /// return an empty `SparseTrieUpdates` even after mutations.
-pub(super) fn test_set_root_does_not_retain_updates_when_not_requested<T: SparseTrie + Default>() {
+pub(super) fn test_set_root_does_not_retain_updates_when_not_requested<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
     let mut key_b = B256::ZERO;
@@ -113,7 +115,7 @@ pub(super) fn test_set_root_does_not_retain_updates_when_not_requested<T: Sparse
 
     let harness = SuiteTestHarness::new(storage);
     // retain_updates = false
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Modify a leaf.
     let changeset: BTreeMap<B256, U256> = BTreeMap::from([(key_a, U256::from(99))]);
@@ -135,8 +137,8 @@ pub(super) fn test_set_root_does_not_retain_updates_when_not_requested<T: Sparse
 ///
 /// Setting the root to `TrieNodeV2::EmptyRoot` should leave the trie in its initial
 /// empty state, returning `EMPTY_ROOT_HASH` from `root()`.
-pub(super) fn test_set_root_with_empty_root<T: SparseTrie + Default>() {
-    let mut trie = T::default();
+pub(super) fn test_set_root_with_empty_root<T: SparseTrie>(new_trie: fn() -> T) {
+    let mut trie = (new_trie)();
     trie.set_root(TrieNodeV2::EmptyRoot, None, true).expect("set_root should succeed");
     assert_eq!(trie.root(), EMPTY_ROOT_HASH);
 }

crates/trie/sparse/tests/suite/size_hint.rs

@@ -5,7 +5,7 @@ use super::*;
 /// Builds a 5-leaf trie, records `size_hint`, adds 2 leaves, records again,
 /// removes 1 leaf, records again. Asserts s2 > s1 and s3 < s2 (monotonic
 /// relative to leaf count changes).
-pub(super) fn test_size_hint_reflects_leaf_count<T: SparseTrie + Default>() {
+pub(super) fn test_size_hint_reflects_leaf_count<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -26,7 +26,7 @@ pub(super) fn test_size_hint_reflects_leaf_count<T: SparseTrie + Default>() {
 
     // Include new key targets so proofs cover them.
     let all_targets = vec![key1, key2, key3, key4, key5, new_key1, new_key2];
-    let mut trie: T = harness.init_trie_with_targets(&all_targets, false);
+    let mut trie: T = harness.init_trie_with_targets(&all_targets, false, new_trie);
 
     let s1 = trie.size_hint();
 

crates/trie/sparse/tests/suite/take_updates.rs

@@ -1,6 +1,8 @@
 use super::*;
 
-pub(super) fn test_take_updates_returns_empty_when_not_tracking<T: SparseTrie + Default>() {
+pub(super) fn test_take_updates_returns_empty_when_not_tracking<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x10;
     let mut key_b = B256::ZERO;
@@ -9,7 +11,7 @@ pub(super) fn test_take_updates_returns_empty_when_not_tracking<T: SparseTrie +
         BTreeMap::from([(key_a, U256::from(1)), (key_b, U256::from(2))]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let updates = trie.take_updates();
     assert!(updates.updated_nodes.is_empty(), "updated_nodes should be empty when not tracking");
@@ -20,7 +22,7 @@ pub(super) fn test_take_updates_returns_empty_when_not_tracking<T: SparseTrie +
 ///
 /// After `take_updates()`, subsequent updates should be tracked independently in a fresh
 /// accumulator. updates1 reflects only the A mutation, updates2 reflects only the B mutation.
-pub(super) fn test_take_updates_resets_after_take<T: SparseTrie + Default>() {
+pub(super) fn test_take_updates_resets_after_take<T: SparseTrie>(new_trie: fn() -> T) {
     let mut storage: BTreeMap<B256, U256> = BTreeMap::new();
     for i in 0u8..16 {
         let mut key = B256::ZERO;
@@ -30,7 +32,7 @@ pub(super) fn test_take_updates_resets_after_take<T: SparseTrie + Default>() {
     }
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial branch hashes.
     let _ = trie.root();
@@ -80,7 +82,9 @@ pub(super) fn test_take_updates_resets_after_take<T: SparseTrie + Default>() {
 /// (non-empty `BranchNodeMasks`). After removing one group entirely and modifying the
 /// other, `take_updates` should report real branches in `removed_nodes` and modified
 /// branches in `updated_nodes`, with the two sets mutually exclusive.
-pub(super) fn test_take_updates_contains_updated_and_removed_nodes<T: SparseTrie + Default>() {
+pub(super) fn test_take_updates_contains_updated_and_removed_nodes<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // 3-level branching under two groups:
     //
     // Group 0x1 (survives, gets modified):
@@ -122,7 +126,7 @@ pub(super) fn test_take_updates_contains_updated_and_removed_nodes<T: SparseTrie
     }
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial branch hashes.
     let _ = trie.root();
@@ -200,7 +204,9 @@ pub(super) fn test_take_updates_contains_updated_and_removed_nodes<T: SparseTrie
 ///   nibble 4, creating branch `[A,A,1,0]` (no short key) as a child of branch `[A,A,1]`. This
 ///   gives `[A,A,1]` non-empty `hash_mask`.
 /// - Changeset 2 removes both, collapsing `[A,A,1]` back to a leaf with empty masks.
-pub(super) fn test_take_updates_cross_cancellation_across_root_calls<T: SparseTrie + Default>() {
+pub(super) fn test_take_updates_cross_cancellation_across_root_calls<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     let val = U256::from(1u64);
 
     let mut key_existing = B256::ZERO;
@@ -227,7 +233,7 @@ pub(super) fn test_take_updates_cross_cancellation_across_root_calls<T: SparseTr
         [(key_existing, val), (key_b, val), (key_other, val)].into_iter().collect();
 
     let harness = SuiteTestHarness::new(initial);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial branch hashes.
     let _ = trie.root();
@@ -270,7 +276,9 @@ pub(super) fn test_take_updates_cross_cancellation_across_root_calls<T: SparseTr
 /// When a branch collapses (leaf removal) and then a new branch is created at the same path
 /// (leaf insertion), `take_updates` must not report the same path in both `updated_nodes` and
 /// `removed_nodes`. The insertion must win.
-pub(super) fn test_take_updates_no_duplicate_updated_and_removed_nodes<T: SparseTrie + Default>() {
+pub(super) fn test_take_updates_no_duplicate_updated_and_removed_nodes<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // 3 leaves sharing the first nibble → branch at nibble 0x0.
     let mut key_a = B256::ZERO;
     key_a.0[0] = 0x00;
@@ -283,7 +291,7 @@ pub(super) fn test_take_updates_no_duplicate_updated_and_removed_nodes<T: Sparse
         BTreeMap::from([(key_a, U256::from(1)), (key_b, U256::from(2)), (key_c, U256::from(3))]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Cache initial hashes.
     let _ = trie.root();

crates/trie/sparse/tests/suite/update_leaves.rs

@@ -4,7 +4,7 @@ use super::*;
 ///
 /// Starting from a 3-leaf trie, inserting a 4th key via `update_leaves` should produce
 /// a root hash matching a reference trie containing all 4 leaves.
-pub(super) fn test_update_leaves_insert_new_leaf<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_insert_new_leaf<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -17,7 +17,7 @@ pub(super) fn test_update_leaves_insert_new_leaf<T: SparseTrie + Default>() {
 
     // Initialize trie with all 3 existing keys revealed, plus the new key target.
     let all_targets = vec![key1, key2, key3, new_key];
-    let mut trie: T = harness.init_trie_with_targets(&all_targets, true);
+    let mut trie: T = harness.init_trie_with_targets(&all_targets, true, new_trie);
 
     // Insert the new leaf.
     let new_value = U256::from(4);
@@ -48,7 +48,7 @@ pub(super) fn test_update_leaves_insert_new_leaf<T: SparseTrie + Default>() {
 ///
 /// Starting from a 3-leaf trie, changing one leaf's value via `update_leaves` should
 /// produce a root hash matching a reference trie with the updated value.
-pub(super) fn test_update_leaves_modify_existing_leaf<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_modify_existing_leaf<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -57,7 +57,7 @@ pub(super) fn test_update_leaves_modify_existing_leaf<T: SparseTrie + Default>()
         BTreeMap::from([(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))]);
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Modify an existing leaf with a new value.
     let new_value = U256::from(999);
@@ -83,11 +83,11 @@ pub(super) fn test_update_leaves_modify_existing_leaf<T: SparseTrie + Default>()
 ///
 /// Calling `update_leaves` with one key on a default (empty) trie should produce a root
 /// hash matching a reference trie with that single leaf.
-pub(super) fn test_insert_single_leaf_into_empty_trie<T: SparseTrie + Default>() {
+pub(super) fn test_insert_single_leaf_into_empty_trie<T: SparseTrie>(new_trie: fn() -> T) {
     let key = B256::with_last_byte(42);
     let value = U256::from(1);
 
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     let mut leaf_updates = SuiteTestHarness::leaf_updates(&BTreeMap::from([(key, value)]));
 
     // Empty trie has no blinded nodes, so update_leaves should succeed in one call.
@@ -112,7 +112,7 @@ pub(super) fn test_insert_single_leaf_into_empty_trie<T: SparseTrie + Default>()
 ///
 /// All 256 keys are inserted in a single `update_leaves` call. The root must match
 /// a reference trie and `take_updates()` must return non-empty results.
-pub(super) fn test_insert_multiple_leaves_into_empty_trie<T: SparseTrie + Default>() {
+pub(super) fn test_insert_multiple_leaves_into_empty_trie<T: SparseTrie>(new_trie: fn() -> T) {
     // Build 256 keys with alternating prefix patterns (matching original test).
     let storage: BTreeMap<B256, U256> = (0..=255u8)
         .map(|b| {
@@ -123,7 +123,7 @@ pub(super) fn test_insert_multiple_leaves_into_empty_trie<T: SparseTrie + Defaul
 
     let expected_harness = SuiteTestHarness::new(storage.clone());
 
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     trie.set_updates(true);
 
     let mut leaf_updates = SuiteTestHarness::leaf_updates(&storage);
@@ -151,7 +151,7 @@ pub(super) fn test_insert_multiple_leaves_into_empty_trie<T: SparseTrie + Defaul
 /// Insert 256 keys with old values, compute root (hash1). Then update all 256 keys with
 /// new values, compute root (hash2). Both must match their respective reference tries,
 /// and hash1 ≠ hash2.
-pub(super) fn test_update_all_leaves_with_new_values<T: SparseTrie + Default>() {
+pub(super) fn test_update_all_leaves_with_new_values<T: SparseTrie>(new_trie: fn() -> T) {
     // Build 256 keys with alternating prefix patterns.
     let keys: Vec<B256> = (0..=255u8)
         .map(|b| if b % 2 == 0 { B256::repeat_byte(b) } else { B256::with_last_byte(b) })
@@ -164,7 +164,7 @@ pub(super) fn test_update_all_leaves_with_new_values<T: SparseTrie + Default>()
     let expected_old = SuiteTestHarness::new(old_storage.clone());
     let expected_new = SuiteTestHarness::new(new_storage.clone());
 
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     trie.set_updates(true);
 
     // Insert all 256 keys with old values.
@@ -194,14 +194,16 @@ pub(super) fn test_update_all_leaves_with_new_values<T: SparseTrie + Default>()
 /// Insert key `0x50..` then key `0x51..` (adjacent first-byte keys that share first nibble `5`),
 /// computing root after each. The final root must match the reference trie with both keys.
 /// `take_updates()` should return empty since no branch masks were set.
-pub(super) fn test_two_leaves_at_adjacent_keys_root_correctness<T: SparseTrie + Default>() {
+pub(super) fn test_two_leaves_at_adjacent_keys_root_correctness<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     let mut key_50 = B256::ZERO;
     key_50.0[0] = 0x50;
     let mut key_51 = B256::ZERO;
     key_51.0[0] = 0x51;
     let value = U256::from(1);
 
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     trie.set_updates(true);
 
     // Insert first leaf and compute root.
@@ -234,7 +236,7 @@ pub(super) fn test_two_leaves_at_adjacent_keys_root_correctness<T: SparseTrie +
 ///
 /// Starting from a 3-leaf trie, removing one key should produce a root hash
 /// matching a reference trie containing only the remaining 2 leaves.
-pub(super) fn test_update_leaves_remove_leaf<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_remove_leaf<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -243,7 +245,7 @@ pub(super) fn test_update_leaves_remove_leaf<T: SparseTrie + Default>() {
         BTreeMap::from([(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))]);
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Remove key2 by setting its value to U256::ZERO (produces LeafUpdate::Changed(vec![])).
     let mut leaf_updates = SuiteTestHarness::leaf_updates(&BTreeMap::from([(key2, U256::ZERO)]));
@@ -267,7 +269,7 @@ pub(super) fn test_update_leaves_remove_leaf<T: SparseTrie + Default>() {
 /// extension. Three leaves sharing prefix `0x5` create a branch at nibble 5; removing one
 /// child should collapse the structure. The root hash must match a reference trie with the
 /// remaining two leaves.
-pub(super) fn test_remove_leaf_branch_collapses_to_extension<T: SparseTrie + Default>() {
+pub(super) fn test_remove_leaf_branch_collapses_to_extension<T: SparseTrie>(new_trie: fn() -> T) {
     // Keys sharing prefix 0x5: two share 0x50 (children at 0x502..) and one at 0x53.
     // This creates a branch at nibble 5 with children at nibbles 0 and 3.
     let mut key_50231 = B256::ZERO;
@@ -291,7 +293,7 @@ pub(super) fn test_remove_leaf_branch_collapses_to_extension<T: SparseTrie + Def
     ]);
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Remove the leaf at key_537 — this collapses the branch at 0x5.
     let mut leaf_updates = SuiteTestHarness::leaf_updates(&BTreeMap::from([(key_537, U256::ZERO)]));
@@ -312,7 +314,7 @@ pub(super) fn test_remove_leaf_branch_collapses_to_extension<T: SparseTrie + Def
 /// Removing one of two leaves from a branch should collapse the
 /// branch into a leaf. Update tracking should report the root branch as removed and NOT as
 /// updated.
-pub(super) fn test_remove_leaf_branch_collapses_to_leaf<T: SparseTrie + Default>() {
+pub(super) fn test_remove_leaf_branch_collapses_to_leaf<T: SparseTrie>(new_trie: fn() -> T) {
     // Two leaves with different first nibbles → branch root.
     let key_a = B256::with_last_byte(0x10); // first nibble = 1
     let key_b = B256::with_last_byte(0x20); // first nibble = 2
@@ -321,7 +323,7 @@ pub(super) fn test_remove_leaf_branch_collapses_to_leaf<T: SparseTrie + Default>
         BTreeMap::from([(key_a, U256::from(100)), (key_b, U256::from(200))]);
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Compute root to cache hashes, take and commit updates to establish baseline masks.
     let _ = trie.root();
@@ -357,12 +359,12 @@ pub(super) fn test_remove_leaf_branch_collapses_to_leaf<T: SparseTrie + Default>
 
 /// Removing the only leaf in a trie should produce
 /// `EMPTY_ROOT_HASH`.
-pub(super) fn test_remove_last_leaf_produces_empty_root<T: SparseTrie + Default>() {
+pub(super) fn test_remove_last_leaf_produces_empty_root<T: SparseTrie>(new_trie: fn() -> T) {
     let key = B256::with_last_byte(0x12);
     let base_storage: BTreeMap<B256, U256> = BTreeMap::from([(key, U256::from(1))]);
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Remove the only leaf.
     let mut leaf_updates = SuiteTestHarness::leaf_updates(&BTreeMap::from([(key, U256::ZERO)]));
@@ -374,7 +376,7 @@ pub(super) fn test_remove_last_leaf_produces_empty_root<T: SparseTrie + Default>
 
 /// Build 6 leaves then remove one-by-one, verifying root at each
 /// step against a reference trie. Final removal produces `EMPTY_ROOT_HASH`.
-pub(super) fn test_insert_then_remove_sequence<T: SparseTrie + Default>() {
+pub(super) fn test_insert_then_remove_sequence<T: SparseTrie>(new_trie: fn() -> T) {
     // Helper: build a B256 key from a nibble prefix, zero-padded.
     let key_from_nibbles = |nibbles: &[u8]| -> B256 {
         let mut bytes = [0u8; 32];
@@ -402,7 +404,7 @@ pub(super) fn test_insert_then_remove_sequence<T: SparseTrie + Default>() {
     let base_storage: BTreeMap<B256, U256> = all_keys.iter().map(|&k| (k, val)).collect();
 
     let mut harness = SuiteTestHarness::new(base_storage.clone());
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     let mut leaf_updates = SuiteTestHarness::leaf_updates(&base_storage);
     harness.reveal_and_update(&mut trie, &mut leaf_updates);
 
@@ -436,7 +438,7 @@ pub(super) fn test_insert_then_remove_sequence<T: SparseTrie + Default>() {
 /// After computing `root()` (which caches hashes on all nodes), attempting to remove a key
 /// that doesn't exist should leave the cache intact so the next `root()` call returns the
 /// same hash without recomputation.
-pub(super) fn test_remove_nonexistent_leaf_preserves_hashes<T: SparseTrie + Default>() {
+pub(super) fn test_remove_nonexistent_leaf_preserves_hashes<T: SparseTrie>(new_trie: fn() -> T) {
     let key_a = B256::with_last_byte(0x10);
     let key_b = B256::with_last_byte(0x20);
     let key_c = B256::with_last_byte(0x30);
@@ -445,7 +447,7 @@ pub(super) fn test_remove_nonexistent_leaf_preserves_hashes<T: SparseTrie + Defa
         BTreeMap::from([(key_a, U256::from(1)), (key_b, U256::from(2)), (key_c, U256::from(3))]);
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Compute root to cache hashes on all nodes.
     let root_before = trie.root();
@@ -468,7 +470,7 @@ pub(super) fn test_remove_nonexistent_leaf_preserves_hashes<T: SparseTrie + Defa
 /// When `update_leaves` encounters a blinded node (insufficient
 /// proof data), it should invoke the `proof_required_fn` callback with the correct target key
 /// and minimum depth, and leave the key in the updates map for retry.
-pub(super) fn test_update_leaves_blinded_node_requests_proof<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_blinded_node_requests_proof<T: SparseTrie>(new_trie: fn() -> T) {
     // Use enough keys under two different first nibbles so that branch children become
     // hash nodes (>32 bytes RLP). This ensures partial reveal leaves blinded subtries.
     let mut base_storage = BTreeMap::new();
@@ -494,7 +496,7 @@ pub(super) fn test_update_leaves_blinded_node_requests_proof<T: SparseTrie + Def
     let harness = SuiteTestHarness::new(base_storage);
 
     // Reveal only group_a keys, leaving group_b's subtrie blinded.
-    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false);
+    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false, new_trie);
 
     // Try to modify a key in group_b's blinded subtrie.
     let target_key = group_b_keys[0];
@@ -518,7 +520,7 @@ pub(super) fn test_update_leaves_blinded_node_requests_proof<T: SparseTrie + Def
     );
 }
 
-pub(super) fn test_update_leaves_retry_after_reveal<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_retry_after_reveal<T: SparseTrie>(new_trie: fn() -> T) {
     // Same setup as blinded_node_requests_proof: two groups of 16 keys each under
     // different first nibbles, so branch children become hash nodes.
     let mut base_storage = BTreeMap::new();
@@ -542,7 +544,7 @@ pub(super) fn test_update_leaves_retry_after_reveal<T: SparseTrie + Default>() {
     let harness = SuiteTestHarness::new(base_storage.clone());
 
     // Reveal only group_a keys, leaving group_b's subtrie blinded.
-    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false);
+    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false, new_trie);
 
     // Modify a key in group_b's blinded subtrie.
     let target_key = group_b_keys[0];
@@ -585,7 +587,7 @@ pub(super) fn test_update_leaves_retry_after_reveal<T: SparseTrie + Default>() {
     );
 }
 
-pub(super) fn test_remove_leaf_blinded_sibling_requires_reveal<T: SparseTrie + Default>() {
+pub(super) fn test_remove_leaf_blinded_sibling_requires_reveal<T: SparseTrie>(new_trie: fn() -> T) {
     // Build a branch with two children: one revealed leaf at nibble 0x1, and a blinded
     // subtrie at nibble 0x2 (16 keys so it becomes a hash node > 32 bytes).
     let mut base_storage = BTreeMap::new();
@@ -608,7 +610,7 @@ pub(super) fn test_remove_leaf_blinded_sibling_requires_reveal<T: SparseTrie + D
     let harness = SuiteTestHarness::new(base_storage.clone());
 
     // Reveal only the single key at nibble 0x1, leaving nibble 0x2's subtrie blinded.
-    let mut trie: T = harness.init_trie_with_targets(&[revealed_key], false);
+    let mut trie: T = harness.init_trie_with_targets(&[revealed_key], false, new_trie);
 
     // Try to remove the revealed leaf. Branch collapse requires the blinded sibling.
     let mut leaf_updates =
@@ -649,9 +651,9 @@ pub(super) fn test_remove_leaf_blinded_sibling_requires_reveal<T: SparseTrie + D
 ///
 /// Same scenario as the value-preservation test, but additionally checks that
 /// `size_hint` (node count) is unchanged and the update remains in the map.
-pub(super) fn test_update_leaves_removal_branch_collapse_blinded_sibling<
-    T: SparseTrie + Default,
->() {
+pub(super) fn test_update_leaves_removal_branch_collapse_blinded_sibling<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Branch: nibble 0x1 = one revealed leaf, nibble 0x2 = 16 blinded keys (hash node).
     let mut base_storage = BTreeMap::new();
 
@@ -671,7 +673,7 @@ pub(super) fn test_update_leaves_removal_branch_collapse_blinded_sibling<
     let harness = SuiteTestHarness::new(base_storage);
 
     // Reveal only the leaf at nibble 0x1, leaving nibble 0x2 blinded.
-    let mut trie: T = harness.init_trie_with_targets(&[revealed_key], false);
+    let mut trie: T = harness.init_trie_with_targets(&[revealed_key], false, new_trie);
 
     // Snapshot state before the removal attempt.
     let revealed_path = Nibbles::unpack(revealed_key);
@@ -712,7 +714,9 @@ pub(super) fn test_update_leaves_removal_branch_collapse_blinded_sibling<
 /// When removals in a subtrie would empty it and collapse the parent branch onto
 /// a blinded sibling, `update_leaves` should detect this and request a proof for
 /// the blinded sibling via the callback, deferring the updates.
-pub(super) fn test_update_leaves_subtrie_collapse_requests_proof<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_subtrie_collapse_requests_proof<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Build a branch with two children:
     //   nibble 0x1 → a subtrie with 2 revealed leaves
     //   nibble 0x2 → 16 blinded keys (hash node > 32 bytes)
@@ -742,7 +746,8 @@ pub(super) fn test_update_leaves_subtrie_collapse_requests_proof<T: SparseTrie +
     let harness = SuiteTestHarness::new(base_storage);
 
     // Reveal only the two subtrie keys at nibble 0x1, leaving nibble 0x2 blinded.
-    let mut trie: T = harness.init_trie_with_targets(&[subtrie_key_a, subtrie_key_b], false);
+    let mut trie: T =
+        harness.init_trie_with_targets(&[subtrie_key_a, subtrie_key_b], false, new_trie);
 
     // Remove both leaves in the subtrie — this would empty the subtrie and
     // require collapsing the parent branch onto the blinded sibling at nibble 0x2.
@@ -770,7 +775,9 @@ pub(super) fn test_update_leaves_subtrie_collapse_requests_proof<T: SparseTrie +
 ///
 /// When multiple keys in the update map all route through the same blinded node,
 /// the callback should be invoked once per key (not deduplicated).
-pub(super) fn test_update_leaves_multiple_keys_same_blinded_node<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_multiple_keys_same_blinded_node<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Branch: nibble 0x1 = 16 revealed keys (hash node), nibble 0x2 = 16 blinded keys.
     let mut base_storage = BTreeMap::new();
 
@@ -791,7 +798,7 @@ pub(super) fn test_update_leaves_multiple_keys_same_blinded_node<T: SparseTrie +
     let harness = SuiteTestHarness::new(base_storage);
 
     // Reveal only group_a, leaving nibble 0x2 blinded.
-    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false);
+    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false, new_trie);
 
     // Submit 3 keys that all start with nibble 0x2 — they all hit the same blinded node.
     let blinded_keys: BTreeMap<B256, U256> = (0u8..3)
@@ -816,7 +823,7 @@ pub(super) fn test_update_leaves_multiple_keys_same_blinded_node<T: SparseTrie +
 }
 
 /// `LeafUpdate::Touched` on a fully revealed path should be a no-op.
-pub(super) fn test_update_leaves_touched_fully_revealed<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_touched_fully_revealed<T: SparseTrie>(new_trie: fn() -> T) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -825,7 +832,7 @@ pub(super) fn test_update_leaves_touched_fully_revealed<T: SparseTrie + Default>
         [(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))].into_iter().collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let root_before = trie.root();
 
@@ -848,7 +855,9 @@ pub(super) fn test_update_leaves_touched_fully_revealed<T: SparseTrie + Default>
 
 /// `LeafUpdate::Touched` on a path with a blinded node should
 /// invoke the callback and keep the key in the updates map. No trie mutation should occur.
-pub(super) fn test_update_leaves_touched_blinded_requests_proof<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_touched_blinded_requests_proof<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Two groups of 16 keys each under different first nibbles so that branch children
     // become hash nodes (>32 bytes RLP). Partial reveal leaves one subtrie blinded.
     let mut base_storage = BTreeMap::new();
@@ -870,7 +879,7 @@ pub(super) fn test_update_leaves_touched_blinded_requests_proof<T: SparseTrie +
     let harness = SuiteTestHarness::new(base_storage);
 
     // Reveal only group_a keys, leaving group_b's subtrie blinded.
-    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false);
+    let mut trie: T = harness.init_trie_with_targets(&group_a_keys, false, new_trie);
 
     let root_before = trie.root();
 
@@ -901,8 +910,8 @@ pub(super) fn test_update_leaves_touched_blinded_requests_proof<T: SparseTrie +
 /// An empty (default) trie has an Empty root — all paths are accessible (no blinded nodes).
 /// `Touched` on a key that doesn't exist should be drained from the map without any
 /// callback invocation or mutation.
-pub(super) fn test_update_leaves_touched_nonexistent_key<T: SparseTrie + Default>() {
-    let mut trie = T::default();
+pub(super) fn test_update_leaves_touched_nonexistent_key<T: SparseTrie>(new_trie: fn() -> T) {
+    let mut trie = (new_trie)();
 
     let target_key = B256::with_last_byte(42);
     let mut leaf_updates: B256Map<LeafUpdate> = once((target_key, LeafUpdate::Touched)).collect();
@@ -924,7 +933,9 @@ pub(super) fn test_update_leaves_touched_nonexistent_key<T: SparseTrie + Default
 
 /// `LeafUpdate::Touched` on a nonexistent key in a fully
 /// revealed, populated trie should be a no-op — no callback, key drained, trie unchanged.
-pub(super) fn test_update_leaves_touched_nonexistent_in_populated_trie<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_touched_nonexistent_in_populated_trie<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     let key1 = B256::with_last_byte(0x10);
     let key2 = B256::with_last_byte(0x20);
     let key3 = B256::with_last_byte(0x30);
@@ -933,7 +944,7 @@ pub(super) fn test_update_leaves_touched_nonexistent_in_populated_trie<T: Sparse
         [(key1, U256::from(1)), (key2, U256::from(2)), (key3, U256::from(3))].into_iter().collect();
 
     let harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     let root_before = trie.root();
 
@@ -960,7 +971,7 @@ pub(super) fn test_update_leaves_touched_nonexistent_in_populated_trie<T: Sparse
 
 /// A single `update_leaves` call with a mix of inserts,
 /// modifications, removals, and touched entries should process all correctly.
-pub(super) fn test_update_leaves_multiple_mixed_updates<T: SparseTrie + Default>() {
+pub(super) fn test_update_leaves_multiple_mixed_updates<T: SparseTrie>(new_trie: fn() -> T) {
     let key_a = B256::with_last_byte(0x10); // will be inserted (new key)
     let key_b = B256::with_last_byte(0x20); // will be modified
     let key_c = B256::with_last_byte(0x30); // will be removed
@@ -980,7 +991,7 @@ pub(super) fn test_update_leaves_multiple_mixed_updates<T: SparseTrie + Default>
 
     // Fully reveal existing trie, plus proof for key_a (new key to be inserted).
     let all_keys = vec![key_a, key_b, key_c, key_d, key_e];
-    let mut trie: T = harness.init_trie_with_targets(&all_keys, false);
+    let mut trie: T = harness.init_trie_with_targets(&all_keys, false, new_trie);
 
     // Build mixed leaf updates.
     let new_value_a = U256::from(100);
@@ -1027,7 +1038,9 @@ pub(super) fn test_update_leaves_multiple_mixed_updates<T: SparseTrie + Default>
 /// not just those that previously had a cached hash. This test inserts leaves without
 /// calling `root()` (so no hashes are cached), then removes a leaf and verifies
 /// `root()` returns the correct hash.
-pub(super) fn test_remove_leaf_marks_ancestors_dirty_unconditionally<T: SparseTrie + Default>() {
+pub(super) fn test_remove_leaf_marks_ancestors_dirty_unconditionally<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Create a trie with 5 leaves.
     let mut keys = Vec::new();
     let mut storage: BTreeMap<B256, U256> = BTreeMap::new();
@@ -1042,7 +1055,7 @@ pub(super) fn test_remove_leaf_marks_ancestors_dirty_unconditionally<T: SparseTr
 
     // Initialize trie: set root and reveal all proofs.
     let root_node = harness.root_node();
-    let mut trie = T::default();
+    let mut trie = (new_trie)();
     trie.set_root(root_node.node, root_node.masks, false).expect("set_root should succeed");
 
     let mut targets: Vec<ProofV2Target> = keys.iter().map(|k| ProofV2Target::new(*k)).collect();
@@ -1082,9 +1095,9 @@ pub(super) fn test_remove_leaf_marks_ancestors_dirty_unconditionally<T: SparseTr
 /// every key with a value must remain findable via `find_leaf` and updatable
 /// via `update_leaves`. This verifies the invariant that structural consistency
 /// is maintained even when branch collapses could orphan value entries.
-pub(super) fn test_orphaned_value_update_falls_through_to_full_insertion<
-    T: SparseTrie + Default,
->() {
+pub(super) fn test_orphaned_value_update_falls_through_to_full_insertion<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Create a trie with 3 leaves sharing a branch prefix, plus 2 additional leaves
     // in different subtries. Keys chosen so removal of key_c collapses the branch
     // at the shared prefix.
@@ -1128,7 +1141,7 @@ pub(super) fn test_orphaned_value_update_falls_through_to_full_insertion<
     .collect();
 
     let mut harness = SuiteTestHarness::new(initial_storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Insert all leaves.
     let mut insert_updates = SuiteTestHarness::leaf_updates(&initial_storage);
@@ -1185,7 +1198,9 @@ pub(super) fn test_orphaned_value_update_falls_through_to_full_insertion<
 /// When removing a leaf causes a branch to collapse at a subtrie
 /// boundary, the remaining sibling leaf's `key_len` metadata must be updated. After the
 /// collapse the remaining leaf must be findable, updatable, and contribute to the correct root.
-pub(super) fn test_branch_collapse_updates_leaf_key_len_across_subtries<T: SparseTrie + Default>() {
+pub(super) fn test_branch_collapse_updates_leaf_key_len_across_subtries<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     // Create two leaves that share a branch at a subtrie boundary.
     // Keys share the same first nibble (0x1) so they form a branch one level down,
     // which is a subtrie boundary for the parallel sparse trie.
@@ -1196,7 +1211,7 @@ pub(super) fn test_branch_collapse_updates_leaf_key_len_across_subtries<T: Spars
         BTreeMap::from([(key_a, U256::from(100)), (key_b, U256::from(200))]);
 
     let mut harness = SuiteTestHarness::new(base_storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Step 1: Remove key_a → branch collapses, key_b becomes the sole child.
     let removal: BTreeMap<B256, U256> = once((key_a, U256::ZERO)).collect();
@@ -1239,7 +1254,7 @@ pub(super) fn test_branch_collapse_updates_leaf_key_len_across_subtries<T: Spars
 /// When a branch collapses during leaf removal and the remaining child's value needs to be
 /// moved between subtries, the operation must not reveal (initialize) blind/unloaded subtries.
 /// Either the removal succeeds cleanly or it requests proofs for the blinded area.
-pub(super) fn test_remove_leaf_does_not_reveal_blind_subtries<T: SparseTrie + Default>() {
+pub(super) fn test_remove_leaf_does_not_reveal_blind_subtries<T: SparseTrie>(new_trie: fn() -> T) {
     // Create a trie with 10 leaves across different first-nibble subtries.
     // We'll prune some subtries, then remove a leaf whose branch collapse
     // involves a sibling in a pruned (blinded) subtrie.
@@ -1253,7 +1268,7 @@ pub(super) fn test_remove_leaf_does_not_reveal_blind_subtries<T: SparseTrie + De
     }
 
     let mut harness = SuiteTestHarness::new(storage.clone());
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Compute initial root and commit to establish baseline.
     let _ = trie.root();
@@ -1307,3 +1322,148 @@ pub(super) fn test_remove_leaf_does_not_reveal_blind_subtries<T: SparseTrie + De
         "root after modifying retained leaf should match reference"
     );
 }
+
+/// Branch collapse across multiple emptied subtries with blinded remaining child.
+///
+/// A branch at nibble `0xd` has 3 children: two revealed subtries (at `0xd7` and `0xdd`)
+/// each containing a single leaf, and one blinded subtrie (at `0xd8`). Removing both
+/// revealed leaves empties their subtries, leaving the branch with a single blinded child
+/// that cannot be collapsed without a proof.
+///
+/// ```text
+/// root (branch)
+///  └─ 0xd (branch, 3 children)
+///       ├─ 0xd7 → Leaf (revealed)
+///       ├─ 0xd8 → Leaf (BLINDED)
+///       └─ 0xdd → Leaf (revealed)
+///
+/// After removing 0xd7 and 0xdd:
+///   0xd branch has single child (0xd8), must collapse,
+///   but 0xd8 is blinded → needs proof
+/// ```
+pub(super) fn test_branch_collapse_multi_empty_subtries_blinded_remaining<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
+    // Three keys sharing first nibble 0xd, differing at second nibble.
+    let key_d7 = {
+        let mut key = B256::ZERO;
+        key.0[0] = 0xd7;
+        key
+    };
+    let key_d8 = {
+        let mut key = B256::ZERO;
+        key.0[0] = 0xd8;
+        key
+    };
+    let key_dd = {
+        let mut key = B256::ZERO;
+        key.0[0] = 0xdd;
+        key
+    };
+
+    let base_storage: BTreeMap<B256, U256> =
+        BTreeMap::from([(key_d7, U256::from(1)), (key_d8, U256::from(2)), (key_dd, U256::from(3))]);
+
+    let harness = SuiteTestHarness::new(base_storage);
+
+    // Reveal only 0xd7 and 0xdd, leaving 0xd8's subtrie blinded.
+    let mut trie: T = harness.init_trie_with_targets(&[key_d7, key_dd], false, new_trie);
+
+    // Remove both revealed leaves — their subtries empty, branch collapses to
+    // single child (0xd8) which is blinded.
+    let mut leaf_updates = SuiteTestHarness::leaf_updates(&BTreeMap::from([
+        (key_d7, U256::ZERO),
+        (key_dd, U256::ZERO),
+    ]));
+
+    let mut targets: Vec<ProofV2Target> = Vec::new();
+    trie.update_leaves(&mut leaf_updates, |key, min_len| {
+        targets.push(ProofV2Target::new(key).with_min_len(min_len));
+    })
+    .expect("update_leaves should succeed");
+
+    // Callback should fire for the blinded child at 0xd8.
+    assert!(!targets.is_empty(), "callback should fire for blinded child during branch collapse");
+    // Removal keys should remain in the map for retry.
+    assert!(!leaf_updates.is_empty(), "removal keys should remain in map after blinded hit");
+
+    // Reveal the blinded subtrie.
+    let (mut proof_nodes, _) = harness.proof_v2(&mut targets);
+    trie.reveal_nodes(&mut proof_nodes).expect("reveal_nodes should succeed");
+
+    // Retry — now the sibling is revealed, branch can collapse.
+    trie.update_leaves(&mut leaf_updates, |_, _| {})
+        .expect("update_leaves should succeed on retry");
+    assert!(leaf_updates.is_empty(), "keys should be drained after successful retry");
+
+    // Root should match reference trie with only key_d8.
+    let expected_harness = SuiteTestHarness::new(BTreeMap::from([(key_d8, U256::from(2))]));
+    let root = trie.root();
+    assert_eq!(
+        root,
+        expected_harness.original_root(),
+        "root should match trie with only the previously-blinded leaf"
+    );
+}
+
+/// Regression: subtrie emptied by deletes mixed with `LeafUpdate::Touched`.
+///
+/// When all `Changed` updates in a subtrie are removals and they would empty the subtrie,
+/// the `might_empty_subtrie` guard must still trigger even if `Touched` entries are present.
+/// `Touched` is a no-op that doesn't prevent the subtrie from being emptied.
+pub(super) fn test_subtrie_emptied_by_deletes_with_touched<T: SparseTrie>(new_trie: fn() -> T) {
+    // Two leaves under prefix 0xAB (the target subtrie), one under 0xAC (sibling at
+    // depth 1 to force the 0xAB child into a subtrie at depth 2), one under 0xCD
+    // (sibling at depth 0 to force a branch at the root).
+    let mut key_ab1 = B256::ZERO;
+    key_ab1[0] = 0xAB;
+    key_ab1[31] = 0x11;
+    let mut key_ab2 = B256::ZERO;
+    key_ab2[0] = 0xAB;
+    key_ab2[31] = 0x22;
+    let mut key_ab3 = B256::ZERO;
+    key_ab3[0] = 0xAB;
+    key_ab3[31] = 0x33;
+    let mut key_ac1 = B256::ZERO;
+    key_ac1[0] = 0xAC;
+    key_ac1[31] = 0x44;
+    let mut key_cd1 = B256::ZERO;
+    key_cd1[0] = 0xCD;
+    key_cd1[31] = 0x01;
+
+    let value = U256::from(1u64);
+
+    let base_storage: BTreeMap<B256, U256> =
+        [(key_ab1, value), (key_ab2, value), (key_ac1, value), (key_cd1, value)]
+            .into_iter()
+            .collect();
+
+    let harness = SuiteTestHarness::new(base_storage.clone());
+    let all_keys = vec![key_ab1, key_ab2, key_ac1, key_cd1];
+    let mut trie: T = harness.init_trie_with_targets(&all_keys, false, new_trie);
+
+    // Verify initial root matches.
+    let root = trie.root();
+    assert_eq!(root, harness.original_root(), "initial root mismatch");
+
+    // Delete both 0xAB leaves + Touched on a third 0xAB key (not in the trie).
+    // Touched is a no-op but must not prevent the might_empty_subtrie guard.
+    let mut leaf_updates: B256Map<LeafUpdate> = [
+        (key_ab1, LeafUpdate::Changed(Vec::new())),
+        (key_ab2, LeafUpdate::Changed(Vec::new())),
+        (key_ab3, LeafUpdate::Touched),
+    ]
+    .into_iter()
+    .collect();
+
+    harness.reveal_and_update(&mut trie, &mut leaf_updates);
+
+    // Root should match reference trie with ab1 and ab2 removed.
+    let mut expected_storage = base_storage;
+    expected_storage.remove(&key_ab1);
+    expected_storage.remove(&key_ab2);
+    let expected_harness = SuiteTestHarness::new(expected_storage);
+
+    let actual_root = trie.root();
+    assert_eq!(actual_root, expected_harness.original_root(), "post-delete root mismatch");
+}

crates/trie/sparse/tests/suite/wipe_clear.rs

@@ -2,7 +2,7 @@ use super::*;
 
 /// Calling `wipe()` resets the trie so that
 /// `root()` returns `EMPTY_ROOT_HASH`.
-pub(super) fn test_wipe_resets_to_empty_root<T: SparseTrie + Default>() {
+pub(super) fn test_wipe_resets_to_empty_root<T: SparseTrie>(new_trie: fn() -> T) {
     let storage: BTreeMap<B256, U256> = BTreeMap::from([
         (B256::with_last_byte(0x10), U256::from(1)),
         (B256::with_last_byte(0x20), U256::from(2)),
@@ -12,7 +12,7 @@ pub(super) fn test_wipe_resets_to_empty_root<T: SparseTrie + Default>() {
     ]);
 
     let harness = SuiteTestHarness::new(storage);
-    let mut trie: T = harness.init_trie_fully_revealed(false);
+    let mut trie: T = harness.init_trie_fully_revealed(false, new_trie);
 
     // Compute root to confirm the trie is populated.
     let root_before = trie.root();
@@ -28,7 +28,9 @@ pub(super) fn test_wipe_resets_to_empty_root<T: SparseTrie + Default>() {
 /// `clear()` resets the trie to empty but preserves
 /// update tracking mode. After clear, `root()` returns `EMPTY_ROOT_HASH` and
 /// `take_updates()` returns empty (non-wiped) updates.
-pub(super) fn test_clear_resets_trie_but_preserves_update_tracking<T: SparseTrie + Default>() {
+pub(super) fn test_clear_resets_trie_but_preserves_update_tracking<T: SparseTrie>(
+    new_trie: fn() -> T,
+) {
     let storage: BTreeMap<B256, U256> = BTreeMap::from([
         (B256::with_last_byte(0x10), U256::from(1)),
         (B256::with_last_byte(0x20), U256::from(2)),
@@ -37,7 +39,7 @@ pub(super) fn test_clear_resets_trie_but_preserves_update_tracking<T: SparseTrie
 
     let harness = SuiteTestHarness::new(storage);
     // retain_updates = true so update tracking is active
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Compute root to populate the trie fully.
     let root_before = trie.root();
@@ -59,7 +61,7 @@ pub(super) fn test_clear_resets_trie_but_preserves_update_tracking<T: SparseTrie
 /// `wipe()` produces special "wiped" updates distinct
 /// from normal empty updates. After wipe, `take_updates()` returns updates with
 /// the wiped flag set and `root()` returns `EMPTY_ROOT_HASH`.
-pub(super) fn test_wipe_produces_wiped_updates<T: SparseTrie + Default>() {
+pub(super) fn test_wipe_produces_wiped_updates<T: SparseTrie>(new_trie: fn() -> T) {
     let storage: BTreeMap<B256, U256> = BTreeMap::from([
         (B256::with_last_byte(0x10), U256::from(1)),
         (B256::with_last_byte(0x20), U256::from(2)),
@@ -68,7 +70,7 @@ pub(super) fn test_wipe_produces_wiped_updates<T: SparseTrie + Default>() {
 
     let harness = SuiteTestHarness::new(storage);
     // retain_updates = true so update tracking is active
-    let mut trie: T = harness.init_trie_fully_revealed(true);
+    let mut trie: T = harness.init_trie_fully_revealed(true, new_trie);
 
     // Compute root to populate the trie fully.
     let root_before = trie.root();
@@ -90,7 +92,7 @@ pub(super) fn test_wipe_produces_wiped_updates<T: SparseTrie + Default>() {
 /// A cleared trie can be fully re-initialized and used
 /// normally. After `clear()`, set a new root from a different dataset, reveal
 /// nodes, insert a leaf, and verify `root()` matches the reference.
-pub(super) fn test_clear_then_reuse_trie<T: SparseTrie + Default>() {
+pub(super) fn test_clear_then_reuse_trie<T: SparseTrie>(new_trie: fn() -> T) {
     // Phase 1: build a trie with 5 leaves and compute root.
     let storage_1: BTreeMap<B256, U256> = BTreeMap::from([
         (B256::with_last_byte(0x10), U256::from(1)),
@@ -100,7 +102,7 @@ pub(super) fn test_clear_then_reuse_trie<T: SparseTrie + Default>() {
         (B256::with_last_byte(0x50), U256::from(5)),
     ]);
     let harness_1 = SuiteTestHarness::new(storage_1);
-    let mut trie: T = harness_1.init_trie_fully_revealed(false);
+    let mut trie: T = harness_1.init_trie_fully_revealed(false, new_trie);
 
     let root_1 = trie.root();
     assert_eq!(root_1, harness_1.original_root());

crates/trie/trie/src/proof_v2/value.rs

@@ -1,9 +1,10 @@
 //! Generic value encoder types for proof calculation with lazy evaluation.
 
 use crate::{
-    hashed_cursor::HashedCursorFactory, proof_v2::ProofCalculator, trie_cursor::TrieCursorFactory,
+    hashed_cursor::HashedCursorFactory, prefix_set::PrefixSet, proof_v2::ProofCalculator,
+    trie_cursor::TrieCursorFactory,
 };
-use alloy_primitives::{B256, U256};
+use alloy_primitives::{map::B256Map, B256, U256};
 use alloy_rlp::Encodable;
 use reth_execution_errors::trie::StateProofError;
 use reth_primitives_traits::Account;
@@ -83,6 +84,8 @@ pub struct SyncAccountValueEncoder<T, H> {
     trie_cursor_factory: Rc<T>,
     /// Factory for creating hashed cursors.
     hashed_cursor_factory: Rc<H>,
+    /// Storage prefix sets keyed by hashed address.
+    storage_prefix_sets: Rc<B256Map<PrefixSet>>,
 }
 
 impl<T, H> SyncAccountValueEncoder<T, H> {
@@ -91,8 +94,17 @@ impl<T, H> SyncAccountValueEncoder<T, H> {
         Self {
             trie_cursor_factory: Rc::new(trie_cursor_factory),
             hashed_cursor_factory: Rc::new(hashed_cursor_factory),
+            storage_prefix_sets: Rc::new(B256Map::default()),
         }
     }
+
+    /// Sets the storage prefix sets. When given, all cached storage trie hashes matching the
+    /// prefix sets will be invalidated during storage root calculation for the corresponding
+    /// accounts.
+    pub fn with_storage_prefix_sets(mut self, storage_prefix_sets: B256Map<PrefixSet>) -> Self {
+        self.storage_prefix_sets = Rc::new(storage_prefix_sets);
+        self
+    }
 }
 
 /// The deferred encoder for an account value with synchronous storage root calculation.
@@ -100,6 +112,7 @@ impl<T, H> SyncAccountValueEncoder<T, H> {
 pub struct SyncAccountDeferredValueEncoder<T, H> {
     trie_cursor_factory: Rc<T>,
     hashed_cursor_factory: Rc<H>,
+    storage_prefix_sets: Rc<B256Map<PrefixSet>>,
     hashed_address: B256,
     account: Account,
 }
@@ -115,6 +128,9 @@ where
             self.hashed_cursor_factory.hashed_storage_cursor(self.hashed_address)?;
 
         let mut storage_proof_calculator = ProofCalculator::new_storage(trie_cursor, hashed_cursor);
+        if let Some(prefix_set) = self.storage_prefix_sets.get(&self.hashed_address) {
+            storage_proof_calculator = storage_proof_calculator.with_prefix_set(prefix_set.clone());
+        }
         let root_node = storage_proof_calculator.storage_root_node(self.hashed_address)?;
         let storage_root = storage_proof_calculator
             .compute_root_hash(&[root_node])?
@@ -143,8 +159,9 @@ where
         // Return a deferred encoder that will synchronously compute the storage root when encode()
         // is called.
         SyncAccountDeferredValueEncoder {
-            trie_cursor_factory: self.trie_cursor_factory.clone(),
-            hashed_cursor_factory: self.hashed_cursor_factory.clone(),
+            trie_cursor_factory: Rc::clone(&self.trie_cursor_factory),
+            hashed_cursor_factory: Rc::clone(&self.hashed_cursor_factory),
+            storage_prefix_sets: Rc::clone(&self.storage_prefix_sets),
             hashed_address,
             account,
         }