chore(ci): continue on error when codspeed is flakey/cancelled on main merge

Signed-off-by: Delweng <delweng@gmail.com>

.changelog/cool-suns-rest.md

@@ -0,0 +1,5 @@
+---
+reth-transaction-pool: minor
+---
+
+Added support for optional custom stateless and stateful validation hooks in `EthTransactionValidator` via `set_additional_stateless_validation` and `set_additional_stateful_validation` methods. Also implemented a manual `Debug` impl to handle the non-`Debug` function pointer fields.

.changelog/keen-geese-bake.md

@@ -0,0 +1,5 @@
+---
+reth-engine-tree: patch
+---
+
+Added sub-phase timing histograms to the sparse trie event loop, tracking channel wait, proof coalescing, multiproof reveal, and trie update durations separately.

.changelog/safe-waves-read.md

@@ -0,0 +1,7 @@
+---
+reth-network-types: minor
+reth-network: minor
+reth-node-core: patch
+---
+
+Added `PersistedPeerInfo` struct to persist richer peer metadata (kind, fork ID, reputation) to disk. Updated `PeersConfig::with_basic_nodes_from_file` to support both the new `PersistedPeerInfo` format and the legacy `Vec<NodeRecord>` format with automatic conversion, and updated `write_peers_to_file` to exclude backed-off and banned peers.

.changelog/tall-stars-shout.md

@@ -0,0 +1,5 @@
+---
+reth-network: minor
+---
+
+Added `fork_id` as a tiebreaker in peer selection when reputations are equal, preferring peers with a discovered `fork_id` as it indicates fork compatibility. Added a test to verify the tiebreaker behavior.

.changelog/tidy-stars-cry.md

@@ -0,0 +1,5 @@
+---
+reth-trie-sparse: patch
+---
+
+Fixed a bug where trie nodes could appear in both `updated_nodes` and `removed_nodes` simultaneously by removing entries from `removed_nodes` when a node is inserted as updated.

.github/CODEOWNERS

@@ -1,7 +1,7 @@
 *                           @gakonst
 crates/chain-state/         @fgimenez @mattsse
 crates/chainspec/           @Rjected @joshieDo @mattsse
-crates/cli/                 @mattsse
+crates/cli/                 @mattsse @Rjected
 crates/config/              @shekhirin @mattsse @Rjected 
 crates/consensus/           @mattsse @Rjected
 crates/e2e-test-utils/      @mattsse @Rjected @klkvr @fgimenez

.github/scripts/bench-reth-build.sh

@@ -14,25 +14,50 @@
 #   baseline: <source-dir>/target/profiling/reth
 #   feature:  <source-dir>/target/profiling/reth, reth-bench installed to cargo bin
 #
-# Required: mc (MinIO client) configured at /home/ubuntu/.mc
+# Required: mc (MinIO client) with a configured alias
 set -euo pipefail
 
-MC="mc --config-dir /home/ubuntu/.mc"
+MC="mc"
 MODE="$1"
 SOURCE_DIR="$2"
 COMMIT="$3"
 
+# Verify a cached reth binary was built from the expected commit.
+# `reth --version` outputs "Commit SHA: <full-sha>" on its own line.
+verify_binary() {
+  local binary="$1" expected_commit="$2"
+  local version binary_sha
+  version=$("$binary" --version 2>/dev/null) || return 1
+  binary_sha=$(echo "$version" | sed -n 's/^Commit SHA: *//p')
+  if [ -z "$binary_sha" ]; then
+    echo "Warning: could not extract commit SHA from version output"
+    return 1
+  fi
+  if [ "$binary_sha" = "$expected_commit" ]; then
+    return 0
+  fi
+  echo "Cache mismatch: binary built from ${binary_sha} but expected ${expected_commit}"
+  return 1
+}
+
 case "$MODE" in
   baseline|main)
     BUCKET="minio/reth-binaries/${COMMIT}"
     mkdir -p "${SOURCE_DIR}/target/profiling"
 
+    CACHE_VALID=false
     if $MC stat "${BUCKET}/reth" &>/dev/null; then
       echo "Cache hit for baseline (${COMMIT}), downloading binary..."
       $MC cp "${BUCKET}/reth" "${SOURCE_DIR}/target/profiling/reth"
       chmod +x "${SOURCE_DIR}/target/profiling/reth"
-    else
-      echo "Cache miss for baseline (${COMMIT}), building from source..."
+      if verify_binary "${SOURCE_DIR}/target/profiling/reth" "${COMMIT}"; then
+        CACHE_VALID=true
+      else
+        echo "Cached baseline binary is stale, rebuilding..."
+      fi
+    fi
+    if [ "$CACHE_VALID" = false ]; then
+      echo "Building baseline (${COMMIT}) from source..."
       cd "${SOURCE_DIR}"
       cargo build --profile profiling --bin reth
       $MC cp target/profiling/reth "${BUCKET}/reth"
@@ -43,14 +68,21 @@ case "$MODE" in
     BRANCH_SHA="${4:-$COMMIT}"
     BUCKET="minio/reth-binaries/${BRANCH_SHA}"
 
+    CACHE_VALID=false
     if $MC stat "${BUCKET}/reth" &>/dev/null && $MC stat "${BUCKET}/reth-bench" &>/dev/null; then
       echo "Cache hit for ${BRANCH_SHA}, downloading binaries..."
       mkdir -p "${SOURCE_DIR}/target/profiling"
       $MC cp "${BUCKET}/reth" "${SOURCE_DIR}/target/profiling/reth"
       $MC cp "${BUCKET}/reth-bench" /home/ubuntu/.cargo/bin/reth-bench
       chmod +x "${SOURCE_DIR}/target/profiling/reth" /home/ubuntu/.cargo/bin/reth-bench
-    else
-      echo "Cache miss for ${BRANCH_SHA}, building from source..."
+      if verify_binary "${SOURCE_DIR}/target/profiling/reth" "${COMMIT}"; then
+        CACHE_VALID=true
+      else
+        echo "Cached feature binary is stale, rebuilding..."
+      fi
+    fi
+    if [ "$CACHE_VALID" = false ]; then
+      echo "Building feature (${COMMIT}) from source..."
       cd "${SOURCE_DIR}"
       rustup show active-toolchain || rustup default stable
       make profiling

.github/scripts/bench-reth-charts.py

@@ -73,22 +73,24 @@ def plot_latency_and_throughput(
         for r in baseline:
             lat_s = r["new_payload_latency_us"] / 1_000_000
             base_ggas.append(r["gas_used"] / lat_s / GIGAGAS if lat_s > 0 else 0)
-        ax1.plot(base_x, base_lat, linewidth=0.8, label=baseline_name, alpha=0.7)
-        ax2.plot(base_x, base_ggas, linewidth=0.8, label=baseline_name, alpha=0.7)
+        l, = ax1.plot(base_x, base_lat, linewidth=0.8, label=baseline_name, alpha=0.7)
+        ax1.axhline(np.median(base_lat), color=l.get_color(), linestyle="--", linewidth=1, alpha=0.7, label=f"{baseline_name} median")
+        l, = ax2.plot(base_x, base_ggas, linewidth=0.8, label=baseline_name, alpha=0.7)
+        ax2.axhline(np.median(base_ggas), color=l.get_color(), linestyle="--", linewidth=1, alpha=0.7, label=f"{baseline_name} median")
 
-    ax1.plot(feat_x, feat_lat, linewidth=0.8, label=feature_name)
+    l, = ax1.plot(feat_x, feat_lat, linewidth=0.8, label=feature_name)
+    ax1.axhline(np.median(feat_lat), color=l.get_color(), linestyle="--", linewidth=1, label=f"{feature_name} median")
     ax1.set_ylabel("Latency (ms)")
     ax1.set_title("newPayload Latency per Block")
     ax1.grid(True, alpha=0.3)
-    if baseline:
-        ax1.legend()
+    ax1.legend()
 
-    ax2.plot(feat_x, feat_ggas, linewidth=0.8, label=feature_name)
+    l, = ax2.plot(feat_x, feat_ggas, linewidth=0.8, label=feature_name)
+    ax2.axhline(np.median(feat_ggas), color=l.get_color(), linestyle="--", linewidth=1, label=f"{feature_name} median")
     ax2.set_ylabel("Ggas/s")
     ax2.set_title("Execution Throughput per Block")
     ax2.grid(True, alpha=0.3)
-    if baseline:
-        ax2.legend()
+    ax2.legend()
 
     if baseline:
         ax3 = axes[2]

.github/scripts/bench-reth-run.sh

@@ -18,14 +18,35 @@ LOG="${OUTPUT_DIR}/node.log"
 cleanup() {
   kill "$TAIL_PID" 2>/dev/null || true
   if [ -n "${RETH_PID:-}" ] && sudo kill -0 "$RETH_PID" 2>/dev/null; then
-    sudo kill "$RETH_PID"
-    for i in $(seq 1 30); do
-      sudo kill -0 "$RETH_PID" 2>/dev/null || break
-      sleep 1
-    done
+    if [ "${BENCH_SAMPLY:-false}" = "true" ]; then
+      # Send SIGINT to the inner reth process by exact name (not -f which
+      # would also match samply's cmdline containing "reth"). Samply will
+      # capture reth's exit and save the profile.
+      sudo pkill -INT -x reth 2>/dev/null || true
+      # Wait for samply to finish writing the profile and exit
+      for i in $(seq 1 120); do
+        sudo pgrep -x samply > /dev/null 2>&1 || break
+        if [ $((i % 10)) -eq 0 ]; then
+          echo "Waiting for samply to finish writing profile... (${i}s)"
+        fi
+        sleep 1
+      done
+      if sudo pgrep -x samply > /dev/null 2>&1; then
+        echo "Samply still running after 120s, sending SIGTERM..."
+        sudo pkill -x samply 2>/dev/null || true
+      fi
+    else
+      sudo kill "$RETH_PID"
+      for i in $(seq 1 30); do
+        sudo kill -0 "$RETH_PID" 2>/dev/null || break
+        sleep 1
+      done
+    fi
     sudo kill -9 "$RETH_PID" 2>/dev/null || true
     sleep 1
   fi
+  # Fix ownership of reth-created files (reth runs as root)
+  sudo chown -R "$(id -un):$(id -gn)" "$OUTPUT_DIR" 2>/dev/null || true
   if mountpoint -q "$SCHELK_MOUNT"; then
     sudo umount -l "$SCHELK_MOUNT" || true
     sudo schelk recover -y || true
@@ -46,18 +67,38 @@ grep Cached /proc/meminfo
 # CPU layout: core 0 = OS/IRQs/reth-bench/aux, cores 1+ = reth node
 RETH_BENCH="$(which reth-bench)"
 ONLINE=$(nproc --all)
-RETH_CPUS="1-$(( ONLINE - 1 ))"
-sudo taskset -c "$RETH_CPUS" nice -n -20 "$BINARY" node \
-  --datadir "$DATADIR" \
-  --engine.accept-execution-requests-hash \
-  --http \
-  --http.port 8545 \
-  --ws \
-  --ws.api all \
-  --authrpc.port 8551 \
-  --disable-discovery \
-  --no-persist-peers \
-  > "$LOG" 2>&1 &
+MAX_RETH=$(( ONLINE - 1 ))
+if [ "${BENCH_CORES:-0}" -gt 0 ] && [ "$BENCH_CORES" -lt "$MAX_RETH" ]; then
+  MAX_RETH=$BENCH_CORES
+fi
+RETH_CPUS="1-${MAX_RETH}"
+
+RETH_ARGS=(
+  node
+  --datadir "$DATADIR"
+  --log.file.directory "$OUTPUT_DIR/reth-logs"
+  --engine.accept-execution-requests-hash
+  --http
+  --http.port 8545
+  --ws
+  --ws.api all
+  --authrpc.port 8551
+  --disable-discovery
+  --no-persist-peers
+)
+
+if [ "${BENCH_SAMPLY:-false}" = "true" ]; then
+  RETH_ARGS+=(--log.samply)
+  SAMPLY="$(which samply)"
+  sudo taskset -c "$RETH_CPUS" nice -n -20 \
+    "$SAMPLY" record --save-only --presymbolicate --rate 10000 \
+    --output "$OUTPUT_DIR/samply-profile.json.gz" \
+    -- "$BINARY" "${RETH_ARGS[@]}" \
+    > "$LOG" 2>&1 &
+else
+  sudo taskset -c "$RETH_CPUS" nice -n -20 "$BINARY" "${RETH_ARGS[@]}" \
+    > "$LOG" 2>&1 &
+fi
 
 RETH_PID=$!
 stdbuf -oL tail -f "$LOG" | sed -u "s/^/[reth] /" &

.github/scripts/bench-reth-snapshot.sh

@@ -116,10 +116,12 @@ wait $PROGRESS_PID 2>/dev/null || true
 update_comment "100"
 echo "Snapshot download complete"
 
-# Sync the new snapshot as the schelk baseline
+# Promote the new snapshot to become the schelk baseline (virgin volume).
+# This copies changed blocks from scratch → virgin so that future
+# `schelk recover` calls restore to this new state.
 sync
-sudo schelk recover -y
+sudo schelk promote -y
 
 # Save ETag marker
 echo "$REMOTE_ETAG" > "$ETAG_FILE"
-echo "Snapshot synced to schelk (ETag: ${REMOTE_ETAG})"
+echo "Snapshot promoted to schelk baseline (ETag: ${REMOTE_ETAG})"

.github/scripts/bench-reth-summary.py

@@ -243,13 +243,6 @@ def compute_paired_stats(
     }
 
 
-def compute_summary(combined: list[dict], gas: list[dict]) -> dict:
-    """Compute aggregate metrics from parsed CSV data."""
-    blocks = len(combined)
-    return {
-        "blocks": blocks,
-    }
-
 
 def format_duration(seconds: float) -> str:
     if seconds >= 60:
@@ -274,22 +267,56 @@ def fmt_mgas(v: float) -> str:
     return f"{v:.2f}"
 
 
+def significance(pct: float, ci_pct: float, lower_is_better: bool) -> str:
+    """Return significance label: 'good', 'bad', or 'neutral'."""
+    significant = abs(pct) > ci_pct
+    if not significant:
+        return "neutral"
+    elif (pct < 0) == lower_is_better:
+        return "good"
+    else:
+        return "bad"
+
+
 def change_str(pct: float, ci_pct: float, lower_is_better: bool) -> str:
     """Format change% with paired CI significance.
 
     Significant if the CI doesn't cross zero (i.e. |pct| > ci_pct).
     """
-    significant = abs(pct) > ci_pct
-    if not significant:
-        emoji = "⚪"
-    elif (pct < 0) == lower_is_better:
-        emoji = "✅"
-    else:
-        emoji = "❌"
-
+    sig = significance(pct, ci_pct, lower_is_better)
+    emoji = {"good": "✅", "bad": "❌", "neutral": "⚪"}[sig]
     return f"{pct:+.2f}% {emoji} (±{ci_pct:.2f}%)"
 
 
+def compute_changes(
+    baseline_stats: dict, feature_stats: dict, paired_stats: dict
+) -> dict:
+    """Pre-compute change percentages and significance for each metric."""
+    def pct(base: float, feat: float) -> float:
+        return (feat - base) / base * 100.0 if base > 0 else 0.0
+
+    def ci_pct(ci_ms: float, base_ms: float) -> float:
+        return ci_ms / base_ms * 100.0 if base_ms > 0 else 0.0
+
+    metrics = [
+        ("mean", "mean_ms", "ci_ms", "mean_ms", True),
+        ("p50", "p50_ms", "p50_ci_ms", "p50_ms", True),
+        ("p90", "p90_ms", "p90_ci_ms", "p90_ms", True),
+        ("p99", "p99_ms", "p99_ci_ms", "p99_ms", True),
+        ("mgas_s", "mean_mgas_s", "mgas_ci", "mean_mgas_s", False),
+    ]
+    changes = {}
+    for name, stat_key, ci_key, base_key, lower_is_better in metrics:
+        p = pct(baseline_stats[stat_key], feature_stats[stat_key])
+        c = ci_pct(paired_stats[ci_key], baseline_stats[base_key])
+        changes[name] = {
+            "pct": round(p, 4),
+            "ci_pct": round(c, 4),
+            "sig": significance(p, c, lower_is_better),
+        }
+    return changes
+
+
 def generate_comparison_table(
     run1: dict,
     run2: dict,
@@ -438,11 +465,6 @@ def main():
     all_baseline = [r for run in baseline_runs for r in run]
     all_feature = [r for run in feature_runs for r in run]
 
-    summary = compute_summary(all_feature, gas)
-    with open(args.output_summary, "w") as f:
-        json.dump(summary, f, indent=2)
-    print(f"Summary written to {args.output_summary}")
-
     baseline_stats = compute_stats(all_baseline)
     feature_stats = compute_stats(all_feature)
     paired_stats = compute_paired_stats(baseline_runs, feature_runs)
@@ -479,13 +501,40 @@ def main():
         ("execution_cache_wait_us", "Execution Cache Update Wait"),
     ]
     wait_time_tables = []
+    wait_time_data = {}
     for field, title in wait_fields:
         b_stats = compute_wait_stats(all_baseline, field)
         f_stats = compute_wait_stats(all_feature, field)
+        if b_stats and f_stats:
+            wait_time_data[field] = {
+                "title": title,
+                "baseline": b_stats,
+                "feature": f_stats,
+            }
         table = generate_wait_time_table(title, b_stats, f_stats, baseline_label, feature_label)
         if table:
             wait_time_tables.append(table)
 
+    summary = {
+        "blocks": paired_stats["blocks"],
+        "baseline": {
+            "name": baseline_name,
+            "ref": baseline_ref,
+            "stats": baseline_stats,
+        },
+        "feature": {
+            "name": feature_name,
+            "ref": feature_sha,
+            "stats": feature_stats,
+        },
+        "paired": paired_stats,
+        "changes": compute_changes(baseline_stats, feature_stats, paired_stats),
+        "wait_times": wait_time_data,
+    }
+    with open(args.output_summary, "w") as f:
+        json.dump(summary, f, indent=2)
+    print(f"Summary written to {args.output_summary}")
+
     markdown = generate_markdown(
         summary, comparison_table,
         wait_time_tables=wait_time_tables,

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

@@ -0,0 +1,342 @@
+// Sends Slack notifications for reth-bench results.
+//
+// Reads from environment:
+//   SLACK_BENCH_BOT_TOKEN  – Slack Bot User OAuth Token (xoxb-...)
+//   SLACK_BENCH_CHANNEL    – Public channel ID for significant improvements
+//   BENCH_WORK_DIR         – Directory containing summary.json
+//   BENCH_PR               – PR number (may be empty)
+//   BENCH_ACTOR            – GitHub user who triggered the bench
+//   BENCH_JOB_URL          – URL to the Actions job page
+//   BENCH_SAMPLY           – 'true' if samply profiling was enabled
+//
+// Usage from actions/github-script:
+//   const notify = require('./.github/scripts/bench-slack-notify.js');
+//   await notify.success({ core, context });
+//   await notify.failure({ core, context, failedStep: '...' });
+
+const fs = require('fs');
+const path = require('path');
+
+const SLACK_API = 'https://slack.com/api/chat.postMessage';
+
+function loadSlackUsers(repoRoot) {
+  try {
+    const raw = fs.readFileSync(path.join(repoRoot, '.github', 'scripts', 'bench-slack-users.json'), 'utf8');
+    const data = JSON.parse(raw);
+    // Filter out non-user-ID entries (like _comment)
+    const users = {};
+    for (const [k, v] of Object.entries(data)) {
+      if (!k.startsWith('_') && typeof v === 'string' && v.startsWith('U')) {
+        users[k] = v;
+      }
+    }
+    return users;
+  } catch {
+    return {};
+  }
+}
+
+async function postToSlack(token, channel, blocks, text, core, threadTs) {
+  const payload = { channel, blocks, text, unfurl_links: false };
+  if (threadTs) payload.thread_ts = threadTs;
+  const resp = await fetch(SLACK_API, {
+    method: 'POST',
+    headers: {
+      'Authorization': `Bearer ${token}`,
+      'Content-Type': 'application/json',
+    },
+    body: JSON.stringify(payload),
+  });
+  const data = await resp.json();
+  if (!data.ok) {
+    core.warning(`Slack API error (channel ${channel}): ${JSON.stringify(data)}`);
+  }
+  return data;
+}
+
+function cell(text) {
+  const s = String(text);
+  return { type: 'raw_text', text: s || ' ' };
+}
+
+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 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 prUrl = prNumber ? `https://github.com/${repo}/pull/${prNumber}` : '';
+  const commitUrl = `https://github.com/${repo}/commit`;
+  const baselineLink = `<${commitUrl}/${summary.baseline.ref}|${summary.baseline.name}>`;
+  const featureLink = `<${commitUrl}/${summary.feature.ref}|${summary.feature.name}>`;
+
+  // Meta line
+  const metaParts = [];
+  if (prNumber) metaParts.push(`*<${prUrl}|PR #${prNumber}>*`);
+  metaParts.push(`triggered by ${actorSlackId ? `<@${actorSlackId}>` : `@${actor}`}`);
+
+  // Baseline/feature lines with samply profile links
+  let baselineLine = `*Baseline:* ${baselineLink}`;
+  const bl1 = samplyUrls['baseline-1'];
+  const bl2 = samplyUrls['baseline-2'];
+  if (bl1) baselineLine += ` | <${bl1}|Samply 1>`;
+  if (bl2) baselineLine += ` | <${bl2}|Samply 2>`;
+
+  let featureLine = `*Feature:* ${featureLink}`;
+  const fl1 = samplyUrls['feature-1'];
+  const fl2 = samplyUrls['feature-2'];
+  if (fl1) featureLine += ` | <${fl1}|Samply 1>`;
+  if (fl2) featureLine += ` | <${fl2}|Samply 2>`;
+
+  const warmup = summary.warmup_blocks || process.env.BENCH_WARMUP_BLOCKS || '';
+  const cores = process.env.BENCH_CORES || '0';
+  const countsParts = [];
+  if (warmup) countsParts.push(`*Warmup:* ${warmup}`);
+  countsParts.push(`*Blocks:* ${summary.blocks}`);
+  if (cores !== '0') countsParts.push(`*Cores:* ${cores}`);
+  const countsLine = countsParts.join(' | ');
+
+  const sectionText = [metaParts.join(' | '), '', baselineLine, featureLine, countsLine].join('\n');
+
+  // Action buttons
+  const diffUrl = `https://github.com/${repo}/compare/${summary.baseline.ref}...${summary.feature.ref}`;
+  const buttons = [
+    {
+      type: 'button',
+      text: { type: 'plain_text', text: 'CI :github:', emoji: true },
+      url: jobUrl,
+      action_id: 'ci_button',
+    },
+    {
+      type: 'button',
+      text: { type: 'plain_text', text: 'Diff :github:', emoji: true },
+      url: diffUrl,
+      action_id: 'diff_button',
+    },
+  ];
+
+  const blocks = [
+    {
+      type: 'header',
+      text: { type: 'plain_text', text: `${headerEmoji} ${headerResult}`, emoji: true },
+    },
+    {
+      type: 'section',
+      text: { type: 'mrkdwn', text: sectionText },
+    },
+    {
+      type: 'table',
+      column_settings: [
+        { align: 'left' },
+        { align: 'right' },
+        { 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))],
+      ],
+    },
+    {
+      type: 'actions',
+      elements: buttons,
+    },
+  ];
+
+  // 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 = [
+      [cell('Wait Time'), cell('Baseline'), cell('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: [
+        { align: 'left' },
+        { align: 'right' },
+        { align: 'right' },
+      ],
+      rows: waitRows,
+    });
+  }
+
+  return { blocks, threadBlocks };
+}
+
+function buildFailureBlocks({ prNumber, actor, actorSlackId, jobUrl, repo, failedStep }) {
+  const prUrl = prNumber ? `https://github.com/${repo}/pull/${prNumber}` : '';
+  const actorMention = actorSlackId ? `<@${actorSlackId}>` : `@${actor}`;
+  const parts = [
+    prNumber ? `*<${prUrl}|PR #${prNumber}>*` : '',
+    `by ${actorMention}`,
+    `failed while *${failedStep}*`,
+  ].filter(Boolean);
+
+  const buttons = [
+    {
+      type: 'button',
+      text: { type: 'plain_text', text: 'CI :github:', emoji: true },
+      url: jobUrl,
+      action_id: 'ci_button',
+    },
+  ];
+
+  return [
+    {
+      type: 'header',
+      text: { type: 'plain_text', text: ':rotating_light: Bench Failed', emoji: true },
+    },
+    {
+      type: 'section',
+      text: { type: 'mrkdwn', text: parts.join(' | ') },
+    },
+    {
+      type: 'actions',
+      elements: buttons,
+    },
+  ];
+}
+
+async function success({ core, context }) {
+  const token = process.env.SLACK_BENCH_BOT_TOKEN;
+  if (!token) {
+    core.info('SLACK_BENCH_BOT_TOKEN not set, skipping Slack notification');
+    return;
+  }
+
+  let summary;
+  try {
+    summary = JSON.parse(fs.readFileSync(process.env.BENCH_WORK_DIR + '/summary.json', 'utf8'));
+  } catch (e) {
+    core.warning('Could not read summary.json for Slack notification');
+    return;
+  }
+
+  const repo = `${context.repo.owner}/${context.repo.repo}`;
+  const prNumber = process.env.BENCH_PR;
+  const actor = process.env.BENCH_ACTOR;
+  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 slackUsers = loadSlackUsers(process.env.GITHUB_WORKSPACE || '.');
+  const actorSlackId = slackUsers[actor];
+
+  const { blocks, threadBlocks } = buildSuccessBlocks({ summary, prNumber, actor, actorSlackId, jobUrl, repo, samplyUrls });
+  const text = `Bench: ${summary.baseline.name} vs ${summary.feature.name}`;
+
+  async function sendWithThread(ch) {
+    const res = await postToSlack(token, ch, blocks, text, core);
+    if (res.ok && res.ts && threadBlocks.length > 0) {
+      for (const tb of threadBlocks) {
+        await postToSlack(token, ch, [tb], 'Wait time breakdown', core, res.ts);
+      }
+    }
+  }
+
+  // Post to public channel if any metric shows significant improvement or regression
+  const channel = process.env.SLACK_BENCH_CHANNEL;
+  let postedToChannel = false;
+  if (channel) {
+    const changes = summary.changes || {};
+    const hasImprovement = Object.values(changes).some(c => c.sig === 'good');
+    if (hasImprovement) {
+      await sendWithThread(channel);
+      postedToChannel = true;
+    } else {
+      core.info('No significant improvement, skipping public channel notification');
+    }
+  }
+
+  // DM the actor only when results were not posted to the public channel
+  if (!postedToChannel) {
+    if (actorSlackId) {
+      await sendWithThread(actorSlackId);
+    } else {
+      core.info(`No Slack user mapping for GitHub user '${actor}', skipping DM`);
+    }
+  } else {
+    core.info(`Results posted to channel, skipping DM to ${actor}`);
+  }
+}
+
+async function failure({ core, context, failedStep }) {
+  const token = process.env.SLACK_BENCH_BOT_TOKEN;
+  if (!token) {
+    core.info('SLACK_BENCH_BOT_TOKEN not set, skipping Slack notification');
+    return;
+  }
+
+  const repo = `${context.repo.owner}/${context.repo.repo}`;
+  const prNumber = process.env.BENCH_PR;
+  const actor = process.env.BENCH_ACTOR;
+  const jobUrl = process.env.BENCH_JOB_URL ||
+    `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
+
+  const slackUsers = loadSlackUsers(process.env.GITHUB_WORKSPACE || '.');
+  const actorSlackId = slackUsers[actor];
+
+  const blocks = buildFailureBlocks({ prNumber, actor, actorSlackId, jobUrl, repo, failedStep });
+  const text = `Bench failed while ${failedStep}`;
+
+  // Always DM the actor
+  if (actorSlackId) {
+    await postToSlack(token, actorSlackId, blocks, text, core);
+  } else {
+    core.info(`No Slack user mapping for GitHub user '${actor}', skipping DM`);
+  }
+
+  // Only DM for failures, don't post to public channel
+}
+
+module.exports = { success, failure };

.github/scripts/bench-slack-users.json

@@ -0,0 +1,13 @@
+{
+  "_comment": "Maps GitHub usernames to Slack user IDs. Find yours: Slack profile > ··· > Copy member ID.",
+  "shekhirin": "U09FAL2UMLJ",
+  "mattsse": "U09FQNPMRT3",
+  "klkvr": "U09FAK95FC2",
+  "joshieDo": "U09LHN6GYAU",
+  "mediocregopher": "U09FF75KMQU",
+  "yongkangc": "U09FB0ECTD4",
+  "gakonst": "U092SEPDM40",
+  "Rjected": "U09F6SCKRGT",
+  "DaniPopes": "U09FAT8EK2A",
+  "emmajam": "U0A34UN92HW"
+}

.github/scripts/hive/ignored_tests.yaml

@@ -11,19 +11,14 @@
 #
 # When a test should no longer be ignored, remove it from this list.
 
-# flaky
-engine-withdrawals:
-  - Withdrawals Fork on Block 1 - 8 Block Re-Org NewPayload (Paris) (reth)
-  - Withdrawals Fork on Block 8 - 10 Block Re-Org NewPayload (Paris) (reth)
-  - Withdrawals Fork on Canonical Block 8 / Side Block 7 - 10 Block Re-Org (Paris) (reth)
-  - Sync after 2 blocks - Withdrawals on Block 2 - Multiple Withdrawal Accounts (Paris) (reth)
-  - Sync after 2 blocks - Withdrawals on Block 2 - Multiple Withdrawal Accounts - No Transactions (Paris) (reth)
-  - Sync after 128 blocks - Withdrawals on Block 2 - Multiple Withdrawal Accounts (Paris) (reth)
 engine-cancun:
   - Transaction Re-Org, New Payload on Revert Back (Cancun) (reth)
   - Transaction Re-Org, Re-Org to Different Block (Cancun) (reth)
   - Transaction Re-Org, Re-Org Out (Cancun) (reth)
   - Invalid Missing Ancestor ReOrg, StateRoot, EmptyTxs=False, Invalid P9 (Cancun) (reth)
+  # Hive test infra bug: geth sidecar switched to PathScheme for state storage, which has
+  # strict trie integrity requirements incompatible with inserting intentionally invalid blocks.
+  # Affects all clients, not just reth. Tracked: https://github.com/ethereum/hive/issues/1382
   - Invalid Missing Ancestor Syncing ReOrg, Timestamp, EmptyTxs=False, CanonicalReOrg=False, Invalid P8 (Cancun) (reth)
   - Invalid Missing Ancestor Syncing ReOrg, Timestamp, EmptyTxs=False, CanonicalReOrg=True, Invalid P8 (Cancun) (reth)
   - Multiple New Payloads Extending Canonical Chain, Wait for Canonical Payload (Cancun) (reth)

.github/workflows/bench.yml

@@ -35,22 +35,34 @@ on:
         required: false
         default: ""
         type: string
+      samply:
+        description: "Enable samply profiling"
+        required: false
+        default: "false"
+        type: boolean
+      cores:
+        description: "Limit reth to N CPU cores (0 = all available)"
+        required: false
+        default: "0"
+        type: string
 
 env:
   CARGO_TERM_COLOR: always
   BASELINE: base
   SEED: reth
   RUSTC_WRAPPER: "sccache"
+  BENCH_RUNNERS: 2
 
 name: bench
 
 permissions:
-  contents: write
+  contents: read
   pull-requests: write
 
 jobs:
   codspeed:
     if: github.event_name == 'push'
+    continue-on-error: true
     runs-on: depot-ubuntu-latest
     concurrency:
       group: bench-codspeed-${{ github.head_ref || github.run_id }}
@@ -91,7 +103,7 @@ jobs:
 
   reth-bench-ack:
     if: |
-      (github.event_name == 'issue_comment' && github.event.issue.pull_request && startsWith(github.event.comment.body, 'derek bench')) ||
+      (github.event_name == 'issue_comment' && github.event.issue.pull_request && (startsWith(github.event.comment.body, '@decofe bench') || startsWith(github.event.comment.body, 'derek bench'))) ||
       github.event_name == 'workflow_dispatch'
     name: reth-bench-ack
     runs-on: ubuntu-latest
@@ -104,6 +116,8 @@ jobs:
       feature: ${{ steps.args.outputs.feature }}
       baseline-name: ${{ steps.args.outputs.baseline-name }}
       feature-name: ${{ steps.args.outputs.feature-name }}
+      samply: ${{ steps.args.outputs.samply }}
+      cores: ${{ steps.args.outputs.cores }}
       comment-id: ${{ steps.ack.outputs.comment-id }}
     steps:
       - name: Check org membership
@@ -129,8 +143,9 @@ jobs:
         id: args
         uses: actions/github-script@v8
         with:
+          github-token: ${{ secrets.DEREK_PAT }}
           script: |
-            let pr, actor, blocks, warmup, baseline, feature;
+            let pr, actor, blocks, warmup, baseline, feature, samply, cores;
 
             if (context.eventName === 'workflow_dispatch') {
               actor = '${{ github.actor }}';
@@ -138,6 +153,8 @@ jobs:
               warmup = '${{ github.event.inputs.warmup }}' || '100';
               baseline = '${{ github.event.inputs.baseline }}';
               feature = '${{ github.event.inputs.feature }}';
+              samply = '${{ github.event.inputs.samply }}' === 'true' ? 'true' : 'false';
+              cores = '${{ github.event.inputs.cores }}' || '0';
 
               // Find PR for the selected branch
               const branch = '${{ github.ref_name }}';
@@ -157,16 +174,21 @@ jobs:
               actor = context.payload.comment.user.login;
 
               const body = context.payload.comment.body.trim();
-              const intArgs = new Set(['blocks', 'warmup']);
+              const intArgs = new Set(['blocks', 'warmup', 'cores']);
               const refArgs = new Set(['baseline', 'feature']);
-              const defaults = { blocks: '500', warmup: '100', baseline: '', feature: '' };
+              const boolArgs = new Set(['samply']);
+              const defaults = { blocks: '500', warmup: '100', baseline: '', feature: '', samply: 'false', cores: '0' };
               const unknown = [];
               const invalid = [];
-              const args = body.replace(/^derek bench\s*/, '');
+              const args = body.replace(/^(?:@decofe|derek) bench\s*/, '');
               for (const part of args.split(/\s+/).filter(Boolean)) {
                 const eq = part.indexOf('=');
                 if (eq === -1) {
-                  unknown.push(part);
+                  if (boolArgs.has(part)) {
+                    defaults[part] = 'true';
+                  } else {
+                    unknown.push(part);
+                  }
                   continue;
                 }
                 const key = part.slice(0, eq);
@@ -191,7 +213,7 @@ jobs:
               if (unknown.length) errors.push(`Unknown argument(s): \`${unknown.join('`, `')}\``);
               if (invalid.length) errors.push(`Invalid value(s): ${invalid.join(', ')}`);
               if (errors.length) {
-                const msg = `❌ **Invalid bench command**\n\n${errors.join('\n')}\n\n**Usage:** \`derek bench [blocks=N] [warmup=N] [baseline=REF] [feature=REF]\``;
+                const msg = `❌ **Invalid bench command**\n\n${errors.join('\n')}\n\n**Usage:** \`@decofe bench [blocks=N] [warmup=N] [baseline=REF] [feature=REF] [samply] [cores=N]\``;
                 await github.rest.issues.createComment({
                   owner: context.repo.owner,
                   repo: context.repo.repo,
@@ -205,6 +227,8 @@ jobs:
               warmup = defaults.warmup;
               baseline = defaults.baseline;
               feature = defaults.feature;
+              samply = defaults.samply;
+              cores = defaults.cores;
             }
 
             // Resolve display names for baseline/feature
@@ -231,11 +255,14 @@ jobs:
             core.setOutput('feature', feature);
             core.setOutput('baseline-name', baselineName);
             core.setOutput('feature-name', featureName);
+            core.setOutput('samply', samply);
+            core.setOutput('cores', cores);
 
       - name: Acknowledge request
         id: ack
         uses: actions/github-script@v8
         with:
+          github-token: ${{ secrets.DEREK_PAT }}
           script: |
             if (context.eventName === 'issue_comment') {
               await github.rest.reactions.createForIssueComment({
@@ -251,9 +278,11 @@ jobs:
 
             const runUrl = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
 
-            // Count queued/waiting bench runs ahead of this one
+            // Count queued/waiting bench runs ahead of this one.
+            // BENCH_RUNNERS is the number of self-hosted runners available.
             let queueMsg = '';
             let ahead = 0;
+            const numRunners = parseInt(process.env.BENCH_RUNNERS) || 1;
             try {
               const statuses = ['queued', 'in_progress', 'waiting', 'requested', 'pending'];
               const allRuns = [];
@@ -271,13 +300,13 @@ jobs:
               const benchRuns = allRuns.filter(r => r.event === 'issue_comment' || r.event === 'workflow_dispatch');
               const thisRun = benchRuns.find(r => r.id === context.runId);
               const thisCreatedAt = thisRun ? new Date(thisRun.created_at) : new Date();
-              ahead = benchRuns.filter(r => r.id !== context.runId && new Date(r.created_at) <= thisCreatedAt).length;
+              const totalAhead = benchRuns.filter(r => r.id !== context.runId && new Date(r.created_at) <= thisCreatedAt).length;
+              ahead = Math.max(0, totalAhead - numRunners + 1);
               if (ahead > 0) {
-                queueMsg = `\n🔢 **Queue position:** \`#${ahead + 1}\` (${ahead} job(s) ahead)`;
+                queueMsg = `\n🔢 **Queue position:** ${ahead} job(s) ahead (${numRunners} runner(s))`;
               }
             } catch (e) {
-              // Non-fatal — queue info is best-effort
-              core.info(`Could not fetch queue info: ${e.message}`);
+              core.info(`Skipping queue tracking: ${e.message}`);
             }
 
             const actor = '${{ steps.args.outputs.actor }}';
@@ -285,7 +314,11 @@ jobs:
             const warmup = '${{ steps.args.outputs.warmup }}';
             const baseline = '${{ steps.args.outputs.baseline-name }}';
             const feature = '${{ steps.args.outputs.feature-name }}';
-            const config = `**Config:** ${blocks} blocks, ${warmup} warmup blocks, baseline: \`${baseline}\`, feature: \`${feature}\``;
+            const samply = '${{ steps.args.outputs.samply }}' === 'true';
+            const samplyNote = samply ? ', samply: `enabled`' : '';
+            const cores = '${{ steps.args.outputs.cores }}';
+            const coresNote = cores && cores !== '0' ? `, cores: \`${cores}\`` : '';
+            const config = `**Config:** ${blocks} blocks, ${warmup} warmup blocks, baseline: \`${baseline}\`, feature: \`${feature}\`${samplyNote}${coresNote}`;
 
             const { data: comment } = await github.rest.issues.createComment({
               owner: context.repo.owner,
@@ -300,6 +333,7 @@ jobs:
         if: steps.ack.outputs.comment-id && steps.ack.outputs.queue-position != '0'
         uses: actions/github-script@v8
         with:
+          github-token: ${{ secrets.DEREK_PAT }}
           script: |
             const pr = '${{ steps.args.outputs.pr }}';
             const commentId = parseInt('${{ steps.ack.outputs.comment-id }}');
@@ -308,9 +342,14 @@ jobs:
             const warmup = '${{ steps.args.outputs.warmup }}';
             const baseline = '${{ steps.args.outputs.baseline-name }}';
             const feature = '${{ steps.args.outputs.feature-name }}';
-            const config = `**Config:** ${blocks} blocks, ${warmup} warmup blocks, baseline: \`${baseline}\`, feature: \`${feature}\``;
+            const samply = '${{ steps.args.outputs.samply }}' === 'true';
+            const samplyNote = samply ? ', samply: `enabled`' : '';
+            const cores = '${{ steps.args.outputs.cores }}';
+            const coresNote = cores && cores !== '0' ? `, cores: \`${cores}\`` : '';
+            const config = `**Config:** ${blocks} blocks, ${warmup} warmup blocks, baseline: \`${baseline}\`, feature: \`${feature}\`${samplyNote}${coresNote}`;
             const runUrl = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
 
+            const numRunners = parseInt(process.env.BENCH_RUNNERS) || 1;
             async function getQueuePosition() {
               const statuses = ['queued', 'in_progress', 'waiting', 'requested', 'pending'];
               const allRuns = [];
@@ -327,7 +366,8 @@ jobs:
               const benchRuns = allRuns.filter(r => r.event === 'issue_comment' || r.event === 'workflow_dispatch');
               const thisRun = benchRuns.find(r => r.id === context.runId);
               const thisCreatedAt = thisRun ? new Date(thisRun.created_at) : new Date();
-              return benchRuns.filter(r => r.id !== context.runId && new Date(r.created_at) <= thisCreatedAt).length;
+              const totalAhead = benchRuns.filter(r => r.id !== context.runId && new Date(r.created_at) <= thisCreatedAt).length;
+              return { ahead: Math.max(0, totalAhead - numRunners + 1), numRunners };
             }
 
             let lastPosition = parseInt('${{ steps.ack.outputs.queue-position }}');
@@ -336,11 +376,11 @@ jobs:
             while (true) {
               await sleep(10_000);
               try {
-                const ahead = await getQueuePosition();
+                const { ahead, numRunners } = await getQueuePosition();
                 if (ahead !== lastPosition) {
                   lastPosition = ahead;
                   const queueMsg = ahead > 0
-                    ? `\n🔢 **Queue position:** \`#${ahead + 1}\` (${ahead} job(s) ahead)`
+                    ? `\n🔢 **Queue position:** ${ahead} job(s) ahead (${numRunners} runner(s))`
                     : '';
                   await github.rest.issues.updateComment({
                     owner: context.repo.owner,
@@ -360,9 +400,6 @@ jobs:
     name: reth-bench
     runs-on: [self-hosted, Linux, X64]
     timeout-minutes: 120
-    concurrency:
-      group: reth-bench-queue
-      cancel-in-progress: false
     env:
       BENCH_RPC_URL: https://ethereum.reth.rs/rpc
       SCHELK_MOUNT: /reth-bench
@@ -371,6 +408,8 @@ jobs:
       BENCH_ACTOR: ${{ needs.reth-bench-ack.outputs.actor }}
       BENCH_BLOCKS: ${{ needs.reth-bench-ack.outputs.blocks }}
       BENCH_WARMUP_BLOCKS: ${{ needs.reth-bench-ack.outputs.warmup }}
+      BENCH_SAMPLY: ${{ needs.reth-bench-ack.outputs.samply }}
+      BENCH_CORES: ${{ needs.reth-bench-ack.outputs.cores }}
       BENCH_COMMENT_ID: ${{ needs.reth-bench-ack.outputs.comment-id }}
     steps:
       - name: Resolve checkout ref
@@ -405,6 +444,7 @@ jobs:
         if: env.BENCH_COMMENT_ID
         uses: actions/github-script@v8
         with:
+          github-token: ${{ secrets.DEREK_PAT }}
           script: |
             const { data: jobs } = await github.rest.actions.listJobsForWorkflowRun({
               owner: context.repo.owner,
@@ -419,7 +459,11 @@ jobs:
             const warmup = process.env.BENCH_WARMUP_BLOCKS;
             const baseline = '${{ needs.reth-bench-ack.outputs.baseline-name }}';
             const feature = '${{ needs.reth-bench-ack.outputs.feature-name }}';
-            core.exportVariable('BENCH_CONFIG', `**Config:** ${blocks} blocks, ${warmup} warmup blocks, baseline: \`${baseline}\`, feature: \`${feature}\``);
+            const samply = process.env.BENCH_SAMPLY === 'true';
+            const samplyNote = samply ? ', samply: `enabled`' : '';
+            const cores = process.env.BENCH_CORES || '0';
+            const coresNote = cores && cores !== '0' ? `, cores: \`${cores}\`` : '';
+            core.exportVariable('BENCH_CONFIG', `**Config:** ${blocks} blocks, ${warmup} warmup blocks, baseline: \`${baseline}\`, feature: \`${feature}\`${samplyNote}${coresNote}`);
 
             const { buildBody } = require('./.github/scripts/bench-update-status.js');
             await github.rest.issues.updateComment({
@@ -433,6 +477,52 @@ jobs:
       - uses: mozilla-actions/sccache-action@v0.0.9
         continue-on-error: true
 
+      - name: Install dependencies
+        env:
+          DEREK_TOKEN: ${{ secrets.DEREK_TOKEN }}
+        run: |
+          mkdir -p "$HOME/.local/bin"
+
+          # apt packages
+          sudo apt-get update -qq
+          sudo apt-get install -y --no-install-recommends \
+            python3 make jq zstd curl dmsetup \
+            linux-tools-"$(uname -r)" || \
+            sudo apt-get install -y --no-install-recommends linux-tools-generic
+
+          # mc (MinIO client)
+          if ! command -v mc &>/dev/null; then
+            curl -sSfL https://dl.min.io/client/mc/release/linux-amd64/mc -o "$HOME/.local/bin/mc"
+            chmod +x "$HOME/.local/bin/mc"
+          fi
+
+          # uv (Python package manager)
+          if ! command -v uv &>/dev/null; then
+            curl -LsSf https://astral.sh/uv/install.sh | env UV_INSTALL_DIR="$HOME/.local/bin" sh
+          fi
+
+          # Configure git auth for private repos
+          git config --global url."https://x-access-token:${DEREK_TOKEN}@github.com/".insteadOf "https://github.com/"
+
+          # thin-provisioning-tools (era_invalidate, required by schelk)
+          if ! command -v era_invalidate &>/dev/null; then
+            git clone --depth 1 https://github.com/jthornber/thin-provisioning-tools /tmp/tpt
+            sudo make -C /tmp/tpt install
+            rm -rf /tmp/tpt
+          fi
+
+          # schelk (snapshot rollback tool, invoked via sudo)
+          if ! sudo sh -c 'command -v schelk' &>/dev/null; then
+            cargo install --git https://github.com/tempoxyz/schelk --locked
+            sudo install "$HOME/.cargo/bin/schelk" /usr/local/bin/
+          fi
+
+          # samply (optional CPU profiler, invoked via sudo)
+          if [ "${BENCH_SAMPLY:-false}" = "true" ] && ! sudo sh -c 'command -v samply' &>/dev/null; then
+            cargo install samply --git https://github.com/DaniPopes/samply --branch edge --locked
+            sudo install "$HOME/.cargo/bin/samply" /usr/local/bin/
+          fi
+
       # Verify all required tools are available
       - name: Check dependencies
         run: |
@@ -470,32 +560,51 @@ jobs:
 
       - name: Resolve refs
         id: refs
-        run: |
-          BASELINE_ARG="${{ needs.reth-bench-ack.outputs.baseline }}"
-          FEATURE_ARG="${{ needs.reth-bench-ack.outputs.feature }}"
+        uses: actions/github-script@v8
+        with:
+          script: |
+            const { execSync } = require('child_process');
+            const run = (cmd) => execSync(cmd, { encoding: 'utf8' }).trim();
 
-          if [ -n "$BASELINE_ARG" ]; then
-            git fetch origin "$BASELINE_ARG" --quiet 2>/dev/null || true
-            BASELINE_REF=$(git rev-parse "$BASELINE_ARG" 2>/dev/null || git rev-parse "origin/$BASELINE_ARG" 2>/dev/null)
-            BASELINE_NAME="$BASELINE_ARG"
-          else
-            BASELINE_REF=$(git merge-base HEAD origin/main 2>/dev/null || echo "${{ github.sha }}")
-            BASELINE_NAME="main"
-          fi
+            const baselineArg = '${{ needs.reth-bench-ack.outputs.baseline }}';
+            const featureArg = '${{ needs.reth-bench-ack.outputs.feature }}';
 
-          if [ -n "$FEATURE_ARG" ]; then
-            git fetch origin "$FEATURE_ARG" --quiet 2>/dev/null || true
-            FEATURE_REF=$(git rev-parse "$FEATURE_ARG" 2>/dev/null || git rev-parse "origin/$FEATURE_ARG" 2>/dev/null)
-            FEATURE_NAME="$FEATURE_ARG"
-          else
-            FEATURE_REF="${{ steps.pr-info.outputs.head-sha }}"
-            FEATURE_NAME="${{ steps.pr-info.outputs.head-ref }}"
-          fi
+            let baselineRef, baselineName, featureRef, featureName;
 
-          echo "baseline-ref=$BASELINE_REF" >> "$GITHUB_OUTPUT"
-          echo "baseline-name=$BASELINE_NAME" >> "$GITHUB_OUTPUT"
-          echo "feature-ref=$FEATURE_REF" >> "$GITHUB_OUTPUT"
-          echo "feature-name=$FEATURE_NAME" >> "$GITHUB_OUTPUT"
+            if (baselineArg) {
+              try { run(`git fetch origin "${baselineArg}" --quiet`); } catch {}
+              try {
+                baselineRef = run(`git rev-parse "${baselineArg}"`);
+              } catch {
+                baselineRef = run(`git rev-parse "origin/${baselineArg}"`);
+              }
+              baselineName = baselineArg;
+            } else {
+              try {
+                baselineRef = run('git merge-base HEAD origin/main');
+              } catch {
+                baselineRef = '${{ github.sha }}';
+              }
+              baselineName = 'main';
+            }
+
+            if (featureArg) {
+              try { run(`git fetch origin "${featureArg}" --quiet`); } catch {}
+              try {
+                featureRef = run(`git rev-parse "${featureArg}"`);
+              } catch {
+                featureRef = run(`git rev-parse "origin/${featureArg}"`);
+              }
+              featureName = featureArg;
+            } else {
+              featureRef = '${{ steps.pr-info.outputs.head-sha }}';
+              featureName = '${{ steps.pr-info.outputs.head-ref }}';
+            }
+
+            core.setOutput('baseline-ref', baselineRef);
+            core.setOutput('baseline-name', baselineName);
+            core.setOutput('feature-ref', featureRef);
+            core.setOutput('feature-name', featureName);
 
       - name: Check if snapshot needs update
         id: snapshot-check
@@ -510,6 +619,7 @@ jobs:
         if: env.BENCH_COMMENT_ID && steps.snapshot-check.outputs.needed == 'true'
         uses: actions/github-script@v8
         with:
+          github-token: ${{ secrets.DEREK_PAT }}
           script: |
             const s = require('./.github/scripts/bench-update-status.js');
             await s({github, context, status: 'Building binaries & downloading snapshot...'});
@@ -523,7 +633,14 @@ jobs:
             git clone . ../reth-baseline
           fi
           git -C ../reth-baseline checkout "$BASELINE_REF"
-          ln -sfn "$(pwd)" ../reth-feature
+
+          FEATURE_REF="${{ steps.refs.outputs.feature-ref }}"
+          if [ -d ../reth-feature ]; then
+            git -C ../reth-feature fetch origin "$FEATURE_REF"
+          else
+            git clone . ../reth-feature
+          fi
+          git -C ../reth-feature checkout "$FEATURE_REF"
 
       - name: Build binaries and download snapshot in parallel
         id: build
@@ -610,6 +727,7 @@ jobs:
         if: success() && env.BENCH_COMMENT_ID
         uses: actions/github-script@v8
         with:
+          github-token: ${{ secrets.DEREK_PAT }}
           script: |
             const s = require('./.github/scripts/bench-update-status.js');
             await s({github, context, status: 'Running benchmarks...'});
@@ -632,6 +750,82 @@ jobs:
         id: run-baseline-2
         run: taskset -c 0 .github/scripts/bench-reth-run.sh baseline ../reth-baseline/target/profiling/reth "$BENCH_WORK_DIR/baseline-2"
 
+      - name: Scan logs for errors
+        if: "!cancelled()"
+        run: |
+          ERRORS_FILE="$BENCH_WORK_DIR/errors.md"
+          found=false
+          for run_dir in baseline-1 feature-1 feature-2 baseline-2; do
+            LOG="$BENCH_WORK_DIR/$run_dir/node.log"
+            if [ ! -f "$LOG" ]; then continue; fi
+
+            panics=$(grep -c -E 'panicked at' "$LOG" || true)
+            errors=$(grep -c ' ERROR ' "$LOG" || true)
+
+            if [ "$panics" -gt 0 ] || [ "$errors" -gt 0 ]; then
+              if [ "$found" = false ]; then
+                printf '### ⚠️ Node Errors\n\n' >> "$ERRORS_FILE"
+                found=true
+              fi
+              printf '<details><summary><b>%s</b>: %d panic(s), %d error(s)</summary>\n\n' "$run_dir" "$panics" "$errors" >> "$ERRORS_FILE"
+              if [ "$panics" -gt 0 ]; then
+                printf '**Panics:**\n```\n' >> "$ERRORS_FILE"
+                grep -E 'panicked at' "$LOG" | head -10 >> "$ERRORS_FILE"
+                printf '```\n' >> "$ERRORS_FILE"
+              fi
+              if [ "$errors" -gt 0 ]; then
+                printf '**Errors (first 20):**\n```\n' >> "$ERRORS_FILE"
+                grep ' ERROR ' "$LOG" | head -20 >> "$ERRORS_FILE"
+                printf '```\n' >> "$ERRORS_FILE"
+              fi
+              printf '\n</details>\n\n' >> "$ERRORS_FILE"
+            fi
+          done
+
+      - name: Upload samply profiles
+        if: success() && env.BENCH_SAMPLY == 'true'
+        run: |
+          PROFILER_API="https://api.profiler.firefox.com"
+          PROFILER_ACCEPT="Accept: application/vnd.firefox-profiler+json;version=1.0"
+
+          for run_dir in baseline-1 baseline-2 feature-1 feature-2; do
+            PROFILE="$BENCH_WORK_DIR/$run_dir/samply-profile.json.gz"
+            if [ ! -f "$PROFILE" ]; then continue; fi
+
+            PROFILE_SIZE=$(du -h "$PROFILE" | cut -f1)
+            echo "Uploading $run_dir samply profile (${PROFILE_SIZE}) to Firefox Profiler..."
+
+            # Upload compressed profile and get JWT back
+            JWT=$(curl -sf -X POST \
+              -H "Content-Type: application/octet-stream" \
+              -H "$PROFILER_ACCEPT" \
+              --data-binary "@$PROFILE" \
+              "$PROFILER_API/compressed-store") || {
+              echo "::warning::Failed to upload $run_dir profile to Firefox Profiler"
+              continue
+            }
+
+            # Extract profileToken from JWT payload (header.payload.signature)
+            PAYLOAD=$(echo "$JWT" | cut -d. -f2)
+            # Fix base64 padding
+            case $(( ${#PAYLOAD} % 4 )) in
+              2) PAYLOAD="${PAYLOAD}==" ;;
+              3) PAYLOAD="${PAYLOAD}=" ;;
+            esac
+            PROFILE_TOKEN=$(echo "$PAYLOAD" | base64 -d 2>/dev/null | python3 -c "import sys,json; print(json.load(sys.stdin)['profileToken'])")
+            PROFILE_URL="https://profiler.firefox.com/public/${PROFILE_TOKEN}"
+            echo "Profile uploaded: $PROFILE_URL"
+
+            # Shorten the URL
+            SHORT_URL=$(curl -sf -X POST \
+              -H "Content-Type: application/json" \
+              -H "$PROFILER_ACCEPT" \
+              -d "{\"longUrl\":\"$PROFILE_URL\"}" \
+              "$PROFILER_API/shorten" | python3 -c "import sys,json; print(json.load(sys.stdin)['shortUrl'])" 2>/dev/null) || SHORT_URL="$PROFILE_URL"
+            echo "$SHORT_URL" > "$BENCH_WORK_DIR/$run_dir/samply-profile-url.txt"
+            echo "Short profile URL for $run_dir: $SHORT_URL"
+          done
+
       # Results & charts
       - name: Parse results
         id: results
@@ -680,7 +874,7 @@ jobs:
           uv run --with matplotlib python3 .github/scripts/bench-reth-charts.py $CHART_ARGS
 
       - name: Upload results
-        if: success()
+        if: "!cancelled()"
         uses: actions/upload-artifact@v6
         with:
           name: bench-reth-results
@@ -688,31 +882,35 @@ jobs:
 
       - name: Push charts
         id: push-charts
-        if: success() && env.BENCH_PR
+        if: success()
         run: |
-          PR_NUMBER=${{ env.BENCH_PR }}
+          PR_NUMBER="${BENCH_PR:-0}"
           RUN_ID=${{ github.run_id }}
           CHART_DIR="pr/${PR_NUMBER}/${RUN_ID}"
+          CHARTS_REPO="https://x-access-token:${{ secrets.DEREK_TOKEN }}@github.com/decofe/reth-bench-charts.git"
 
-          if git fetch origin bench-charts 2>/dev/null; then
-            git checkout bench-charts
+          TMP_DIR=$(mktemp -d)
+          if git clone --depth 1 "${CHARTS_REPO}" "${TMP_DIR}" 2>/dev/null; then
+            true
           else
-            git checkout --orphan bench-charts
-            git rm -rf . 2>/dev/null || true
+            git init "${TMP_DIR}"
+            git -C "${TMP_DIR}" remote add origin "${CHARTS_REPO}"
           fi
 
-          mkdir -p "${CHART_DIR}"
-          cp "$BENCH_WORK_DIR"/charts/*.png "${CHART_DIR}/"
-          git add "${CHART_DIR}"
-          git -c user.name="github-actions" -c user.email="github-actions@github.com" \
+          mkdir -p "${TMP_DIR}/${CHART_DIR}"
+          cp "$BENCH_WORK_DIR"/charts/*.png "${TMP_DIR}/${CHART_DIR}/"
+          git -C "${TMP_DIR}" add "${CHART_DIR}"
+          git -C "${TMP_DIR}" -c user.name="github-actions" -c user.email="github-actions@github.com" \
             commit -m "bench charts for PR #${PR_NUMBER} run ${RUN_ID}"
-          git push origin bench-charts
-          echo "sha=$(git rev-parse HEAD)" >> "$GITHUB_OUTPUT"
+          git -C "${TMP_DIR}" push origin HEAD:main
+          echo "sha=$(git -C "${TMP_DIR}" rev-parse HEAD)" >> "$GITHUB_OUTPUT"
+          rm -rf "${TMP_DIR}"
 
       - name: Compare & comment
         if: success()
         uses: actions/github-script@v8
         with:
+          github-token: ${{ secrets.DEREK_PAT }}
           script: |
             const fs = require('fs');
 
@@ -724,11 +922,11 @@ jobs:
             }
 
             const sha = '${{ steps.push-charts.outputs.sha }}';
-            const prNumber = process.env.BENCH_PR;
+            const prNumber = process.env.BENCH_PR || '0';
             const runId = '${{ github.run_id }}';
 
-            if (sha && prNumber) {
-              const baseUrl = `https://raw.githubusercontent.com/${context.repo.owner}/${context.repo.repo}/${sha}/pr/${prNumber}/${runId}`;
+            if (sha) {
+              const baseUrl = `https://raw.githubusercontent.com/decofe/reth-bench-charts/${sha}/pr/${prNumber}/${runId}`;
               const charts = [
                 { file: 'latency_throughput.png', label: 'Latency, Throughput & Diff' },
                 { file: 'wait_breakdown.png', label: 'Wait Time Breakdown' },
@@ -743,6 +941,31 @@ jobs:
               comment += chartMarkdown;
             }
 
+            // Samply profile links (URLs point directly to Firefox Profiler)
+            if (process.env.BENCH_SAMPLY === 'true') {
+              const runs = ['baseline-1', 'feature-1', 'feature-2', 'baseline-2'];
+              const links = [];
+              for (const run of runs) {
+                try {
+                  const url = fs.readFileSync(`${process.env.BENCH_WORK_DIR}/${run}/samply-profile-url.txt`, 'utf8').trim();
+                  if (url) {
+                    links.push(`- **${run}**: [Firefox Profiler](${url})`);
+                  }
+                } catch (e) {}
+              }
+              if (links.length > 0) {
+                comment += `\n\n### Samply Profiles\n\n${links.join('\n')}\n`;
+              }
+            }
+
+            // Node errors (panics / ERROR logs)
+            try {
+              const errors = fs.readFileSync(process.env.BENCH_WORK_DIR + '/errors.md', 'utf8');
+              if (errors.trim()) {
+                comment += '\n\n' + errors;
+              }
+            } catch (e) {}
+
             const jobUrl = process.env.BENCH_JOB_URL || `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
             const body = `cc @${process.env.BENCH_ACTOR}\n\n✅ Benchmark complete! [View job](${jobUrl})\n\n${comment}`;
             const ackCommentId = process.env.BENCH_COMMENT_ID;
@@ -759,10 +982,22 @@ jobs:
               await core.summary.addRaw(body).write();
             }
 
+      - name: Send Slack notification (success)
+        if: success()
+        uses: actions/github-script@v8
+        env:
+          SLACK_BENCH_BOT_TOKEN: ${{ secrets.SLACK_BENCH_BOT_TOKEN }}
+          SLACK_BENCH_CHANNEL: ${{ secrets.SLACK_BENCH_CHANNEL }}
+        with:
+          script: |
+            const notify = require('./.github/scripts/bench-slack-notify.js');
+            await notify.success({ core, context });
+
       - name: Update status (failed)
         if: failure() && env.BENCH_COMMENT_ID
         uses: actions/github-script@v8
         with:
+          github-token: ${{ secrets.DEREK_PAT }}
           script: |
             const steps_status = [
               ['building binaries${{ steps.snapshot-check.outputs.needed == 'true' && ' & downloading snapshot' || '' }}', '${{ steps.build.outcome }}'],
@@ -774,19 +1009,54 @@ jobs:
             const failed = steps_status.find(([, o]) => o === 'failure');
             const failedStep = failed ? failed[0] : 'unknown step';
 
+            const fs = require('fs');
+            let errorDetails = '';
+            try {
+              const errors = fs.readFileSync(process.env.BENCH_WORK_DIR + '/errors.md', 'utf8');
+              if (errors.trim()) {
+                errorDetails = '\n\n' + errors;
+              }
+            } catch (e) {}
+
+            const jobUrl = process.env.BENCH_JOB_URL || `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
             await github.rest.issues.updateComment({
               owner: context.repo.owner, repo: context.repo.repo,
               comment_id: parseInt(process.env.BENCH_COMMENT_ID),
-              body: `cc @${process.env.BENCH_ACTOR}\n\n❌ Benchmark failed while ${failedStep}. [View logs](${process.env.BENCH_JOB_URL})`,
+              body: `cc @${process.env.BENCH_ACTOR}\n\n❌ Benchmark failed while ${failedStep}. [View logs](${jobUrl})${errorDetails}`,
             });
 
-      - name: Upload node log
+      - name: Send Slack notification (failure)
         if: failure()
-        uses: actions/upload-artifact@v6
+        uses: actions/github-script@v8
+        env:
+          SLACK_BENCH_BOT_TOKEN: ${{ secrets.SLACK_BENCH_BOT_TOKEN }}
+          SLACK_BENCH_CHANNEL: ${{ secrets.SLACK_BENCH_CHANNEL }}
         with:
-          name: reth-node-log
-          path: |
-            ${{ env.BENCH_WORK_DIR }}/*/node.log
+          script: |
+            const steps_status = [
+              ['building binaries${{ steps.snapshot-check.outputs.needed == 'true' && ' & downloading snapshot' || '' }}', '${{ steps.build.outcome }}'],
+              ['running baseline benchmark (1/2)', '${{ steps.run-baseline-1.outcome }}'],
+              ['running feature benchmark (1/2)', '${{ steps.run-feature-1.outcome }}'],
+              ['running feature benchmark (2/2)', '${{ steps.run-feature-2.outcome }}'],
+              ['running baseline benchmark (2/2)', '${{ steps.run-baseline-2.outcome }}'],
+            ];
+            const failed = steps_status.find(([, o]) => o === 'failure');
+            const failedStep = failed ? failed[0] : 'unknown step';
+            const notify = require('./.github/scripts/bench-slack-notify.js');
+            await notify.failure({ core, context, failedStep });
+
+      - name: Update status (cancelled)
+        if: cancelled() && env.BENCH_COMMENT_ID
+        uses: actions/github-script@v8
+        with:
+          github-token: ${{ secrets.DEREK_PAT }}
+          script: |
+            const jobUrl = process.env.BENCH_JOB_URL || `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
+            await github.rest.issues.updateComment({
+              owner: context.repo.owner, repo: context.repo.repo,
+              comment_id: parseInt(process.env.BENCH_COMMENT_ID),
+              body: `cc @${process.env.BENCH_ACTOR}\n\n⚠️ Benchmark cancelled. [View logs](${jobUrl})`,
+            });
 
       - name: Restore system settings
         if: always()

.github/workflows/release.yml

@@ -74,7 +74,7 @@ jobs:
             profile: maxperf
             allow_fail: false
           - target: aarch64-unknown-linux-gnu
-            os: ubuntu-24.04
+            os: ubuntu-24.04-arm
             profile: maxperf
             allow_fail: false
           - target: x86_64-apple-darwin
@@ -85,10 +85,6 @@ jobs:
             os: macos-14
             profile: maxperf
             allow_fail: false
-          - target: riscv64gc-unknown-linux-gnu
-            os: ubuntu-24.04
-            profile: maxperf
-            allow_fail: true
         build:
           - command: build
             binary: reth

CLAUDE.md

@@ -172,10 +172,97 @@ Before submitting changes, ensure:
 2. **Clippy**: No warnings
 3. **Tests Pass**: All unit and integration tests
 4. **Documentation**: Update relevant docs and add doc comments with `cargo docs --document-private-items`
-5. **Commit Messages**: Follow conventional format (feat:, fix:, chore:, etc.)
+5. **CLI Docs** (if CLI changed): Run `make update-book-cli` (see below)
+6. **Commit Messages**: Follow conventional format (feat:, fix:, chore:, etc.)
+
+### CLI Reference Docs (`book` CI Job)
+
+The CLI reference pages under `docs/vocs/docs/pages/cli/` are **auto-generated** from the `reth` binary's `--help` output. **Do not edit these files manually** — any hand edits will be overwritten and CI will fail regardless.
+
+When you add, remove, or modify CLI commands, subcommands, or flags, regenerate the CLI docs by running:
+
+```bash
+make update-book-cli
+```
+
+This builds `reth` in debug mode and runs `docs/cli/update.sh` to regenerate all CLI pages. Commit the resulting changes.
+
+The `book` CI job (`.github/workflows/lint.yml`) enforces this by regenerating the docs and running `git diff --exit-code`. If the committed docs don't match the generated output, CI fails. Manually editing these pages is never productive — always use `make update-book-cli`.
 
 ### Opening PRs against <https://github.com/paradigmxyz/reth>
 
+#### Titles
+
+Use [Conventional Commits](https://www.conventionalcommits.org/) with an optional scope:
+
+```
+<type>(<scope>): <short description>
+```
+
+**Types**: `feat`, `fix`, `perf`, `refactor`, `docs`, `test`, `chore`
+
+**Scope** (optional): crate or area, e.g. `evm`, `trie`, `rpc`, `engine`, `net`
+
+Examples:
+- `fix(rpc): correct gas estimation for ERC-20 transfers`
+- `perf: batch trie updates to reduce cursor overhead`
+- `feat(engine): add new_payload_interval metric`
+
+#### Descriptions
+
+Keep it short. Say what changed and why — nothing more.
+
+**Do:**
+- Write 1–3 sentences summarizing the change
+- Explain _why_ if the diff doesn't make it obvious
+- Link related issues or EIPs
+- Include benchmark numbers for perf changes
+
+**Don't:**
+- List every file changed — that's what the diff is for
+- Repeat the title in the body
+- Add "Files changed" or "Changes" sections
+- Write walls of text that go stale when the diff is updated
+- Use filler like "This PR introduces...", "comprehensive", "robust", "enhance", "leverage"
+
+**Template:**
+
+```
+Closes #<issue>
+
+<what changed, 1-3 sentences>
+
+<why, if not obvious from the diff>
+```
+
+**Good example:**
+
+```
+Closes #16800
+
+Adds fallback for external IP resolution so node startup doesn't fail
+when STUN is unreachable. Falls back to the configured default.
+```
+
+**Bad example:**
+
+```
+## Summary
+This PR introduces comprehensive improvements to the IP resolution system.
+
+## Changes
+- Modified `crates/net/discv4/src/lib.rs` to add fallback
+- Modified `crates/net/discv4/src/config.rs` to add default IP
+- Added tests in `crates/net/discv4/src/tests/ip.rs`
+
+## Files Changed
+- crates/net/discv4/src/lib.rs
+- crates/net/discv4/src/config.rs
+- crates/net/discv4/src/tests/ip.rs
+```
+
+#### Labels and CI
+
 Label PRs appropriately, first check the available labels and then apply the relevant ones:
 * when changes are RPC related, add A-rpc label
 * when changes are docs related, add C-docs label
@@ -455,5 +542,8 @@ cargo build --release
 cargo check --workspace --all-features
 
 # Check documentation
-cargo docs --document-private-items 
+cargo docs --document-private-items
+
+# Regenerate CLI reference docs (after CLI changes)
+make update-book-cli
 ```

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace.package]
-version = "1.11.0"
+version = "1.11.1"
 edition = "2024"
 rust-version = "1.93"
 license = "MIT OR Apache-2.0"
@@ -138,6 +138,7 @@ members = [
     "examples/exex-subscription",
     "examples/exex-test",
     "examples/full-contract-state",
+    "examples/migrate-trie-to-packed",
     "examples/manual-p2p/",
     "examples/network-txpool/",
     "examples/network/",
@@ -397,7 +398,7 @@ reth-payload-builder-primitives = { path = "crates/payload/builder-primitives" }
 reth-payload-primitives = { path = "crates/payload/primitives" }
 reth-payload-validator = { path = "crates/payload/validator" }
 reth-payload-util = { path = "crates/payload/util" }
-reth-primitives = { path = "crates/primitives", default-features = false }
+reth-primitives = { path = "crates/primitives", default-features = false, features = ["__internal"] }
 reth-primitives-traits = { path = "crates/primitives-traits", default-features = false }
 reth-provider = { path = "crates/storage/provider" }
 reth-prune = { path = "crates/prune/prune" }
@@ -532,13 +533,13 @@ quanta = "0.12"
 paste = "1.0"
 rand = "0.9"
 rayon = "1.7"
+thread-priority = "3.0.0"
 rustc-hash = { version = "2.0", default-features = false }
 schnellru = "0.2"
 serde = { version = "1.0", default-features = false }
 serde_json = { version = "1.0", default-features = false, features = ["alloc"] }
 serde_with = { version = "3", default-features = false, features = ["macros"] }
 sha2 = { version = "0.10", default-features = false }
-shellexpand = "3.0.0"
 shlex = "1.3"
 smallvec = "1"
 strum = { version = "0.27", default-features = false }

bin/reth-bench-compare/Cargo.toml

@@ -45,9 +45,6 @@ serde_json.workspace = true
 # Time handling
 chrono = { workspace = true, features = ["serde"] }
 
-# Path manipulation
-shellexpand.workspace = true
-
 # CSV handling
 csv.workspace = true
 

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

@@ -289,11 +289,7 @@ impl Args {
     /// Get the JWT secret path - either provided or derived from datadir
     pub(crate) fn jwt_secret_path(&self) -> PathBuf {
         match &self.jwt_secret {
-            Some(path) => {
-                let jwt_secret_str = path.to_string_lossy();
-                let expanded = shellexpand::tilde(&jwt_secret_str);
-                PathBuf::from(expanded.as_ref())
-            }
+            Some(path) => path.clone(),
             None => {
                 // Use the same logic as reth: <datadir>/<chain>/jwt.hex
                 let chain_path = self.datadir.clone().resolve_datadir(self.chain);
@@ -308,10 +304,9 @@ impl Args {
         chain_path.data_dir().to_path_buf()
     }
 
-    /// Get the expanded output directory path
+    /// Get the output directory path
     pub(crate) fn output_dir_path(&self) -> PathBuf {
-        let expanded = shellexpand::tilde(&self.output_dir);
-        PathBuf::from(expanded.as_ref())
+        PathBuf::from(&self.output_dir)
     }
 
     /// Get the effective warmup blocks value - either specified or defaults to blocks

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

@@ -31,6 +31,8 @@ pub(crate) struct BenchContext {
     pub(crate) is_optimism: bool,
     /// Whether to use `reth_newPayload` endpoint instead of `engine_newPayload*`.
     pub(crate) use_reth_namespace: bool,
+    /// Whether to fetch and replay RLP-encoded blocks.
+    pub(crate) rlp_blocks: bool,
 }
 
 impl BenchContext {
@@ -142,7 +144,8 @@ impl BenchContext {
         };
 
         let next_block = first_block.header.number + 1;
-        let use_reth_namespace = bench_args.reth_new_payload;
+        let rlp_blocks = bench_args.rlp_blocks;
+        let use_reth_namespace = bench_args.reth_new_payload || rlp_blocks;
         Ok(Self {
             auth_provider,
             block_provider,
@@ -150,6 +153,7 @@ impl BenchContext {
             next_block,
             is_optimism,
             use_reth_namespace,
+            rlp_blocks,
         })
     }
 }

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

@@ -6,7 +6,9 @@ use crate::{
         helpers::{build_payload, parse_gas_limit, prepare_payload_request, rpc_block_to_header},
         output::GasRampPayloadFile,
     },
-    valid_payload::{call_forkchoice_updated, call_new_payload_with_reth, payload_to_new_payload},
+    valid_payload::{
+        call_forkchoice_updated_with_reth, call_new_payload_with_reth, payload_to_new_payload,
+    },
 };
 use alloy_eips::BlockNumberOrTag;
 use alloy_provider::{network::AnyNetwork, Provider, RootProvider};
@@ -19,6 +21,7 @@ use reth_chainspec::ChainSpec;
 use reth_cli_runner::CliContext;
 use reth_ethereum_primitives::TransactionSigned;
 use reth_primitives_traits::constants::{GAS_LIMIT_BOUND_DIVISOR, MAXIMUM_GAS_LIMIT_BLOCK};
+use reth_rpc_api::RethNewPayloadInput;
 use std::{path::PathBuf, time::Instant};
 use tracing::info;
 
@@ -147,7 +150,7 @@ impl Command {
             }
         }
         if self.reth_new_payload {
-            info!("Using reth_newPayload endpoint");
+            info!("Using reth_newPayload and reth_forkchoiceUpdated endpoints");
         }
 
         let mut blocks_processed = 0u64;
@@ -182,28 +185,32 @@ impl Command {
                 Some(new_payload_version),
             )?;
 
+            let (version, params) = if self.reth_new_payload {
+                (None, serde_json::to_value((RethNewPayloadInput::ExecutionData(execution_data),))?)
+            } else {
+                (Some(version), params)
+            };
+
             // Save payload to file with version info for replay
             let payload_path =
                 self.output.join(format!("payload_block_{}.json", block.header.number));
             let file = GasRampPayloadFile {
-                version: version as u8,
+                version: version.map(|v| v as u8),
                 block_hash,
                 params: params.clone(),
-                execution_data: Some(execution_data.clone()),
             };
             let payload_json = serde_json::to_string_pretty(&file)?;
             std::fs::write(&payload_path, &payload_json)?;
             info!(target: "reth-bench", block_number = block.header.number, path = %payload_path.display(), "Saved payload");
 
-            let reth_data = self.reth_new_payload.then_some(execution_data);
-            let _ = call_new_payload_with_reth(&provider, version, params, reth_data).await?;
+            let _ = call_new_payload_with_reth(&provider, version, params).await?;
 
             let forkchoice_state = ForkchoiceState {
                 head_block_hash: block_hash,
                 safe_block_hash: block_hash,
                 finalized_block_hash: block_hash,
             };
-            call_forkchoice_updated(&provider, version, forkchoice_state, None).await?;
+            call_forkchoice_updated_with_reth(&provider, version, forkchoice_state).await?;
 
             parent_header = block.header;
             parent_hash = block_hash;

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

@@ -21,9 +21,11 @@ use crate::{
             derive_ws_rpc_url, setup_persistence_subscription, PersistenceWaiter,
         },
     },
-    valid_payload::{block_to_new_payload, call_forkchoice_updated, call_new_payload_with_reth},
+    valid_payload::{
+        block_to_new_payload, call_forkchoice_updated_with_reth, call_new_payload_with_reth,
+    },
 };
-use alloy_provider::Provider;
+use alloy_provider::{ext::DebugApi, Provider};
 use alloy_rpc_types_engine::ForkchoiceState;
 use clap::Parser;
 use eyre::{Context, OptionExt};
@@ -152,6 +154,7 @@ impl Command {
             mut next_block,
             is_optimism,
             use_reth_namespace,
+            rlp_blocks,
         } = BenchContext::new(&self.benchmark, self.rpc_url).await?;
 
         let total_blocks = benchmark_mode.total_blocks();
@@ -159,7 +162,7 @@ impl Command {
         let mut metrics_scraper = MetricsScraper::maybe_new(self.benchmark.metrics_url.clone());
 
         if use_reth_namespace {
-            info!("Using reth_newPayload endpoint");
+            info!("Using reth_newPayload and reth_forkchoiceUpdated endpoints");
         }
 
         let buffer_size = self.rpc_block_buffer_size;
@@ -184,6 +187,21 @@ impl Command {
                     }
                 };
 
+                let rlp = if rlp_blocks {
+                    let rlp = match block_provider.debug_get_raw_block(next_block.into()).await {
+                        Ok(rlp) => rlp,
+                        Err(e) => {
+                            tracing::error!(target: "reth-bench", "Failed to fetch raw block {next_block}: {e}");
+                            let _ = error_sender
+                                .send(eyre::eyre!("Failed to fetch raw block {next_block}: {e}"));
+                            break;
+                        }
+                    };
+                    Some(rlp)
+                } else {
+                    None
+                };
+
                 let head_block_hash = block.header.hash;
                 let safe_block_hash = block_provider
                     .get_block_by_number(block.header.number.saturating_sub(32).into());
@@ -205,7 +223,7 @@ impl Command {
 
                 next_block += 1;
                 if let Err(e) = sender
-                    .send((block, head_block_hash, safe_block_hash, finalized_block_hash))
+                    .send((block, head_block_hash, safe_block_hash, finalized_block_hash, rlp))
                     .await
                 {
                     tracing::error!(target: "reth-bench", "Failed to send block data: {e}");
@@ -219,7 +237,7 @@ impl Command {
         let total_benchmark_duration = Instant::now();
         let mut total_wait_time = Duration::ZERO;
 
-        while let Some((block, head, safe, finalized)) = {
+        while let Some((block, head, safe, finalized, rlp)) = {
             let wait_start = Instant::now();
             let result = receiver.recv().await;
             total_wait_time += wait_start.elapsed();
@@ -238,11 +256,11 @@ impl Command {
                 finalized_block_hash: finalized,
             };
 
-            let (version, params, execution_data) = block_to_new_payload(block, is_optimism)?;
+            let (version, params) =
+                block_to_new_payload(block, is_optimism, rlp, use_reth_namespace)?;
             let start = Instant::now();
-            let reth_data = use_reth_namespace.then_some(execution_data);
             let server_timings =
-                call_new_payload_with_reth(&auth_provider, version, params, reth_data).await?;
+                call_new_payload_with_reth(&auth_provider, version, params).await?;
 
             let np_latency =
                 server_timings.as_ref().map(|t| t.latency).unwrap_or_else(|| start.elapsed());
@@ -261,7 +279,7 @@ impl Command {
             };
 
             let fcu_start = Instant::now();
-            call_forkchoice_updated(&auth_provider, version, forkchoice_state, None).await?;
+            call_forkchoice_updated_with_reth(&auth_provider, version, forkchoice_state).await?;
             let fcu_latency = fcu_start.elapsed();
 
             let total_latency = if server_timings.is_some() {

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

@@ -11,7 +11,7 @@ use crate::{
     },
     valid_payload::{block_to_new_payload, call_new_payload_with_reth},
 };
-use alloy_provider::Provider;
+use alloy_provider::{ext::DebugApi, Provider};
 use clap::Parser;
 use csv::Writer;
 use eyre::{Context, OptionExt};
@@ -51,6 +51,7 @@ impl Command {
             mut next_block,
             is_optimism,
             use_reth_namespace,
+            rlp_blocks,
         } = BenchContext::new(&self.benchmark, self.rpc_url).await?;
 
         let total_blocks = benchmark_mode.total_blocks();
@@ -83,8 +84,21 @@ impl Command {
                     }
                 };
 
+                let rlp = if rlp_blocks {
+                    let Ok(rlp) = block_provider.debug_get_raw_block(next_block.into()).await
+                    else {
+                        tracing::error!(target: "reth-bench", "Failed to fetch raw block {next_block}");
+                        let _ = error_sender
+                            .send(eyre::eyre!("Failed to fetch raw block {next_block}"));
+                        break;
+                    };
+                    Some(rlp)
+                } else {
+                    None
+                };
+
                 next_block += 1;
-                if let Err(e) = sender.send(block).await {
+                if let Err(e) = sender.send((block, rlp)).await {
                     tracing::error!(target: "reth-bench", "Failed to send block data: {e}");
                     break;
                 }
@@ -96,7 +110,7 @@ impl Command {
         let total_benchmark_duration = Instant::now();
         let mut total_wait_time = Duration::ZERO;
 
-        while let Some(block) = {
+        while let Some((block, rlp)) = {
             let wait_start = Instant::now();
             let result = receiver.recv().await;
             total_wait_time += wait_start.elapsed();
@@ -108,12 +122,12 @@ impl Command {
 
             debug!(target: "reth-bench", number=?block.header.number, "Sending payload to engine");
 
-            let (version, params, execution_data) = block_to_new_payload(block, is_optimism)?;
+            let (version, params) =
+                block_to_new_payload(block, is_optimism, rlp, use_reth_namespace)?;
 
             let start = Instant::now();
-            let reth_data = use_reth_namespace.then_some(execution_data);
             let server_timings =
-                call_new_payload_with_reth(&auth_provider, version, params, reth_data).await?;
+                call_new_payload_with_reth(&auth_provider, version, params).await?;
 
             let latency =
                 server_timings.as_ref().map(|t| t.latency).unwrap_or_else(|| start.elapsed());

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

@@ -22,14 +22,14 @@ pub(crate) const NEW_PAYLOAD_OUTPUT_SUFFIX: &str = "new_payload_latency.csv";
 #[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct GasRampPayloadFile {
     /// Engine API version (1-5).
-    pub(crate) version: u8,
+    ///
+    /// `None` indicates that `reth_newPayload` should be used.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) version: Option<u8>,
     /// The block hash for FCU.
     pub(crate) block_hash: B256,
     /// The params to pass to newPayload.
     pub(crate) params: serde_json::Value,
-    /// The execution data for `reth_newPayload`.
-    #[serde(skip_serializing_if = "Option::is_none", default)]
-    pub(crate) execution_data: Option<alloy_rpc_types_engine::ExecutionData>,
 }
 
 /// This represents the results of a single `newPayload` call in the benchmark, containing the gas

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

@@ -24,10 +24,10 @@ use crate::{
             derive_ws_rpc_url, setup_persistence_subscription, PersistenceWaiter,
         },
     },
-    valid_payload::{call_forkchoice_updated, call_new_payload_with_reth},
+    valid_payload::{call_forkchoice_updated_with_reth, call_new_payload_with_reth},
 };
 use alloy_primitives::B256;
-use alloy_provider::{ext::EngineApi, network::AnyNetwork, Provider, RootProvider};
+use alloy_provider::{network::AnyNetwork, Provider, RootProvider};
 use alloy_rpc_client::ClientBuilder;
 use alloy_rpc_types_engine::{
     CancunPayloadFields, ExecutionData, ExecutionPayloadEnvelopeV4, ExecutionPayloadSidecar,
@@ -38,6 +38,7 @@ use eyre::Context;
 use reth_cli_runner::CliContext;
 use reth_engine_primitives::config::DEFAULT_PERSISTENCE_THRESHOLD;
 use reth_node_api::EngineApiMessageVersion;
+use reth_rpc_api::RethNewPayloadInput;
 use std::{
     path::PathBuf,
     time::{Duration, Instant},
@@ -161,7 +162,9 @@ struct GasRampPayload {
     /// Block number from filename.
     block_number: u64,
     /// Engine API version for newPayload.
-    version: EngineApiMessageVersion,
+    ///
+    /// `None` indicates that `reth_newPayload` should be used.
+    version: Option<EngineApiMessageVersion>,
     /// The file contents.
     file: GasRampPayloadFile,
 }
@@ -184,7 +187,7 @@ impl Command {
             );
         }
         if self.reth_new_payload {
-            info!("Using reth_newPayload endpoint");
+            info!("Using reth_newPayload and reth_forkchoiceUpdated endpoints");
         }
 
         // Set up waiter based on configured options
@@ -273,13 +276,10 @@ impl Command {
                 "Executing gas ramp payload (newPayload + FCU)"
             );
 
-            let reth_data =
-                if self.reth_new_payload { payload.file.execution_data.clone() } else { None };
             let _ = call_new_payload_with_reth(
                 &auth_provider,
                 payload.version,
                 payload.file.params.clone(),
-                reth_data,
             )
             .await?;
 
@@ -288,7 +288,7 @@ impl Command {
                 safe_block_hash: parent_hash,
                 finalized_block_hash: parent_hash,
             };
-            call_forkchoice_updated(&auth_provider, payload.version, fcu_state, None).await?;
+            call_forkchoice_updated_with_reth(&auth_provider, payload.version, fcu_state).await?;
 
             info!(target: "reth-bench", gas_ramp_payload = i + 1, "Gas ramp payload executed successfully");
 
@@ -336,31 +336,34 @@ impl Command {
                 "Sending newPayload"
             );
 
-            let params = serde_json::to_value((
-                execution_payload.clone(),
-                Vec::<B256>::new(),
-                B256::ZERO,
-                envelope.execution_requests.to_vec(),
-            ))?;
+            let (version, params) = if self.reth_new_payload {
+                let reth_data = ExecutionData {
+                    payload: execution_payload.clone().into(),
+                    sidecar: ExecutionPayloadSidecar::v4(
+                        CancunPayloadFields {
+                            versioned_hashes: Vec::new(),
+                            parent_beacon_block_root: B256::ZERO,
+                        },
+                        PraguePayloadFields {
+                            requests: envelope.execution_requests.clone().into(),
+                        },
+                    ),
+                };
+                (None, serde_json::to_value((RethNewPayloadInput::ExecutionData(reth_data),))?)
+            } else {
+                (
+                    Some(EngineApiMessageVersion::V4),
+                    serde_json::to_value((
+                        execution_payload.clone(),
+                        Vec::<B256>::new(),
+                        B256::ZERO,
+                        envelope.execution_requests.to_vec(),
+                    ))?,
+                )
+            };
 
-            let reth_data = self.reth_new_payload.then(|| ExecutionData {
-                payload: execution_payload.clone().into(),
-                sidecar: ExecutionPayloadSidecar::v4(
-                    CancunPayloadFields {
-                        versioned_hashes: Vec::new(),
-                        parent_beacon_block_root: B256::ZERO,
-                    },
-                    PraguePayloadFields { requests: envelope.execution_requests.clone().into() },
-                ),
-            });
-
-            let server_timings = call_new_payload_with_reth(
-                &auth_provider,
-                EngineApiMessageVersion::V4,
-                params,
-                reth_data,
-            )
-            .await?;
+            let server_timings =
+                call_new_payload_with_reth(&auth_provider, version, params).await?;
 
             let np_latency =
                 server_timings.as_ref().map(|t| t.latency).unwrap_or_else(|| start.elapsed());
@@ -384,10 +387,8 @@ impl Command {
                 finalized_block_hash: parent_hash,
             };
 
-            debug!(target: "reth-bench", method = "engine_forkchoiceUpdatedV3", ?fcu_state, "Sending forkchoiceUpdated");
-
             let fcu_start = Instant::now();
-            let fcu_result = auth_provider.fork_choice_updated_v3(fcu_state, None).await?;
+            call_forkchoice_updated_with_reth(&auth_provider, version, fcu_state).await?;
             let fcu_latency = fcu_start.elapsed();
 
             let total_latency =
@@ -420,7 +421,6 @@ impl Command {
                 TotalGasRow { block_number, transaction_count, gas_used, time: current_duration };
             results.push((gas_row, combined_result));
 
-            debug!(target: "reth-bench", ?fcu_result, "Payload executed successfully");
             parent_hash = block_hash;
         }
 
@@ -549,13 +549,18 @@ impl Command {
             let file: GasRampPayloadFile = serde_json::from_str(&content)
                 .wrap_err_with(|| format!("Failed to parse {:?}", path))?;
 
-            let version = match file.version {
-                1 => EngineApiMessageVersion::V1,
-                2 => EngineApiMessageVersion::V2,
-                3 => EngineApiMessageVersion::V3,
-                4 => EngineApiMessageVersion::V4,
-                5 => EngineApiMessageVersion::V5,
-                v => return Err(eyre::eyre!("Invalid version {} in {:?}", v, path)),
+            let version = if let Some(version) = file.version {
+                match version {
+                    1 => EngineApiMessageVersion::V1,
+                    2 => EngineApiMessageVersion::V2,
+                    3 => EngineApiMessageVersion::V3,
+                    4 => EngineApiMessageVersion::V4,
+                    5 => EngineApiMessageVersion::V5,
+                    v => return Err(eyre::eyre!("Invalid version {} in {:?}", v, path)),
+                }
+                .into()
+            } else {
+                None
             };
 
             info!(

bin/reth-bench/src/valid_payload.rs

@@ -3,7 +3,7 @@
 //! before sending additional calls.
 
 use alloy_eips::eip7685::Requests;
-use alloy_primitives::B256;
+use alloy_primitives::{Bytes, B256};
 use alloy_provider::{ext::EngineApi, network::AnyRpcBlock, Network, Provider};
 use alloy_rpc_types_engine::{
     ExecutionData, ExecutionPayload, ExecutionPayloadInputV2, ExecutionPayloadSidecar,
@@ -12,6 +12,7 @@ use alloy_rpc_types_engine::{
 use alloy_transport::TransportResult;
 use op_alloy_rpc_types_engine::OpExecutionPayloadV4;
 use reth_node_api::EngineApiMessageVersion;
+use reth_rpc_api::RethNewPayloadInput;
 use serde::Deserialize;
 use std::time::Duration;
 use tracing::{debug, error};
@@ -169,7 +170,15 @@ where
 pub(crate) fn block_to_new_payload(
     block: AnyRpcBlock,
     is_optimism: bool,
-) -> eyre::Result<(EngineApiMessageVersion, serde_json::Value, ExecutionData)> {
+    rlp: Option<Bytes>,
+    reth_new_payload: bool,
+) -> eyre::Result<(Option<EngineApiMessageVersion>, serde_json::Value)> {
+    if let Some(rlp) = rlp {
+        return Ok((
+            None,
+            serde_json::to_value((RethNewPayloadInput::<ExecutionData>::BlockRlp(rlp),))?,
+        ));
+    }
     let block = block
         .into_inner()
         .map_header(|header| header.map(|h| h.into_header_with_defaults()))
@@ -181,7 +190,14 @@ pub(crate) fn block_to_new_payload(
 
     // Convert to execution payload
     let (payload, sidecar) = ExecutionPayload::from_block_slow(&block);
-    payload_to_new_payload(payload, sidecar, is_optimism, block.withdrawals_root, None)
+    let (version, params, execution_data) =
+        payload_to_new_payload(payload, sidecar, is_optimism, block.withdrawals_root, None)?;
+
+    if reth_new_payload {
+        Ok((None, serde_json::to_value((RethNewPayloadInput::ExecutionData(execution_data),))?))
+    } else {
+        Ok((Some(version), params))
+    }
 }
 
 /// Converts an execution payload and sidecar into versioned engine API params and an
@@ -266,17 +282,15 @@ pub(crate) fn payload_to_new_payload(
 #[allow(dead_code)]
 pub(crate) async fn call_new_payload<N: Network, P: Provider<N>>(
     provider: P,
-    version: EngineApiMessageVersion,
+    version: Option<EngineApiMessageVersion>,
     params: serde_json::Value,
-) -> TransportResult<Option<NewPayloadTimingBreakdown>> {
-    call_new_payload_with_reth(provider, version, params, None).await
+) -> eyre::Result<Option<NewPayloadTimingBreakdown>> {
+    call_new_payload_with_reth(provider, version, params).await
 }
 
 /// Response from `reth_newPayload` endpoint, which includes server-measured latency.
 #[derive(Debug, Deserialize)]
 struct RethPayloadStatus {
-    #[serde(flatten)]
-    status: PayloadStatus,
     latency_us: u64,
     #[serde(default)]
     persistence_wait_us: Option<u64>,
@@ -300,72 +314,50 @@ pub(crate) struct NewPayloadTimingBreakdown {
 }
 
 /// Calls either `engine_newPayload*` or `reth_newPayload` depending on whether
-/// `reth_execution_data` is provided.
+/// `version` is provided.
 ///
-/// When `reth_execution_data` is `Some`, uses the `reth_newPayload` endpoint which takes
-/// `ExecutionData` directly and waits for persistence and cache updates to complete.
+/// When `version` is `None`, uses `reth_newPayload` endpoint with provided params.
 ///
 /// Returns the server-reported timing breakdown when using the reth namespace, or `None` for
 /// the standard engine namespace.
 pub(crate) async fn call_new_payload_with_reth<N: Network, P: Provider<N>>(
     provider: P,
-    version: EngineApiMessageVersion,
+    version: Option<EngineApiMessageVersion>,
     params: serde_json::Value,
-    reth_execution_data: Option<ExecutionData>,
-) -> TransportResult<Option<NewPayloadTimingBreakdown>> {
-    if let Some(execution_data) = reth_execution_data {
-        let method = "reth_newPayload";
-        let reth_params = serde_json::to_value((execution_data.clone(),))
-            .expect("ExecutionData serialization cannot fail");
+) -> eyre::Result<Option<NewPayloadTimingBreakdown>> {
+    let method = version.map(|v| v.method_name()).unwrap_or("reth_newPayload");
 
-        debug!(target: "reth-bench", method, "Sending newPayload");
+    debug!(target: "reth-bench", method, "Sending newPayload");
 
-        let mut resp: RethPayloadStatus = provider.client().request(method, &reth_params).await?;
+    let resp = loop {
+        let resp: serde_json::Value = provider.client().request(method, &params).await?;
+        let status = PayloadStatus::deserialize(&resp)?;
 
-        while !resp.status.is_valid() {
-            if resp.status.is_invalid() {
-                error!(target: "reth-bench", status=?resp.status, "Invalid {method}");
-                return Err(alloy_json_rpc::RpcError::LocalUsageError(Box::new(
-                    std::io::Error::other(format!("Invalid {method}: {:?}", resp.status)),
-                )))
-            }
-            if resp.status.is_syncing() {
-                return Err(alloy_json_rpc::RpcError::UnsupportedFeature(
-                    "invalid range: no canonical state found for parent of requested block",
-                ))
-            }
-            resp = provider.client().request(method, &reth_params).await?;
+        if status.is_valid() {
+            break resp;
         }
-
-        Ok(Some(NewPayloadTimingBreakdown {
-            latency: Duration::from_micros(resp.latency_us),
-            persistence_wait: resp.persistence_wait_us.map(Duration::from_micros),
-            execution_cache_wait: Duration::from_micros(resp.execution_cache_wait_us),
-            sparse_trie_wait: Duration::from_micros(resp.sparse_trie_wait_us),
-        }))
-    } else {
-        let method = version.method_name();
-
-        debug!(target: "reth-bench", method, "Sending newPayload");
-
-        let mut status: PayloadStatus = provider.client().request(method, &params).await?;
-
-        while !status.is_valid() {
-            if status.is_invalid() {
-                error!(target: "reth-bench", ?status, ?params, "Invalid {method}",);
-                return Err(alloy_json_rpc::RpcError::LocalUsageError(Box::new(
-                    std::io::Error::other(format!("Invalid {method}: {status:?}")),
-                )))
-            }
-            if status.is_syncing() {
-                return Err(alloy_json_rpc::RpcError::UnsupportedFeature(
-                    "invalid range: no canonical state found for parent of requested block",
-                ))
-            }
-            status = provider.client().request(method, &params).await?;
+        if status.is_invalid() {
+            return Err(eyre::eyre!("Invalid {method}: {status:?}"));
         }
-        Ok(None)
+        if status.is_syncing() {
+            return Err(eyre::eyre!(
+                "invalid range: no canonical state found for parent of requested block"
+            ));
+        }
+    };
+
+    if version.is_some() {
+        return Ok(None);
     }
+
+    let resp: RethPayloadStatus = serde_json::from_value(resp)?;
+
+    Ok(Some(NewPayloadTimingBreakdown {
+        latency: Duration::from_micros(resp.latency_us),
+        persistence_wait: resp.persistence_wait_us.map(Duration::from_micros),
+        execution_cache_wait: Duration::from_micros(resp.execution_cache_wait_us),
+        sparse_trie_wait: Duration::from_micros(resp.sparse_trie_wait_us),
+    }))
 }
 
 /// Calls the correct `engine_forkchoiceUpdated` method depending on the given
@@ -392,3 +384,47 @@ pub(crate) async fn call_forkchoice_updated<N, P: EngineApiValidWaitExt<N>>(
         }
     }
 }
+
+/// Calls either `reth_forkchoiceUpdated` or the standard `engine_forkchoiceUpdated*` depending
+/// on `use_reth`.
+///
+/// When `use_reth` is true, uses the `reth_forkchoiceUpdated` endpoint which sends a regular FCU
+/// with no payload attributes.
+pub(crate) async fn call_forkchoice_updated_with_reth<
+    N: Network,
+    P: Provider<N> + EngineApiValidWaitExt<N>,
+>(
+    provider: P,
+    message_version: Option<EngineApiMessageVersion>,
+    forkchoice_state: ForkchoiceState,
+) -> TransportResult<ForkchoiceUpdated> {
+    if let Some(message_version) = message_version {
+        call_forkchoice_updated(provider, message_version, forkchoice_state, None).await
+    } else {
+        let method = "reth_forkchoiceUpdated";
+        let reth_params = serde_json::to_value((forkchoice_state,))
+            .expect("ForkchoiceState serialization cannot fail");
+
+        debug!(target: "reth-bench", method, "Sending forkchoiceUpdated");
+
+        loop {
+            let resp: ForkchoiceUpdated = provider.client().request(method, &reth_params).await?;
+
+            if resp.is_valid() {
+                break Ok(resp)
+            }
+
+            if resp.is_invalid() {
+                error!(target: "reth-bench", ?resp, "Invalid {method}");
+                return Err(alloy_json_rpc::RpcError::LocalUsageError(Box::new(
+                    std::io::Error::other(format!("Invalid {method}: {resp:?}")),
+                )))
+            }
+            if resp.is_syncing() {
+                return Err(alloy_json_rpc::RpcError::UnsupportedFeature(
+                    "invalid range: no canonical state found for parent of requested block",
+                ))
+            }
+        }
+    }
+}

bin/reth/Cargo.toml

@@ -30,7 +30,8 @@ workspace = true
 # reth
 reth-ethereum-cli.workspace = true
 reth-chainspec.workspace = true
-reth-primitives.workspace = true
+reth-primitives-traits.workspace = true
+reth-ethereum-primitives.workspace = true
 reth-db = { workspace = true, features = ["mdbx"] }
 reth-provider.workspace = true
 reth-revm.workspace = true
@@ -110,7 +111,6 @@ dev = ["reth-ethereum-cli/dev"]
 
 asm-keccak = [
     "reth-node-core/asm-keccak",
-    "reth-primitives/asm-keccak",
     "reth-ethereum-cli/asm-keccak",
     "reth-node-ethereum/asm-keccak",
     "alloy-primitives/asm-keccak",

bin/reth/src/lib.rs

@@ -124,9 +124,11 @@ pub mod providers {
     pub use reth_provider::*;
 }
 
-/// Re-exported from `reth_primitives`.
+/// Re-exported primitives.
+#[allow(ambiguous_glob_reexports)]
 pub mod primitives {
-    pub use reth_primitives::*;
+    pub use reth_ethereum_primitives::*;
+    pub use reth_primitives_traits::*;
 }
 
 /// Re-exported from `reth_ethereum_consensus`.

crates/chain-state/src/in_memory.rs

@@ -1061,14 +1061,6 @@ mod tests {
         ) -> ProviderResult<Option<StorageValue>> {
             Ok(None)
         }
-
-        fn storage_by_hashed_key(
-            &self,
-            _address: Address,
-            _hashed_storage_key: StorageKey,
-        ) -> ProviderResult<Option<StorageValue>> {
-            Ok(None)
-        }
     }
 
     impl BytecodeReader for MockStateProvider {

crates/chain-state/src/memory_overlay.rs

@@ -223,26 +223,6 @@ impl<N: NodePrimitives> StateProvider for MemoryOverlayStateProviderRef<'_, N> {
 
         self.historical.storage(address, storage_key)
     }
-
-    fn storage_by_hashed_key(
-        &self,
-        address: Address,
-        hashed_storage_key: StorageKey,
-    ) -> ProviderResult<Option<StorageValue>> {
-        let hashed_address = keccak256(address);
-        let state = &self.trie_input().state;
-
-        if let Some(hs) = state.storages.get(&hashed_address) {
-            if let Some(value) = hs.storage.get(&hashed_storage_key) {
-                return Ok(Some(*value));
-            }
-            if hs.wiped {
-                return Ok(Some(StorageValue::ZERO));
-            }
-        }
-
-        self.historical.storage_by_hashed_key(address, hashed_storage_key)
-    }
 }
 
 impl<N: NodePrimitives> BytecodeReader for MemoryOverlayStateProviderRef<'_, N> {

crates/chainspec/src/api.rs

@@ -132,6 +132,6 @@ impl<H: BlockHeader> EthChainSpec for ChainSpec<H> {
     }
 
     fn final_paris_total_difficulty(&self) -> Option<U256> {
-        self.paris_block_and_final_difficulty.map(|(_, final_difficulty)| final_difficulty)
+        self.get_final_paris_total_difficulty()
     }
 }

crates/chainspec/src/spec.rs

@@ -855,15 +855,9 @@ impl From<Genesis> for ChainSpec {
                             // those networks we use the activation
                             // blocks of those networks
                             match genesis.config.chain_id {
-                                1 => {
-                                    if ttd == MAINNET_PARIS_TTD {
-                                        return Some(MAINNET_PARIS_BLOCK)
-                                    }
-                                }
-                                11155111 => {
-                                    if ttd == SEPOLIA_PARIS_TTD {
-                                        return Some(SEPOLIA_PARIS_BLOCK)
-                                    }
+                                1 if ttd == MAINNET_PARIS_TTD => return Some(MAINNET_PARIS_BLOCK),
+                                11155111 if ttd == SEPOLIA_PARIS_TTD => {
+                                    return Some(SEPOLIA_PARIS_BLOCK)
                                 }
                                 _ => {}
                             };

crates/cli/cli/Cargo.toml

@@ -18,6 +18,5 @@ alloy-genesis.workspace = true
 
 # misc
 clap.workspace = true
-shellexpand.workspace = true
 eyre.workspace = true
 serde_json.workspace = true

crates/cli/cli/src/chainspec.rs

@@ -73,7 +73,7 @@ pub trait ChainSpecParser: Clone + Send + Sync + 'static {
 /// A helper to parse a [`Genesis`](alloy_genesis::Genesis) as argument or from disk.
 pub fn parse_genesis(s: &str) -> eyre::Result<alloy_genesis::Genesis> {
     // try to read json from path first
-    let raw = match fs::read_to_string(PathBuf::from(shellexpand::full(s)?.into_owned())) {
+    let raw = match fs::read_to_string(PathBuf::from(s)) {
         Ok(raw) => raw,
         Err(io_err) => {
             // valid json may start with "\n", but must contain "{"

crates/cli/commands/Cargo.toml

@@ -53,7 +53,7 @@ reth-tasks.workspace = true
 reth-storage-api.workspace = true
 reth-trie = { workspace = true, features = ["metrics"] }
 reth-trie-db = { workspace = true, features = ["metrics"] }
-reth-trie-common.workspace = true
+reth-trie-common = { workspace = true, optional = true }
 reth-primitives-traits.workspace = true
 reth-discv4.workspace = true
 reth-discv5.workspace = true
@@ -113,6 +113,7 @@ arbitrary = [
     "dep:proptest",
     "dep:arbitrary",
     "dep:proptest-arbitrary-interop",
+    "dep:reth-trie-common",
     "reth-db-api/arbitrary",
     "reth-eth-wire/arbitrary",
     "reth-db/arbitrary",
@@ -123,11 +124,11 @@ arbitrary = [
     "reth-codecs/test-utils",
     "reth-prune-types/test-utils",
     "reth-stages-types/test-utils",
-    "reth-trie-common/test-utils",
+    "reth-trie-common?/test-utils",
     "reth-codecs/arbitrary",
     "reth-prune-types/arbitrary",
     "reth-stages-types?/arbitrary",
-    "reth-trie-common/arbitrary",
+    "reth-trie-common?/arbitrary",
     "alloy-consensus/arbitrary",
     "reth-primitives-traits/arbitrary",
     "reth-ethereum-primitives/arbitrary",

crates/cli/commands/src/common.rs

@@ -146,11 +146,24 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
                 })
             }
         };
-        let rocksdb_provider = RocksDBProvider::builder(data_dir.rocksdb())
-            .with_default_tables()
-            .with_database_log_level(self.db.log_level)
-            .with_read_only(!access.is_read_write())
-            .build()?;
+        let rocksdb_provider = if !access.is_read_write() && !RocksDBProvider::exists(&rocksdb_path)
+        {
+            // RocksDB database doesn't exist yet (e.g. datadir restored from a snapshot
+            // or created before RocksDB storage). Create an empty one so read-only
+            // commands can proceed.
+            debug!(target: "reth::cli", ?rocksdb_path, "RocksDB not found, initializing empty database");
+            reth_fs_util::create_dir_all(&rocksdb_path)?;
+            RocksDBProvider::builder(data_dir.rocksdb())
+                .with_default_tables()
+                .with_database_log_level(self.db.log_level)
+                .build()?
+        } else {
+            RocksDBProvider::builder(data_dir.rocksdb())
+                .with_default_tables()
+                .with_database_log_level(self.db.log_level)
+                .with_read_only(!access.is_read_write())
+                .build()?
+        };
 
         let provider_factory =
             self.create_provider_factory(&config, db, sfp, rocksdb_provider, access, runtime)?;

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

@@ -14,7 +14,7 @@ use reth_db_api::{
 use reth_db_common::DbTool;
 use reth_node_builder::{NodeTypesWithDB, NodeTypesWithDBAdapter};
 use reth_provider::{providers::ProviderNodeTypes, DBProvider, StaticFileProviderFactory};
-use reth_static_file_types::StaticFileSegment;
+use reth_static_file_types::{ChangesetOffset, StaticFileSegment};
 use std::{
     hash::{BuildHasher, Hasher},
     time::{Duration, Instant},
@@ -134,12 +134,12 @@ fn checksum_static_file<N: CliNodeTypes<ChainSpec: EthereumHardforks>>(
         .ok_or_else(|| eyre::eyre!("No static files found for segment: {}", segment))?;
 
     let start_time = Instant::now();
-    let mut hasher = checksum_hasher();
-    let mut total = 0usize;
     let limit = limit.unwrap_or(usize::MAX);
+    let mut checksummer = Checksummer::new(checksum_hasher(), limit);
 
     let start_block = start_block.unwrap_or(0);
     let end_block = end_block.unwrap_or(u64::MAX);
+    let is_change_based = segment.is_change_based();
 
     info!(
         "Computing checksum for {} static files, start_block={}, end_block={}, limit={:?}",
@@ -149,7 +149,8 @@ fn checksum_static_file<N: CliNodeTypes<ChainSpec: EthereumHardforks>>(
         if limit == usize::MAX { None } else { Some(limit) }
     );
 
-    'outer: for (block_range, _header) in ranges.iter().sorted_by_key(|(range, _)| range.start()) {
+    let mut reached_limit = false;
+    for (block_range, _header) in ranges.iter().sorted_by_key(|(range, _)| range.start()) {
         if block_range.end() < start_block || block_range.start() > end_block {
             continue;
         }
@@ -167,28 +168,42 @@ fn checksum_static_file<N: CliNodeTypes<ChainSpec: EthereumHardforks>>(
 
         let mut cursor = jar_provider.cursor()?;
 
-        while let Ok(Some(row)) = cursor.next_row() {
-            for col_data in row.iter() {
-                hasher.write(col_data);
-            }
+        if is_change_based {
+            let offsets = jar_provider.read_changeset_offsets()?.ok_or_else(|| {
+                eyre::eyre!(
+                    "Missing changeset offsets sidecar for segment {} at range {}",
+                    segment,
+                    block_range
+                )
+            })?;
+            let input = ChangeBasedChecksumInput {
+                segment,
+                block_range_start: block_range.start(),
+                start_block,
+                end_block,
+                offsets: &offsets,
+            };
 
-            total += 1;
-
-            if total.is_multiple_of(PROGRESS_LOG_INTERVAL) {
-                info!("Hashed {total} entries.");
-            }
-
-            if total >= limit {
-                break 'outer;
+            reached_limit = checksum_change_based_segment(&mut checksummer, input, &mut cursor)?;
+        } else {
+            while let Some(row) = cursor.next_row()? {
+                if checksummer.write_row(&row) {
+                    reached_limit = true;
+                    break;
+                }
             }
         }
 
         // Explicitly drop provider before removing from cache to avoid deadlock
         drop(jar_provider);
         static_file_provider.remove_cached_provider(segment, fixed_block_range.end());
+
+        if reached_limit {
+            break;
+        }
     }
 
-    let checksum = hasher.finish();
+    let (checksum, total) = checksummer.finish();
     let elapsed = start_time.elapsed();
 
     info!(
@@ -267,7 +282,7 @@ impl<N: ProviderNodeTypes> TableViewer<(u64, Duration)> for ChecksumViewer<'_, N
 
             total = index + 1;
             if total >= limit {
-                break
+                break;
             }
         }
 
@@ -285,3 +300,139 @@ impl<N: ProviderNodeTypes> TableViewer<(u64, Duration)> for ChecksumViewer<'_, N
         Ok((checksum, elapsed))
     }
 }
+
+/// Accumulates a checksum over key-value entries, tracking count and limit.
+struct Checksummer<H> {
+    hasher: H,
+    total: usize,
+    limit: usize,
+}
+
+impl<H: Hasher> Checksummer<H> {
+    fn new(hasher: H, limit: usize) -> Self {
+        Self { hasher, total: 0, limit }
+    }
+
+    /// Hash a row's columns (non-changeset segments). Returns `true` if the limit is reached.
+    fn write_row(&mut self, row: &[&[u8]]) -> bool {
+        for col in row {
+            self.hasher.write(col);
+        }
+        self.advance()
+    }
+
+    /// Hash a key + value as two separate writes, matching MDBX raw entry semantics.
+    /// Write boundaries matter: foldhash rotates its accumulator by `len` on each `write`.
+    fn write_entry(&mut self, key: &[u8], value: &[u8]) -> bool {
+        self.hasher.write(key);
+        self.hasher.write(value);
+        self.advance()
+    }
+
+    fn advance(&mut self) -> bool {
+        self.total += 1;
+        if self.total.is_multiple_of(PROGRESS_LOG_INTERVAL) {
+            info!("Hashed {} entries.", self.total);
+        }
+        self.total >= self.limit
+    }
+
+    fn finish(self) -> (u64, usize) {
+        (self.hasher.finish(), self.total)
+    }
+}
+
+/// Reconstruct MDBX `StorageChangeSets` key/value boundaries from a static-file row.
+///
+/// MDBX layout:
+/// - key: `BlockNumberAddress` => `[8B block_number][20B address]`
+/// - value: `StorageEntry` => `[32B storage_key][compact U256 value]`
+///
+/// Static-file row layout for `StorageBeforeTx`:
+/// - `[20B address][32B storage_key][compact U256 value]`
+fn split_storage_changeset_row(block_number: u64, row: &[u8]) -> eyre::Result<([u8; 28], &[u8])> {
+    if row.len() < 20 {
+        return Err(eyre::eyre!(
+            "Storage changeset row too short: expected at least 20 bytes, got {}",
+            row.len()
+        ));
+    }
+
+    let mut key_buf = [0u8; 28];
+    key_buf[..8].copy_from_slice(&block_number.to_be_bytes());
+    key_buf[8..].copy_from_slice(&row[..20]);
+    Ok((key_buf, &row[20..]))
+}
+
+struct ChangeBasedChecksumInput<'a> {
+    segment: StaticFileSegment,
+    block_range_start: u64,
+    start_block: u64,
+    end_block: u64,
+    offsets: &'a [ChangesetOffset],
+}
+
+fn checksum_change_based_segment<H: Hasher>(
+    checksummer: &mut Checksummer<H>,
+    input: ChangeBasedChecksumInput<'_>,
+    cursor: &mut reth_db::static_file::StaticFileCursor<'_>,
+) -> eyre::Result<bool> {
+    let ChangeBasedChecksumInput { segment, block_range_start, start_block, end_block, offsets } =
+        input;
+    let is_storage = segment.is_storage_change_sets();
+    let mut reached_limit = false;
+
+    for (offset_index, offset) in offsets.iter().enumerate() {
+        let block_number = block_range_start + offset_index as u64;
+        let include = block_number >= start_block && block_number <= end_block;
+
+        for _ in 0..offset.num_changes() {
+            let row = cursor.next_row()?.ok_or_else(|| {
+                eyre::eyre!(
+                    "Unexpected EOF while checksumming {} static file at range starting {}",
+                    segment,
+                    block_range_start
+                )
+            })?;
+
+            if !include {
+                continue;
+            }
+
+            // Reconstruct MDBX key/value write boundaries. foldhash rotates
+            // its accumulator by `len` on each write(), so boundaries must
+            // match exactly.
+            let done = if is_storage {
+                // StorageChangeSets: MDBX key = BlockNumberAddress (28B),
+                // value = compact StorageEntry. Column 0 is compact
+                // StorageBeforeTx = [20B address][32B key][compact U256].
+                let col = row[0];
+                let (key, value) = split_storage_changeset_row(block_number, col)?;
+                checksummer.write_entry(&key, value)
+            } else {
+                // AccountChangeSets: MDBX key = BlockNumber (8B),
+                // value = compact AccountBeforeTx (= column 0).
+                checksummer.write_entry(&block_number.to_be_bytes(), row[0])
+            };
+
+            if done {
+                reached_limit = true;
+                break;
+            }
+        }
+
+        if reached_limit {
+            break;
+        }
+    }
+
+    if !reached_limit && cursor.next_row()?.is_some() {
+        return Err(eyre::eyre!(
+            "Changeset offsets do not cover all rows for {} at range starting {}",
+            segment,
+            block_range_start
+        ));
+    }
+
+    Ok(reached_limit)
+}

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

@@ -98,7 +98,7 @@ impl Command {
                         )?;
 
                         if let Some(entry) = entry {
-                            let se: reth_primitives_traits::StorageEntry = entry.into();
+                            let se: reth_primitives_traits::StorageEntry = entry;
                             println!("{}", serde_json::to_string_pretty(&se)?);
                         } else {
                             error!(target: "reth::cli", "No content for the given table key.");
@@ -110,7 +110,7 @@ impl Command {
                     let serializable: Vec<_> = changesets
                         .into_iter()
                         .map(|(addr, entry)| {
-                            let se: reth_primitives_traits::StorageEntry = entry.into();
+                            let se: reth_primitives_traits::StorageEntry = entry;
                             (addr, se)
                         })
                         .collect();

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

@@ -19,6 +19,7 @@ mod list;
 mod prune_checkpoints;
 mod repair_trie;
 mod settings;
+mod stage_checkpoints;
 mod state;
 mod static_file_header;
 mod stats;
@@ -70,6 +71,8 @@ pub enum Subcommands {
     Settings(settings::Command),
     /// View or set prune checkpoints
     PruneCheckpoints(prune_checkpoints::Command),
+    // View or set stage checkpoints
+    StageCheckpoints(stage_checkpoints::Command),
     /// Gets storage size information for an account
     AccountStorage(account_storage::Command),
     /// Gets account state and storage at a specific block
@@ -213,6 +216,11 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
                     command.execute(&tool)?;
                 });
             }
+            Subcommands::StageCheckpoints(command) => {
+                db_exec!(self.env, tool, N, command.access_rights(), {
+                    command.execute(&tool)?;
+                });
+            }
             Subcommands::AccountStorage(command) => {
                 db_exec!(self.env, tool, N, AccessRights::RO, {
                     command.execute(&tool)?;

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

@@ -5,7 +5,6 @@ use reth_cli_util::parse_socket_address;
 use reth_db_api::{
     cursor::{DbCursorRO, DbCursorRW, DbDupCursorRO},
     database::Database,
-    tables,
     transaction::{DbTx, DbTxMut},
 };
 use reth_db_common::DbTool;
@@ -21,13 +20,15 @@ use reth_node_metrics::{
 };
 use reth_provider::{providers::ProviderNodeTypes, ChainSpecProvider, StageCheckpointReader};
 use reth_stages::StageId;
+use reth_storage_api::StorageSettingsCache;
 use reth_tasks::TaskExecutor;
 use reth_trie::{
     verify::{Output, Verifier},
     Nibbles,
 };
-use reth_trie_common::{StorageTrieEntry, StoredNibbles, StoredNibblesSubKey};
-use reth_trie_db::{DatabaseHashedCursorFactory, DatabaseTrieCursorFactory};
+use reth_trie_db::{
+    DatabaseHashedCursorFactory, DatabaseTrieCursorFactory, StorageTrieEntryLike, TrieTableAdapter,
+};
 use std::{
     net::SocketAddr,
     time::{Duration, Instant},
@@ -116,9 +117,13 @@ fn verify_only<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()> {
     let mut tx = db.tx()?;
     tx.disable_long_read_transaction_safety();
 
+    reth_trie_db::with_adapter!(tool.provider_factory, |A| do_verify_only::<_, A>(&tx))
+}
+
+fn do_verify_only<TX: DbTx, A: TrieTableAdapter>(tx: &TX) -> eyre::Result<()> {
     // Create the verifier
-    let hashed_cursor_factory = DatabaseHashedCursorFactory::new(&tx);
-    let trie_cursor_factory = DatabaseTrieCursorFactory::new(&tx);
+    let hashed_cursor_factory = DatabaseHashedCursorFactory::new(tx);
+    let trie_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(tx);
     let verifier = Verifier::new(&trie_cursor_factory, hashed_cursor_factory)?;
 
     let metrics = RepairTrieMetrics::new();
@@ -209,17 +214,37 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
     // Check that a pipeline sync isn't in progress.
     verify_checkpoints(provider_rw.as_ref())?;
 
+    let inconsistent_nodes = reth_trie_db::with_adapter!(tool.provider_factory, |A| {
+        do_verify_and_repair::<_, A>(&mut provider_rw)?
+    });
+
+    if inconsistent_nodes == 0 {
+        info!("No inconsistencies found");
+    } else {
+        provider_rw.commit()?;
+        info!("Repaired {} inconsistencies and committed changes", inconsistent_nodes);
+    }
+
+    Ok(())
+}
+
+fn do_verify_and_repair<N: ProviderNodeTypes, A: TrieTableAdapter>(
+    provider_rw: &mut reth_provider::DatabaseProviderRW<N::DB, N>,
+) -> eyre::Result<usize>
+where
+    <N::DB as reth_db_api::database::Database>::TXMut: DbTxMut + DbTx,
+{
     // Create cursors for making modifications with
     let tx = provider_rw.tx_mut();
     tx.disable_long_read_transaction_safety();
-    let mut account_trie_cursor = tx.cursor_write::<tables::AccountsTrie>()?;
-    let mut storage_trie_cursor = tx.cursor_dup_write::<tables::StoragesTrie>()?;
+    let mut account_trie_cursor = tx.cursor_write::<A::AccountTrieTable>()?;
+    let mut storage_trie_cursor = tx.cursor_dup_write::<A::StorageTrieTable>()?;
 
-    // Create the cursor factories. These cannot accept the `&mut` tx above because they require it
-    // to be AsRef.
+    // Create the cursor factories. These cannot accept the `&mut` tx above because they
+    // require it to be AsRef.
     let tx = provider_rw.tx_ref();
     let hashed_cursor_factory = DatabaseHashedCursorFactory::new(tx);
-    let trie_cursor_factory = DatabaseTrieCursorFactory::new(tx);
+    let trie_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(tx);
 
     // Create the verifier
     let verifier = Verifier::new(&trie_cursor_factory, hashed_cursor_factory)?;
@@ -257,17 +282,17 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
         match output {
             Output::AccountExtra(path, _node) => {
                 // Extra account node in trie, remove it
-                let nibbles = StoredNibbles(path);
-                if account_trie_cursor.seek_exact(nibbles)?.is_some() {
+                let key: A::AccountKey = path.into();
+                if account_trie_cursor.seek_exact(key)?.is_some() {
                     account_trie_cursor.delete_current()?;
                 }
             }
             Output::StorageExtra(account, path, _node) => {
                 // Extra storage node in trie, remove it
-                let nibbles = StoredNibblesSubKey(path);
+                let subkey: A::StorageSubKey = path.into();
                 if storage_trie_cursor
-                    .seek_by_key_subkey(account, nibbles.clone())?
-                    .filter(|e| e.nibbles == nibbles)
+                    .seek_by_key_subkey(account, subkey.clone())?
+                    .filter(|e| *e.nibbles() == subkey)
                     .is_some()
                 {
                     storage_trie_cursor.delete_current()?;
@@ -276,23 +301,23 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
             Output::AccountWrong { path, expected: node, .. } |
             Output::AccountMissing(path, node) => {
                 // Wrong/missing account node value, upsert it
-                let nibbles = StoredNibbles(path);
-                account_trie_cursor.upsert(nibbles, &node)?;
+                let key: A::AccountKey = path.into();
+                account_trie_cursor.upsert(key, &node)?;
             }
             Output::StorageWrong { account, path, expected: node, .. } |
             Output::StorageMissing(account, path, node) => {
                 // Wrong/missing storage node value, upsert it
                 // (We can't just use `upsert` method with a dup cursor, it's not properly
                 // supported)
-                let nibbles = StoredNibblesSubKey(path);
+                let subkey: A::StorageSubKey = path.into();
+                let entry = A::StorageValue::new(subkey.clone(), node);
                 if storage_trie_cursor
-                    .seek_by_key_subkey(account, nibbles.clone())?
-                    .filter(|v| v.nibbles == nibbles)
+                    .seek_by_key_subkey(account, subkey.clone())?
+                    .filter(|v| *v.nibbles() == subkey)
                     .is_some()
                 {
                     storage_trie_cursor.delete_current()?;
                 }
-                let entry = StorageTrieEntry { nibbles, node };
                 storage_trie_cursor.upsert(account, &entry)?;
             }
             Output::Progress(path) => {
@@ -304,14 +329,7 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
         }
     }
 
-    if inconsistent_nodes == 0 {
-        info!("No inconsistencies found");
-    } else {
-        provider_rw.commit()?;
-        info!("Repaired {} inconsistencies and committed changes", inconsistent_nodes);
-    }
-
-    Ok(())
+    Ok(inconsistent_nodes as usize)
 }
 
 /// Output progress information based on the last seen account path.

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

@@ -0,0 +1,297 @@
+//! `reth db stage-checkpoints` command for viewing and setting stage checkpoint values.
+
+use clap::{Args, Parser, Subcommand, ValueEnum};
+use reth_db_common::DbTool;
+use reth_provider::{
+    providers::ProviderNodeTypes, DBProvider, DatabaseProviderFactory, StageCheckpointReader,
+    StageCheckpointWriter,
+};
+use reth_stages::StageId;
+
+use crate::common::AccessRights;
+
+/// `reth db stage-checkpoints` subcommand
+#[derive(Debug, Parser)]
+pub struct Command {
+    #[command(subcommand)]
+    command: Subcommands,
+}
+
+impl Command {
+    /// Returns database access rights required for the command.
+    pub fn access_rights(&self) -> AccessRights {
+        match &self.command {
+            Subcommands::Get { .. } => AccessRights::RO,
+            Subcommands::Set(_) => AccessRights::RW,
+        }
+    }
+
+    /// Execute the command
+    pub fn execute<N: ProviderNodeTypes>(self, tool: &DbTool<N>) -> eyre::Result<()> {
+        match self.command {
+            Subcommands::Get { stage } => Self::get(tool, stage),
+            Subcommands::Set(args) => Self::set(tool, args),
+        }
+    }
+
+    fn get<N: ProviderNodeTypes>(tool: &DbTool<N>, stage: Option<StageArg>) -> eyre::Result<()> {
+        let provider = tool.provider_factory.provider()?;
+
+        match stage {
+            Some(stage) => {
+                let stage_id = stage.into();
+                let checkpoint = provider.get_stage_checkpoint(stage_id)?;
+                println!("{stage_id}: {checkpoint:?}");
+            }
+            None => {
+                let mut checkpoints = provider.get_all_checkpoints()?;
+                checkpoints.sort_by(|a, b| a.0.cmp(&b.0));
+                for (stage, checkpoint) in checkpoints {
+                    println!("{stage}: {checkpoint:?}");
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    fn set<N: ProviderNodeTypes>(tool: &DbTool<N>, args: SetArgs) -> eyre::Result<()> {
+        let stage_id: StageId = args.stage.into();
+        let provider_rw = tool.provider_factory.database_provider_rw()?;
+
+        let previous = provider_rw.get_stage_checkpoint(stage_id)?;
+        let mut checkpoint = previous.unwrap_or_default();
+        checkpoint.block_number = args.block_number;
+
+        if args.clear_stage_unit {
+            checkpoint.stage_checkpoint = None;
+        }
+
+        provider_rw.save_stage_checkpoint(stage_id, checkpoint)?;
+
+        provider_rw.commit()?;
+
+        println!("Updated checkpoint for {stage_id}: {checkpoint:?}");
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, Subcommand)]
+enum Subcommands {
+    /// Get stage checkpoint(s) from database.
+    Get {
+        /// Specific stage to query. If omitted, shows all stages.
+        #[arg(long, value_enum)]
+        stage: Option<StageArg>,
+    },
+    /// Set a stage checkpoint.
+    Set(SetArgs),
+}
+
+/// Arguments for the `set` subcommand.
+#[derive(Debug, Args)]
+pub struct SetArgs {
+    /// Stage to update.
+    #[arg(long, value_enum)]
+    stage: StageArg,
+
+    /// Block number to set as stage checkpoint.
+    #[arg(long)]
+    block_number: u64,
+
+    /// Clear stage-specific unit checkpoint payload.
+    #[arg(long)]
+    clear_stage_unit: bool,
+}
+
+/// CLI-friendly stage names.
+#[derive(Debug, Clone, Copy, ValueEnum)]
+#[clap(rename_all = "kebab-case")]
+pub enum StageArg {
+    Era,
+    Headers,
+    Bodies,
+    SenderRecovery,
+    Execution,
+    PruneSenderRecovery,
+    MerkleUnwind,
+    AccountHashing,
+    StorageHashing,
+    MerkleExecute,
+    TransactionLookup,
+    IndexStorageHistory,
+    IndexAccountHistory,
+    Prune,
+    Finish,
+}
+
+impl From<StageArg> for StageId {
+    fn from(arg: StageArg) -> Self {
+        match arg {
+            StageArg::Era => Self::Era,
+            StageArg::Headers => Self::Headers,
+            StageArg::Bodies => Self::Bodies,
+            StageArg::SenderRecovery => Self::SenderRecovery,
+            StageArg::Execution => Self::Execution,
+            StageArg::PruneSenderRecovery => Self::PruneSenderRecovery,
+            StageArg::MerkleUnwind => Self::MerkleUnwind,
+            StageArg::AccountHashing => Self::AccountHashing,
+            StageArg::StorageHashing => Self::StorageHashing,
+            StageArg::MerkleExecute => Self::MerkleExecute,
+            StageArg::TransactionLookup => Self::TransactionLookup,
+            StageArg::IndexStorageHistory => Self::IndexStorageHistory,
+            StageArg::IndexAccountHistory => Self::IndexAccountHistory,
+            StageArg::Prune => Self::Prune,
+            StageArg::Finish => Self::Finish,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use clap::Parser;
+    use reth_provider::{
+        test_utils::create_test_provider_factory, DBProvider, DatabaseProviderFactory,
+        StageCheckpointReader, StageCheckpointWriter,
+    };
+    use reth_stages::StageCheckpoint;
+
+    #[test]
+    fn parse_set_args() {
+        let command = Command::parse_from([
+            "stage-checkpoints",
+            "set",
+            "--stage",
+            "headers",
+            "--block-number",
+            "123",
+        ]);
+
+        assert!(matches!(
+            command.command,
+            Subcommands::Set(SetArgs {
+                stage: StageArg::Headers,
+                block_number: 123,
+                clear_stage_unit: false,
+            })
+        ));
+    }
+
+    #[test]
+    fn set_overwrites_block_number() {
+        let provider_factory = create_test_provider_factory();
+        let tool = DbTool::new(provider_factory.clone()).expect("db tool");
+
+        {
+            let provider_rw = provider_factory.database_provider_rw().expect("rw provider");
+            provider_rw
+                .save_stage_checkpoint(StageId::Headers, StageCheckpoint::new(10))
+                .expect("save checkpoint");
+            provider_rw.commit().expect("commit initial checkpoint");
+        }
+
+        let command = Command {
+            command: Subcommands::Set(SetArgs {
+                stage: StageArg::Headers,
+                block_number: 42,
+                clear_stage_unit: false,
+            }),
+        };
+
+        command.execute(&tool).expect("execute command");
+
+        let provider = provider_factory.provider().expect("provider");
+        let checkpoint = provider
+            .get_stage_checkpoint(StageId::Headers)
+            .expect("get stage checkpoint")
+            .expect("missing stage checkpoint");
+
+        assert_eq!(checkpoint.block_number, 42);
+    }
+
+    #[test]
+    fn set_preserves_stage_unit_checkpoint_unless_cleared() {
+        let provider_factory = create_test_provider_factory();
+        let tool = DbTool::new(provider_factory.clone()).expect("db tool");
+
+        {
+            let provider_rw = provider_factory.database_provider_rw().expect("rw provider");
+            let checkpoint = StageCheckpoint::new(10).with_block_range(&StageId::Execution, 5, 10);
+            provider_rw
+                .save_stage_checkpoint(StageId::Execution, checkpoint)
+                .expect("save checkpoint");
+            provider_rw.commit().expect("commit initial checkpoint");
+        }
+
+        Command {
+            command: Subcommands::Set(SetArgs {
+                stage: StageArg::Execution,
+                block_number: 11,
+                clear_stage_unit: false,
+            }),
+        }
+        .execute(&tool)
+        .expect("execute command");
+
+        let provider = provider_factory.provider().expect("provider");
+        let checkpoint = provider
+            .get_stage_checkpoint(StageId::Execution)
+            .expect("get stage checkpoint")
+            .expect("missing stage checkpoint");
+        assert!(checkpoint.stage_checkpoint.is_some());
+
+        Command {
+            command: Subcommands::Set(SetArgs {
+                stage: StageArg::Execution,
+                block_number: 12,
+                clear_stage_unit: true,
+            }),
+        }
+        .execute(&tool)
+        .expect("execute command");
+
+        let checkpoint = provider_factory
+            .provider()
+            .expect("provider")
+            .get_stage_checkpoint(StageId::Execution)
+            .expect("get stage checkpoint")
+            .expect("missing stage checkpoint");
+        assert!(checkpoint.stage_checkpoint.is_none());
+    }
+
+    #[test]
+    fn set_preserves_checkpoint_progress() {
+        let provider_factory = create_test_provider_factory();
+        let tool = DbTool::new(provider_factory.clone()).expect("db tool");
+
+        {
+            let provider_rw = provider_factory.database_provider_rw().expect("rw provider");
+            provider_rw
+                .save_stage_checkpoint(StageId::MerkleExecute, StageCheckpoint::new(10))
+                .expect("save checkpoint");
+            provider_rw
+                .save_stage_checkpoint_progress(StageId::MerkleExecute, vec![1, 2, 3])
+                .expect("save progress");
+            provider_rw.commit().expect("commit initial checkpoint");
+        }
+
+        Command {
+            command: Subcommands::Set(SetArgs {
+                stage: StageArg::MerkleExecute,
+                block_number: 20,
+                clear_stage_unit: false,
+            }),
+        }
+        .execute(&tool)
+        .expect("execute command");
+
+        let provider = provider_factory.provider().expect("provider");
+        let progress = provider
+            .get_stage_checkpoint_progress(StageId::MerkleExecute)
+            .expect("get stage checkpoint progress");
+
+        assert_eq!(progress, Some(vec![1, 2, 3]));
+    }
+}

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

@@ -297,21 +297,18 @@ where
     }
 
     match event {
-        Event::Key(key) => {
-            if key.kind == event::KeyEventKind::Press {
-                match key.code {
-                    KeyCode::Char('q') | KeyCode::Char('Q') => return Ok(true),
-                    KeyCode::Down => app.next(),
-                    KeyCode::Up => app.previous(),
-                    KeyCode::Right => app.next_page(),
-                    KeyCode::Left => app.previous_page(),
-                    KeyCode::Char('G') => {
-                        app.mode = ViewMode::GoToPage;
-                    }
-                    _ => {}
-                }
+        Event::Key(key) if key.kind == event::KeyEventKind::Press => match key.code {
+            KeyCode::Char('q') | KeyCode::Char('Q') => return Ok(true),
+            KeyCode::Down => app.next(),
+            KeyCode::Up => app.previous(),
+            KeyCode::Right => app.next_page(),
+            KeyCode::Left => app.previous_page(),
+            KeyCode::Char('G') => {
+                app.mode = ViewMode::GoToPage;
             }
-        }
+            _ => {}
+        },
+        Event::Key(_) => {}
         Event::Mouse(e) => match e.kind {
             MouseEventKind::ScrollDown => app.next(),
             MouseEventKind::ScrollUp => app.previous(),

crates/cli/commands/src/re_execute.rs

@@ -20,7 +20,10 @@ use reth_provider::{
 use reth_revm::database::StateProviderDatabase;
 use reth_stages::stages::calculate_gas_used_from_headers;
 use std::{
-    sync::Arc,
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
     time::{Duration, Instant},
 };
 use tokio::{sync::mpsc, task::JoinSet};
@@ -46,6 +49,10 @@ pub struct Command<C: ChainSpecParser> {
     #[arg(long)]
     num_tasks: Option<u64>,
 
+    /// Number of blocks each worker processes before grabbing the next chunk.
+    #[arg(long, default_value = "5000")]
+    blocks_per_chunk: u64,
+
     /// Continues with execution when an invalid block is encountered and collects these blocks.
     #[arg(long)]
     skip_invalid_blocks: bool,
@@ -92,12 +99,10 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
             std::thread::available_parallelism().map(|n| n.get() as u64).unwrap_or(10)
         });
 
-        let total_blocks = max_block - min_block;
         let total_gas = calculate_gas_used_from_headers(
             &provider_factory.static_file_provider(),
             min_block..=max_block,
         )?;
-        let blocks_per_task = total_blocks / num_tasks;
 
         let db_at = {
             let provider_factory = provider_factory.clone();
@@ -109,18 +114,17 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
         };
 
         let skip_invalid_blocks = self.skip_invalid_blocks;
+        let blocks_per_chunk = self.blocks_per_chunk;
         let (stats_tx, mut stats_rx) = mpsc::unbounded_channel();
         let (info_tx, mut info_rx) = mpsc::unbounded_channel();
         let cancellation = CancellationToken::new();
         let _guard = cancellation.drop_guard();
 
-        let mut tasks = JoinSet::new();
-        for i in 0..num_tasks {
-            let start_block = min_block + i * blocks_per_task;
-            let end_block =
-                if i == num_tasks - 1 { max_block } else { start_block + blocks_per_task };
+        // Shared counter for work stealing: workers atomically grab the next chunk of blocks.
+        let next_block = Arc::new(AtomicU64::new(min_block));
 
-            // Spawn thread executing blocks
+        let mut tasks = JoinSet::new();
+        for _ in 0..num_tasks {
             let provider_factory = provider_factory.clone();
             let evm_config = components.evm_config().clone();
             let consensus = components.consensus().clone();
@@ -128,95 +132,122 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
             let stats_tx = stats_tx.clone();
             let info_tx = info_tx.clone();
             let cancellation = cancellation.clone();
+            let next_block = Arc::clone(&next_block);
             tasks.spawn_blocking(move || {
-                let mut executor = evm_config.batch_executor(db_at(start_block - 1));
-                let mut executor_created = Instant::now();
                 let executor_lifetime = Duration::from_secs(120);
 
-                'blocks: for block in start_block..end_block {
+                loop {
                     if cancellation.is_cancelled() {
-                        // exit if the program is being terminated
-                        break
+                        break;
                     }
 
-                    let block = provider_factory
-                        .recovered_block(block.into(), TransactionVariant::NoHash)?
-                        .unwrap();
+                    // Atomically grab the next chunk of blocks.
+                    let chunk_start =
+                        next_block.fetch_add(blocks_per_chunk, Ordering::Relaxed);
+                    if chunk_start >= max_block {
+                        break;
+                    }
+                    let chunk_end = (chunk_start + blocks_per_chunk).min(max_block);
 
-                    let result = match executor.execute_one(&block) {
-                        Ok(result) => result,
-                        Err(err) => {
-                            if skip_invalid_blocks {
-                                executor = evm_config.batch_executor(db_at(block.number()));
-                                let _ = info_tx.send((block, eyre::Report::new(err)));
-                                continue
-                            }
-                            return Err(err.into())
+                    let mut executor = evm_config.batch_executor(db_at(chunk_start - 1));
+                    let mut executor_created = Instant::now();
+
+                    'blocks: for block in chunk_start..chunk_end {
+                        if cancellation.is_cancelled() {
+                            break;
                         }
-                    };
 
-                    if let Err(err) = consensus
-                        .validate_block_post_execution(&block, &result, None)
-                        .wrap_err_with(|| {
-                            format!("Failed to validate block {} {}", block.number(), block.hash())
-                        })
-                    {
-                        let correct_receipts =
-                            provider_factory.receipts_by_block(block.number().into())?.unwrap();
+                        let block = provider_factory
+                            .recovered_block(block.into(), TransactionVariant::NoHash)?
+                            .unwrap();
 
-                        for (i, (receipt, correct_receipt)) in
-                            result.receipts.iter().zip(correct_receipts.iter()).enumerate()
-                        {
-                            if receipt != correct_receipt {
-                                let tx_hash = block.body().transactions()[i].tx_hash();
-                                error!(
-                                    ?receipt,
-                                    ?correct_receipt,
-                                    index = i,
-                                    ?tx_hash,
-                                    "Invalid receipt"
-                                );
-                                let expected_gas_used = correct_receipt.cumulative_gas_used() -
-                                    if i == 0 {
-                                        0
-                                    } else {
-                                        correct_receipts[i - 1].cumulative_gas_used()
-                                    };
-                                let got_gas_used = receipt.cumulative_gas_used() -
-                                    if i == 0 {
-                                        0
-                                    } else {
-                                        result.receipts[i - 1].cumulative_gas_used()
-                                    };
-                                if got_gas_used != expected_gas_used {
-                                    let mismatch = GotExpected {
-                                        expected: expected_gas_used,
-                                        got: got_gas_used,
-                                    };
-
-                                    error!(number=?block.number(), ?mismatch, "Gas usage mismatch");
-                                    if skip_invalid_blocks {
-                                        executor = evm_config.batch_executor(db_at(block.number()));
-                                        let _ = info_tx.send((block, err));
-                                        continue 'blocks;
-                                    }
-                                    return Err(err);
+                        let result = match executor.execute_one(&block) {
+                            Ok(result) => result,
+                            Err(err) => {
+                                if skip_invalid_blocks {
+                                    executor =
+                                        evm_config.batch_executor(db_at(block.number()));
+                                    let _ =
+                                        info_tx.send((block, eyre::Report::new(err)));
+                                    continue
                                 }
-                            } else {
-                                continue;
+                                return Err(err.into())
                             }
+                        };
+
+                        if let Err(err) = consensus
+                            .validate_block_post_execution(&block, &result, None)
+                            .wrap_err_with(|| {
+                                format!(
+                                    "Failed to validate block {} {}",
+                                    block.number(),
+                                    block.hash()
+                                )
+                            })
+                        {
+                            let correct_receipts = provider_factory
+                                .receipts_by_block(block.number().into())?
+                                .unwrap();
+
+                            for (i, (receipt, correct_receipt)) in
+                                result.receipts.iter().zip(correct_receipts.iter()).enumerate()
+                            {
+                                if receipt != correct_receipt {
+                                    let tx_hash =
+                                        block.body().transactions()[i].tx_hash();
+                                    error!(
+                                        ?receipt,
+                                        ?correct_receipt,
+                                        index = i,
+                                        ?tx_hash,
+                                        "Invalid receipt"
+                                    );
+                                    let expected_gas_used =
+                                        correct_receipt.cumulative_gas_used() -
+                                            if i == 0 {
+                                                0
+                                            } else {
+                                                correct_receipts[i - 1]
+                                                    .cumulative_gas_used()
+                                            };
+                                    let got_gas_used = receipt.cumulative_gas_used() -
+                                        if i == 0 {
+                                            0
+                                        } else {
+                                            result.receipts[i - 1].cumulative_gas_used()
+                                        };
+                                    if got_gas_used != expected_gas_used {
+                                        let mismatch = GotExpected {
+                                            expected: expected_gas_used,
+                                            got: got_gas_used,
+                                        };
+
+                                        error!(number=?block.number(), ?mismatch, "Gas usage mismatch");
+                                        if skip_invalid_blocks {
+                                            executor = evm_config
+                                                .batch_executor(db_at(block.number()));
+                                            let _ = info_tx.send((block, err));
+                                            continue 'blocks;
+                                        }
+                                        return Err(err);
+                                    }
+                                } else {
+                                    continue;
+                                }
+                            }
+
+                            return Err(err);
                         }
+                        let _ = stats_tx.send(block.gas_used());
 
-                        return Err(err);
-                    }
-                    let _ = stats_tx.send(block.gas_used());
-
-                    // Reset DB once in a while to avoid OOM or read tx timeouts
-                    if executor.size_hint() > 1_000_000 ||
-                        executor_created.elapsed() > executor_lifetime
-                    {
-                        executor = evm_config.batch_executor(db_at(block.number()));
-                        executor_created = Instant::now();
+                        // Reset DB once in a while to avoid OOM or read tx timeouts
+                        if executor.size_hint() > 1_000_000 ||
+                            executor_created.elapsed() > executor_lifetime
+                        {
+                            executor =
+                                evm_config.batch_executor(db_at(block.number()));
+                            executor_created = Instant::now();
+                        }
                     }
                 }
 

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

@@ -52,7 +52,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
         Comp: CliNodeComponents<N>,
         F: FnOnce(Arc<C::ChainSpec>) -> Comp,
     {
-        let Environment { provider_factory, config, .. } =
+        let Environment { provider_factory, config, data_dir: _ } =
             self.env.init::<N>(AccessRights::RW, runtime)?;
 
         let target = self.command.unwind_target(provider_factory.clone())?;

crates/consensus/common/Cargo.toml

@@ -19,6 +19,7 @@ reth-consensus.workspace = true
 reth-primitives-traits.workspace = true
 alloy-consensus.workspace = true
 alloy-eips.workspace = true
+alloy-primitives.workspace = true
 
 [dev-dependencies]
 alloy-primitives = { workspace = true, features = ["rand"] }

crates/consensus/common/src/validation.rs

@@ -2,6 +2,7 @@
 
 use alloy_consensus::{BlockHeader as _, EMPTY_OMMER_ROOT_HASH};
 use alloy_eips::{eip4844::DATA_GAS_PER_BLOB, eip7840::BlobParams};
+use alloy_primitives::B256;
 use reth_chainspec::{EthChainSpec, EthereumHardfork, EthereumHardforks};
 use reth_consensus::ConsensusError;
 use reth_primitives_traits::{
@@ -141,6 +142,27 @@ pub fn validate_block_pre_execution<B, ChainSpec>(
     block: &SealedBlock<B>,
     chain_spec: &ChainSpec,
 ) -> Result<(), ConsensusError>
+where
+    B: Block,
+    ChainSpec: EthChainSpec + EthereumHardforks,
+{
+    validate_block_pre_execution_with_tx_root(block, chain_spec, None)
+}
+
+/// Validate a block without regard for state using an optional pre-computed transaction root.
+///
+/// - Compares the ommer hash in the block header to the block body
+/// - Compares the transactions root in the block header to the block body
+/// - Pre-execution transaction validation
+///
+/// If `transaction_root` is provided, it is used instead of recomputing the transaction trie
+/// root from the block body. The caller must ensure this value was derived from
+/// `block.body().calculate_tx_root()`.
+pub fn validate_block_pre_execution_with_tx_root<B, ChainSpec>(
+    block: &SealedBlock<B>,
+    chain_spec: &ChainSpec,
+    transaction_root: Option<B256>,
+) -> Result<(), ConsensusError>
 where
     B: Block,
     ChainSpec: EthChainSpec + EthereumHardforks,
@@ -148,8 +170,14 @@ where
     post_merge_hardfork_fields(block, chain_spec)?;
 
     // Check transaction root
-    if let Err(error) = block.ensure_transaction_root_valid() {
-        return Err(ConsensusError::BodyTransactionRootDiff(error.into()))
+    let expected_transaction_root = block.header().transactions_root();
+    let calculated_transaction_root =
+        transaction_root.unwrap_or_else(|| block.body().calculate_tx_root());
+    if calculated_transaction_root != expected_transaction_root {
+        return Err(ConsensusError::BodyTransactionRootDiff(
+            GotExpected { got: calculated_transaction_root, expected: expected_transaction_root }
+                .into(),
+        ))
     }
 
     Ok(())
@@ -426,7 +454,7 @@ pub fn validate_against_parent_4844<H: BlockHeader>(
 mod tests {
     use super::*;
     use alloy_consensus::{BlockBody, Header, TxEip4844};
-    use alloy_eips::eip4895::Withdrawals;
+    use alloy_eips::{eip4844::DATA_GAS_PER_BLOB, eip4895::Withdrawals};
     use alloy_primitives::{Address, Bytes, Signature, U256};
     use rand::Rng;
     use reth_chainspec::ChainSpecBuilder;
@@ -507,4 +535,66 @@ mod tests {
         // Test with custom larger limit - should pass
         assert!(validate_header_extra_data(&header_33, 64).is_ok());
     }
+
+    #[test]
+    fn precomputed_tx_root_correct_passes() {
+        let chain_spec = ChainSpecBuilder::mainnet().cancun_activated().build();
+
+        let transaction = mock_blob_tx(1, 1);
+        let tx_root = proofs::calculate_transaction_root(std::slice::from_ref(&transaction));
+
+        let header = Header {
+            base_fee_per_gas: Some(1337),
+            withdrawals_root: Some(proofs::calculate_withdrawals_root(&[])),
+            transactions_root: tx_root,
+            blob_gas_used: Some(DATA_GAS_PER_BLOB),
+            excess_blob_gas: Some(0),
+            ..Default::default()
+        };
+        let body = BlockBody {
+            transactions: vec![transaction],
+            ommers: vec![],
+            withdrawals: Some(Withdrawals::default()),
+        };
+
+        let block = SealedBlock::seal_slow(alloy_consensus::Block { header, body });
+
+        // Some(correct_root) should pass just like None
+        assert!(
+            validate_block_pre_execution_with_tx_root(&block, &chain_spec, Some(tx_root)).is_ok()
+        );
+        assert!(validate_block_pre_execution_with_tx_root(&block, &chain_spec, None).is_ok());
+    }
+
+    #[test]
+    fn precomputed_tx_root_wrong_fails() {
+        let chain_spec = ChainSpecBuilder::mainnet().cancun_activated().build();
+
+        let transaction = mock_blob_tx(1, 1);
+        let tx_root = proofs::calculate_transaction_root(std::slice::from_ref(&transaction));
+
+        let header = Header {
+            base_fee_per_gas: Some(1337),
+            withdrawals_root: Some(proofs::calculate_withdrawals_root(&[])),
+            transactions_root: tx_root,
+            blob_gas_used: Some(DATA_GAS_PER_BLOB),
+            excess_blob_gas: Some(0),
+            ..Default::default()
+        };
+        let body = BlockBody {
+            transactions: vec![transaction],
+            ommers: vec![],
+            withdrawals: Some(Withdrawals::default()),
+        };
+
+        let block = SealedBlock::seal_slow(alloy_consensus::Block { header, body });
+
+        let wrong_root = B256::repeat_byte(0xff);
+        assert!(matches!(
+            validate_block_pre_execution_with_tx_root(&block, &chain_spec, Some(wrong_root))
+                .unwrap_err(),
+            ConsensusError::BodyTransactionRootDiff(diff)
+                if diff.0.got == wrong_root && diff.0.expected == tx_root
+        ));
+    }
 }

crates/consensus/consensus/src/lib.rs

@@ -21,6 +21,12 @@ use core::error::Error;
 /// When provided to [`FullConsensus::validate_block_post_execution`], this allows skipping
 /// the receipt root computation and using the pre-computed values instead.
 pub type ReceiptRootBloom = (B256, Bloom);
+
+/// Pre-computed transaction root.
+///
+/// When provided to [`Consensus::validate_block_pre_execution_with_tx_root`], this allows
+/// skipping transaction trie reconstruction from the block body.
+pub type TransactionRoot = B256;
 use reth_execution_types::BlockExecutionResult;
 use reth_primitives_traits::{
     constants::{GAS_LIMIT_BOUND_DIVISOR, MAXIMUM_GAS_LIMIT_BLOCK, MINIMUM_GAS_LIMIT},
@@ -78,6 +84,22 @@ pub trait Consensus<B: Block>: HeaderValidator<B::Header> {
     ///
     /// Note: validating blocks does not include other validations of the Consensus
     fn validate_block_pre_execution(&self, block: &SealedBlock<B>) -> Result<(), ConsensusError>;
+
+    /// Validate a block disregarding world state using an optional pre-computed transaction root.
+    ///
+    /// If `transaction_root` is provided, the implementation should use the pre-computed
+    /// transaction root instead of recomputing it from the block body. The value must have been
+    /// derived from `block.body().calculate_tx_root()`.
+    ///
+    /// By default this falls back to [`Self::validate_block_pre_execution`].
+    fn validate_block_pre_execution_with_tx_root(
+        &self,
+        block: &SealedBlock<B>,
+        transaction_root: Option<TransactionRoot>,
+    ) -> Result<(), ConsensusError> {
+        let _ = transaction_root;
+        self.validate_block_pre_execution(block)
+    }
 }
 
 /// `HeaderValidator` is a protocol that validates headers and their relationships.

crates/e2e-test-utils/Cargo.toml

@@ -38,7 +38,6 @@ reth-ethereum-primitives.workspace = true
 reth-cli-commands.workspace = true
 reth-config.workspace = true
 reth-consensus.workspace = true
-reth-primitives.workspace = true
 reth-db-common.workspace = true
 reth-primitives-traits.workspace = true
 

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

@@ -275,8 +275,9 @@ mod tests {
     use crate::test_rlp_utils::{create_fcu_json, generate_test_blocks, write_blocks_to_rlp};
     use reth_chainspec::{ChainSpecBuilder, MAINNET};
     use reth_db::mdbx::DatabaseArguments;
+    use reth_ethereum_primitives::Block;
     use reth_payload_builder::EthPayloadBuilderAttributes;
-    use reth_primitives::SealedBlock;
+    use reth_primitives_traits::SealedBlock;
     use reth_provider::{
         test_utils::MockNodeTypesWithDB, BlockHashReader, BlockNumReader, BlockReaderIdExt,
     };
@@ -448,7 +449,7 @@ mod tests {
         chain_spec: &ChainSpec,
         block_count: u64,
         temp_dir: &Path,
-    ) -> (Vec<SealedBlock>, PathBuf) {
+    ) -> (Vec<SealedBlock<Block>>, PathBuf) {
         let test_blocks = generate_test_blocks(chain_spec, block_count);
         assert_eq!(
             test_blocks.len(),

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

@@ -6,14 +6,13 @@ use alloy_primitives::{Address, B256, B64, U256};
 use alloy_rlp::Encodable;
 use reth_chainspec::{ChainSpec, EthereumHardforks};
 use reth_ethereum_primitives::{Block, BlockBody};
-use reth_primitives::SealedBlock;
-use reth_primitives_traits::Block as BlockTrait;
+use reth_primitives_traits::{Block as BlockTrait, SealedBlock};
 use std::{io::Write, path::Path};
 use tracing::debug;
 
 /// Generate test blocks for a given chain spec
-pub fn generate_test_blocks(chain_spec: &ChainSpec, count: u64) -> Vec<SealedBlock> {
-    let mut blocks: Vec<SealedBlock> = Vec::new();
+pub fn generate_test_blocks(chain_spec: &ChainSpec, count: u64) -> Vec<SealedBlock<Block>> {
+    let mut blocks: Vec<SealedBlock<Block>> = Vec::new();
     let genesis_header = chain_spec.sealed_genesis_header();
     let mut parent_hash = genesis_header.hash();
     let mut parent_number = genesis_header.number();
@@ -139,7 +138,7 @@ pub fn generate_test_blocks(chain_spec: &ChainSpec, count: u64) -> Vec<SealedBlo
 }
 
 /// Write blocks to RLP file
-pub fn write_blocks_to_rlp(blocks: &[SealedBlock], path: &Path) -> std::io::Result<()> {
+pub fn write_blocks_to_rlp(blocks: &[SealedBlock<Block>], path: &Path) -> std::io::Result<()> {
     let mut file = std::fs::File::create(path)?;
     let mut total_bytes = 0;
 
@@ -173,7 +172,7 @@ pub fn write_blocks_to_rlp(blocks: &[SealedBlock], path: &Path) -> std::io::Resu
 }
 
 /// Create FCU JSON for the tip of the chain
-pub fn create_fcu_json(tip: &SealedBlock) -> serde_json::Value {
+pub fn create_fcu_json(tip: &SealedBlock<Block>) -> serde_json::Value {
     serde_json::json!({
         "params": [{
             "headBlockHash": format!("0x{:x}", tip.hash()),

crates/engine/primitives/src/config.rs

@@ -9,32 +9,8 @@ pub const DEFAULT_PERSISTENCE_THRESHOLD: u64 = 2;
 /// How close to the canonical head we persist blocks.
 pub const DEFAULT_MEMORY_BLOCK_BUFFER_TARGET: u64 = 0;
 
-/// Returns the default number of storage worker threads based on available parallelism.
-fn default_storage_worker_count() -> usize {
-    #[cfg(feature = "std")]
-    {
-        std::thread::available_parallelism().map_or(8, |n| n.get() * 2)
-    }
-    #[cfg(not(feature = "std"))]
-    {
-        8
-    }
-}
-
-/// Returns the default number of account worker threads.
-///
-/// Account workers coordinate storage proof collection and account trie traversal.
-/// They are set to the same count as storage workers for simplicity.
-fn default_account_worker_count() -> usize {
-    default_storage_worker_count()
-}
-
 /// The size of proof targets chunk to spawn in one multiproof calculation.
-pub const DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE: usize = 60;
-
-/// The size of proof targets chunk optimized for small blocks (≤20M gas used).
-/// Benchmarks: <https://gist.github.com/yongkangc/fda9c24846f0ba891376bcf81b002008>
-pub const SMALL_BLOCK_MULTIPROOF_CHUNK_SIZE: usize = 30;
+pub const DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE: usize = 5;
 
 /// Gas threshold below which the small block chunk size is used.
 pub const SMALL_BLOCK_GAS_THRESHOLD: u64 = 20_000_000;
@@ -127,8 +103,6 @@ pub struct TreeConfig {
     cross_block_cache_size: usize,
     /// Whether the host has enough parallelism to run state root task.
     has_enough_parallelism: bool,
-    /// Whether multiproof task should chunk proof targets.
-    multiproof_chunking_enabled: bool,
     /// Multiproof task chunk size for proof targets.
     multiproof_chunk_size: usize,
     /// Number of reserved CPU cores for non-reth processes
@@ -153,10 +127,6 @@ pub struct TreeConfig {
     always_process_payload_attributes_on_canonical_head: bool,
     /// Whether to unwind canonical header to ancestor during forkchoice updates.
     allow_unwind_canonical_header: bool,
-    /// Number of storage proof worker threads.
-    storage_worker_count: usize,
-    /// Number of account proof worker threads.
-    account_worker_count: usize,
     /// Whether to disable cache metrics recording (can be expensive with large cached state).
     disable_cache_metrics: bool,
     /// Depth for sparse trie pruning after state root computation.
@@ -187,15 +157,12 @@ impl Default for TreeConfig {
             state_provider_metrics: false,
             cross_block_cache_size: DEFAULT_CROSS_BLOCK_CACHE_SIZE,
             has_enough_parallelism: has_enough_parallelism(),
-            multiproof_chunking_enabled: true,
             multiproof_chunk_size: DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE,
             reserved_cpu_cores: DEFAULT_RESERVED_CPU_CORES,
             precompile_cache_disabled: false,
             state_root_fallback: false,
             always_process_payload_attributes_on_canonical_head: false,
             allow_unwind_canonical_header: false,
-            storage_worker_count: default_storage_worker_count(),
-            account_worker_count: default_account_worker_count(),
             disable_cache_metrics: false,
             sparse_trie_prune_depth: DEFAULT_SPARSE_TRIE_PRUNE_DEPTH,
             sparse_trie_max_storage_tries: DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES,
@@ -221,15 +188,12 @@ impl TreeConfig {
         state_provider_metrics: bool,
         cross_block_cache_size: usize,
         has_enough_parallelism: bool,
-        multiproof_chunking_enabled: bool,
         multiproof_chunk_size: usize,
         reserved_cpu_cores: usize,
         precompile_cache_disabled: bool,
         state_root_fallback: bool,
         always_process_payload_attributes_on_canonical_head: bool,
         allow_unwind_canonical_header: bool,
-        storage_worker_count: usize,
-        account_worker_count: usize,
         disable_cache_metrics: bool,
         sparse_trie_prune_depth: usize,
         sparse_trie_max_storage_tries: usize,
@@ -248,15 +212,12 @@ impl TreeConfig {
             state_provider_metrics,
             cross_block_cache_size,
             has_enough_parallelism,
-            multiproof_chunking_enabled,
             multiproof_chunk_size,
             reserved_cpu_cores,
             precompile_cache_disabled,
             state_root_fallback,
             always_process_payload_attributes_on_canonical_head,
             allow_unwind_canonical_header,
-            storage_worker_count,
-            account_worker_count,
             disable_cache_metrics,
             sparse_trie_prune_depth,
             sparse_trie_max_storage_tries,
@@ -290,11 +251,6 @@ impl TreeConfig {
         self.max_execute_block_batch_size
     }
 
-    /// Return whether the multiproof task chunking is enabled.
-    pub const fn multiproof_chunking_enabled(&self) -> bool {
-        self.multiproof_chunking_enabled
-    }
-
     /// Return the multiproof task chunk size.
     pub const fn multiproof_chunk_size(&self) -> usize {
         self.multiproof_chunk_size
@@ -458,15 +414,6 @@ impl TreeConfig {
         self
     }
 
-    /// Setter for whether multiproof task should chunk proof targets.
-    pub const fn with_multiproof_chunking_enabled(
-        mut self,
-        multiproof_chunking_enabled: bool,
-    ) -> Self {
-        self.multiproof_chunking_enabled = multiproof_chunking_enabled;
-        self
-    }
-
     /// Setter for multiproof task chunk size for proof targets.
     pub const fn with_multiproof_chunk_size(mut self, multiproof_chunk_size: usize) -> Self {
         self.multiproof_chunk_size = multiproof_chunk_size;
@@ -502,42 +449,6 @@ impl TreeConfig {
         self.has_enough_parallelism && !self.legacy_state_root
     }
 
-    /// Return the number of storage proof worker threads.
-    pub const fn storage_worker_count(&self) -> usize {
-        self.storage_worker_count
-    }
-
-    /// Setter for the number of storage proof worker threads.
-    ///
-    /// No-op if it's [`None`].
-    pub const fn with_storage_worker_count_opt(
-        mut self,
-        storage_worker_count: Option<usize>,
-    ) -> Self {
-        if let Some(count) = storage_worker_count {
-            self.storage_worker_count = count;
-        }
-        self
-    }
-
-    /// Return the number of account proof worker threads.
-    pub const fn account_worker_count(&self) -> usize {
-        self.account_worker_count
-    }
-
-    /// Setter for the number of account proof worker threads.
-    ///
-    /// No-op if it's [`None`].
-    pub const fn with_account_worker_count_opt(
-        mut self,
-        account_worker_count: Option<usize>,
-    ) -> Self {
-        if let Some(count) = account_worker_count {
-            self.account_worker_count = count;
-        }
-        self
-    }
-
     /// Returns whether cache metrics recording is disabled.
     pub const fn disable_cache_metrics(&self) -> bool {
         self.disable_cache_metrics

crates/engine/tree/benches/state_root_task.rs

@@ -12,7 +12,8 @@ use rand::Rng;
 use reth_chainspec::ChainSpec;
 use reth_db_common::init::init_genesis;
 use reth_engine_tree::tree::{
-    precompile_cache::PrecompileCacheMap, PayloadProcessor, StateProviderBuilder, TreeConfig,
+    precompile_cache::PrecompileCacheMap, ExecutionEnv, PayloadProcessor, StateProviderBuilder,
+    TreeConfig,
 };
 use reth_ethereum_primitives::TransactionSigned;
 use reth_evm::OnStateHook;
@@ -230,7 +231,7 @@ fn bench_state_root(c: &mut Criterion) {
                     |(genesis_hash, mut payload_processor, provider, state_updates)| {
                         black_box({
                             let mut handle = payload_processor.spawn(
-                                Default::default(),
+                                ExecutionEnv::test_default(),
                                 (
                                     Vec::<
                                         Result<

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

@@ -351,14 +351,6 @@ impl<S: StateProvider, const PREWARM: bool> StateProvider for CachedStateProvide
             self.state_provider.storage(account, storage_key)
         }
     }
-
-    fn storage_by_hashed_key(
-        &self,
-        address: Address,
-        hashed_storage_key: StorageKey,
-    ) -> ProviderResult<Option<StorageValue>> {
-        self.state_provider.storage_by_hashed_key(address, hashed_storage_key)
-    }
 }
 
 impl<S: BytecodeReader, const PREWARM: bool> BytecodeReader for CachedStateProvider<S, PREWARM> {

crates/engine/tree/src/tree/instrumented_state.rs

@@ -199,17 +199,6 @@ impl<S: StateProvider> StateProvider for InstrumentedStateProvider<S> {
         self.record_storage_fetch(start.elapsed());
         res
     }
-
-    fn storage_by_hashed_key(
-        &self,
-        address: Address,
-        hashed_storage_key: StorageKey,
-    ) -> ProviderResult<Option<StorageValue>> {
-        let start = Instant::now();
-        let res = self.state_provider.storage_by_hashed_key(address, hashed_storage_key);
-        self.record_storage_fetch(start.elapsed());
-        res
-    }
 }
 
 impl<S: BytecodeReader> BytecodeReader for InstrumentedStateProvider<S> {

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

@@ -38,7 +38,7 @@ use reth_provider::{
 };
 use reth_revm::database::StateProviderDatabase;
 use reth_stages_api::ControlFlow;
-use reth_tasks::spawn_os_thread;
+use reth_tasks::{spawn_os_thread, utils::increase_thread_priority};
 use reth_trie_db::ChangesetCache;
 use revm::interpreter::debug_unreachable;
 use state::TreeState;
@@ -420,7 +420,10 @@ where
             use_hashed_state,
         );
         let incoming = task.incoming_tx.clone();
-        spawn_os_thread("engine", || task.run());
+        spawn_os_thread("engine", || {
+            increase_thread_priority();
+            task.run()
+        });
         (incoming, outgoing)
     }
 
@@ -1413,7 +1416,7 @@ where
         // Spawn a background task to trigger computation so it's ready when the next payload
         // arrives.
         if let Some(overlay) = self.state.tree_state.prepare_canonical_overlay() {
-            rayon::spawn(move || {
+            tokio::task::spawn_blocking(move || {
                 let _ = overlay.get();
             });
         }
@@ -2580,6 +2583,51 @@ where
         Some(TreeEvent::Download(request))
     }
 
+    /// Handles a downloaded block that was successfully inserted as valid.
+    ///
+    /// If the block matches the sync target head, returns [`TreeAction::MakeCanonical`].
+    /// If it matches a non-head sync target (safe or finalized), makes it canonical inline
+    /// and triggers a download for the remaining blocks towards the actual head.
+    /// Otherwise, tries to connect buffered blocks.
+    fn on_valid_downloaded_block(
+        &mut self,
+        block_num_hash: BlockNumHash,
+    ) -> Result<Option<TreeEvent>, InsertBlockFatalError> {
+        // check if we just inserted a block that's part of sync targets,
+        // i.e. head, safe, or finalized
+        if let Some(sync_target) = self.state.forkchoice_state_tracker.sync_target_state() &&
+            sync_target.contains(block_num_hash.hash)
+        {
+            debug!(target: "engine::tree", ?sync_target, "appended downloaded sync target block");
+
+            if sync_target.head_block_hash == block_num_hash.hash {
+                // we just inserted the sync target head block, make it canonical
+                return Ok(Some(TreeEvent::TreeAction(TreeAction::MakeCanonical {
+                    sync_target_head: block_num_hash.hash,
+                })))
+            }
+
+            // This block is part of the sync target (safe or finalized) but not the
+            // head. Make it canonical and try to connect any buffered children, then
+            // continue downloading towards the actual head if needed.
+            self.make_canonical(block_num_hash.hash)?;
+            self.try_connect_buffered_blocks(block_num_hash)?;
+
+            // Check if we've reached the sync target head after connecting buffered
+            // blocks (e.g. the head block may have already been buffered).
+            if self.state.tree_state.canonical_block_hash() != sync_target.head_block_hash {
+                let target = self.lowest_buffered_ancestor_or(sync_target.head_block_hash);
+                trace!(target: "engine::tree", %target, "sync target head not yet reached, downloading head block");
+                return Ok(Some(TreeEvent::Download(DownloadRequest::single_block(target))))
+            }
+
+            return Ok(None)
+        }
+        trace!(target: "engine::tree", "appended downloaded block");
+        self.try_connect_buffered_blocks(block_num_hash)?;
+        Ok(None)
+    }
+
     /// Invoked with a block downloaded from the network
     ///
     /// Returns an event with the appropriate action to take, such as:
@@ -2602,22 +2650,11 @@ where
 
         // try to append the block
         match self.insert_block(block) {
-            Ok(InsertPayloadOk::Inserted(BlockStatus::Valid)) => {
-                // check if we just inserted a block that's part of sync targets,
-                // i.e. head, safe, or finalized
-                if let Some(sync_target) = self.state.forkchoice_state_tracker.sync_target_state() &&
-                    sync_target.contains(block_num_hash.hash)
-                {
-                    debug!(target: "engine::tree", ?sync_target, "appended downloaded sync target block");
-
-                    // we just inserted a block that we know is part of the canonical chain, so we
-                    // can make it canonical
-                    return Ok(Some(TreeEvent::TreeAction(TreeAction::MakeCanonical {
-                        sync_target_head: block_num_hash.hash,
-                    })))
-                }
-                trace!(target: "engine::tree", "appended downloaded block");
-                self.try_connect_buffered_blocks(block_num_hash)?;
+            Ok(
+                InsertPayloadOk::Inserted(BlockStatus::Valid) |
+                InsertPayloadOk::AlreadySeen(BlockStatus::Valid),
+            ) => {
+                return self.on_valid_downloaded_block(block_num_hash);
             }
             Ok(InsertPayloadOk::Inserted(BlockStatus::Disconnected { head, missing_ancestor })) => {
                 // block is not connected to the canonical head, we need to download

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

@@ -20,7 +20,6 @@ use multiproof::*;
 use parking_lot::RwLock;
 use prewarm::PrewarmMetrics;
 use rayon::prelude::*;
-use reth_engine_primitives::{SMALL_BLOCK_GAS_THRESHOLD, SMALL_BLOCK_MULTIPROOF_CHUNK_SIZE};
 use reth_evm::{
     block::ExecutableTxParts,
     execute::{ExecutableTxFor, WithTxEnv},
@@ -33,7 +32,7 @@ use reth_provider::{
     BlockExecutionOutput, BlockReader, DatabaseProviderROFactory, StateProviderFactory, StateReader,
 };
 use reth_revm::{db::BundleState, state::EvmState};
-use reth_tasks::{ForEachOrdered, Runtime};
+use reth_tasks::{utils::increase_thread_priority, ForEachOrdered, Runtime};
 use reth_trie::{hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory};
 use reth_trie_parallel::{
     proof_task::{ProofTaskCtx, ProofWorkerHandle},
@@ -287,7 +286,7 @@ where
 
         let parent_state_root = env.parent_state_root;
         let transaction_count = env.transaction_count;
-        let chunk_size = Self::adaptive_chunk_size(config, env.gas_used);
+        let chunk_size = config.multiproof_chunk_size();
         let prewarm_handle = self.spawn_caching_with(
             env,
             prewarm_rx,
@@ -369,24 +368,6 @@ where
     /// waiting for rayon scheduling.
     const PARALLEL_PREFETCH_COUNT: usize = 4;
 
-    /// Returns the multiproof chunk size adapted to the block's gas usage.
-    ///
-    /// For blocks with ≤20M gas used, a smaller chunk size (30) yields better throughput.
-    /// For larger blocks, the configured default chunk size is used.
-    const fn adaptive_chunk_size(config: &TreeConfig, gas_used: u64) -> Option<usize> {
-        if !config.multiproof_chunking_enabled() {
-            return None;
-        }
-
-        let size = if gas_used > 0 && gas_used <= SMALL_BLOCK_GAS_THRESHOLD {
-            SMALL_BLOCK_MULTIPROOF_CHUNK_SIZE
-        } else {
-            config.multiproof_chunk_size()
-        };
-
-        Some(size)
-    }
-
     /// Spawns a task advancing transaction env iterator and streaming updates through a channel.
     ///
     /// For blocks with fewer than [`Self::SMALL_BLOCK_TX_THRESHOLD`] transactions, uses
@@ -416,9 +397,7 @@ where
                 transaction_count,
                 "using sequential sig recovery for small block"
             );
-            self.executor.spawn_blocking(move || {
-                let _enter =
-                    debug_span!(target: "engine::tree::payload_processor", "tx_iterator").entered();
+            self.executor.spawn_blocking_named("tx-iterator", move || {
                 let (transactions, convert) = transactions.into_parts();
                 convert_serial(transactions.into_iter(), &convert, &prewarm_tx, &execute_tx);
             });
@@ -430,9 +409,7 @@ where
             // few transactions are recovered sequentially and sent immediately before
             // entering the parallel iterator for the remainder.
             let prefetch = Self::PARALLEL_PREFETCH_COUNT.min(transaction_count);
-            self.executor.spawn_blocking(move || {
-                let _enter =
-                    debug_span!(target: "engine::tree::payload_processor", "tx_iterator").entered();
+            self.executor.spawn_blocking_named("tx-iterator", move || {
                 let (transactions, convert) = transactions.into_parts();
                 let mut all: Vec<_> = transactions.into_iter().collect();
                 let rest = all.split_off(prefetch.min(all.len()));
@@ -447,15 +424,17 @@ where
                     .map(|(i, tx)| {
                         let idx = i + prefetch;
                         let tx = convert.convert(tx);
-                        tx.map(|tx| {
+                        let tx = tx.map(|tx| {
                             let (tx_env, tx) = tx.into_parts();
                             let tx = WithTxEnv { tx_env, tx: Arc::new(tx) };
                             let _ = prewarm_tx.send((idx, tx.clone()));
                             tx
-                        })
+                        });
+                        (idx, tx)
                     })
-                    .for_each_ordered(|tx| {
+                    .for_each_ordered(|(idx, tx)| {
                         let _ = execute_tx.send(tx);
+                        debug!(target: "engine::tree::payload_processor", idx, "yielded transaction");
                     });
             });
         }
@@ -551,7 +530,7 @@ where
         state_root_tx: mpsc::Sender<Result<StateRootComputeOutcome, ParallelStateRootError>>,
         from_multi_proof: CrossbeamReceiver<MultiProofMessage>,
         parent_state_root: B256,
-        chunk_size: Option<usize>,
+        chunk_size: usize,
     ) {
         let preserved_sparse_trie = self.sparse_state_trie.clone();
         let trie_metrics = self.trie_metrics.clone();
@@ -562,6 +541,8 @@ where
 
         let parent_span = Span::current();
         self.executor.spawn_blocking_named("sparse-trie", move || {
+            increase_thread_priority();
+
             let _enter = debug_span!(target: "engine::tree::payload_processor", parent: parent_span, "sparse_trie_task")
                 .entered();
 
@@ -603,8 +584,6 @@ where
             );
 
             let result = task.run();
-            // Capture the computed state_root before sending the result
-            let computed_state_root = result.as_ref().ok().map(|outcome| outcome.state_root);
 
             // Acquire the guard before sending the result to prevent a race condition:
             // Without this, the next block could start after send() but before store(),
@@ -613,6 +592,7 @@ where
             // block's take() blocks until we've stored the trie for reuse.
             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()
             if state_root_tx.send(result).is_err() {
                 // Receiver dropped - payload was likely invalid or cancelled.
@@ -636,7 +616,7 @@ where
             // A failed computation may have left the trie in a partially updated state.
             let _enter =
                 debug_span!(target: "engine::tree::payload_processor", "preserve").entered();
-            let deferred = if let Some(state_root) = computed_state_root {
+            let deferred = if let Some(result) = task_result {
                 let start = Instant::now();
                 let (trie, deferred) = task.into_trie_for_reuse(
                     prune_depth,
@@ -644,11 +624,12 @@ where
                     SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
                     SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
                     disable_cache_pruning,
+                    &result.trie_updates,
                 );
                 trie_metrics
                     .into_trie_for_reuse_duration_histogram
                     .record(start.elapsed().as_secs_f64());
-                guard.store(PreservedSparseTrie::anchored(trie, state_root));
+                guard.store(PreservedSparseTrie::anchored(trie, result.state_root));
                 deferred
             } else {
                 debug!(
@@ -740,6 +721,7 @@ fn convert_serial<RawTx, Tx, TxEnv, InnerTx, Recovered, Err, C>(
             let _ = prewarm_tx.send((idx, tx.clone()));
         }
         let _ = execute_tx.send(tx);
+        debug!(target: "engine::tree::payload_processor", idx, "yielded transaction");
     }
 }
 
@@ -840,9 +822,7 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
 
     /// Returns iterator yielding transactions from the stream.
     pub fn iter_transactions(&mut self) -> impl Iterator<Item = Result<Tx, Err>> + '_ {
-        core::iter::repeat_with(|| self.transactions.recv())
-            .take_while(|res| res.is_ok())
-            .map(|res| res.unwrap())
+        self.transactions.iter()
     }
 }
 
@@ -1074,11 +1054,13 @@ pub struct ExecutionEnv<Evm: ConfigureEvm> {
     pub withdrawals: Option<Vec<Withdrawal>>,
 }
 
-impl<Evm: ConfigureEvm> Default for ExecutionEnv<Evm>
+impl<Evm: ConfigureEvm> ExecutionEnv<Evm>
 where
     EvmEnvFor<Evm>: Default,
 {
-    fn default() -> Self {
+    /// Creates a new [`ExecutionEnv`] with default values for testing.
+    #[cfg(any(test, feature = "test-utils"))]
+    pub fn test_default() -> Self {
         Self {
             evm_env: Default::default(),
             hash: Default::default(),
@@ -1096,7 +1078,7 @@ mod tests {
     use super::PayloadExecutionCache;
     use crate::tree::{
         cached_state::{CachedStateMetrics, ExecutionCache, SavedCache},
-        payload_processor::{evm_state_to_hashed_post_state, PayloadProcessor},
+        payload_processor::{evm_state_to_hashed_post_state, ExecutionEnv, PayloadProcessor},
         precompile_cache::PrecompileCacheMap,
         StateProviderBuilder, TreeConfig,
     };
@@ -1371,7 +1353,7 @@ mod tests {
         let provider_factory = BlockchainProvider::new(factory).unwrap();
 
         let mut handle = payload_processor.spawn(
-            Default::default(),
+            ExecutionEnv::test_default(),
             (
                 Vec::<Result<Recovered<TransactionSigned>, core::convert::Infallible>>::new(),
                 std::convert::identity,

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

@@ -162,6 +162,14 @@ pub(crate) struct MultiProofTaskMetrics {
 
     /// Histogram of sparse trie update durations.
     pub sparse_trie_update_duration_histogram: Histogram,
+    /// Histogram of durations spent revealing multiproof results into the sparse trie.
+    pub sparse_trie_reveal_multiproof_duration_histogram: Histogram,
+    /// Histogram of durations spent coalescing multiple proof results from the channel.
+    pub sparse_trie_proof_coalesce_duration_histogram: Histogram,
+    /// Histogram of durations the event loop spent blocked waiting on channels.
+    pub sparse_trie_channel_wait_duration_histogram: Histogram,
+    /// Histogram of durations spent processing trie updates and promoting pending accounts.
+    pub sparse_trie_process_updates_duration_histogram: Histogram,
     /// Histogram of sparse trie final update durations.
     pub sparse_trie_final_update_duration_histogram: Histogram,
     /// Histogram of sparse trie total durations.
@@ -189,7 +197,7 @@ pub(crate) struct MultiProofTaskMetrics {
 pub(crate) fn dispatch_with_chunking<T, I>(
     items: T,
     chunking_len: usize,
-    chunk_size: Option<usize>,
+    chunk_size: usize,
     max_targets_for_chunking: usize,
     available_account_workers: usize,
     available_storage_workers: usize,
@@ -203,10 +211,7 @@ where
         available_account_workers > 1 ||
         available_storage_workers > 1;
 
-    if should_chunk &&
-        let Some(chunk_size) = chunk_size &&
-        chunking_len > chunk_size
-    {
+    if should_chunk && chunking_len > chunk_size {
         let mut num_chunks = 0usize;
         for chunk in chunker(items, chunk_size) {
             dispatch(chunk);

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

@@ -20,9 +20,8 @@ use crate::tree::{
 use alloy_consensus::transaction::TxHashRef;
 use alloy_eip7928::BlockAccessList;
 use alloy_eips::eip4895::Withdrawal;
-use alloy_evm::Database;
 use alloy_primitives::{keccak256, StorageKey, B256};
-use crossbeam_channel::{Receiver as CrossbeamReceiver, Sender as CrossbeamSender};
+use crossbeam_channel::Sender as CrossbeamSender;
 use metrics::{Counter, Gauge, Histogram};
 use rayon::prelude::*;
 use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, RecoveredTx, SpecFor};
@@ -33,11 +32,11 @@ use reth_provider::{
     StateReader,
 };
 use reth_revm::{database::StateProviderDatabase, state::EvmState};
-use reth_tasks::Runtime;
+use reth_tasks::{pool::WorkerPool, Runtime};
 use reth_trie_parallel::targets_v2::MultiProofTargetsV2;
 use std::sync::{
     atomic::{AtomicBool, Ordering},
-    mpsc::{self, channel, Receiver, Sender, SyncSender},
+    mpsc::{self, channel, Receiver, Sender},
     Arc,
 };
 use tracing::{debug, debug_span, instrument, trace, warn, Span};
@@ -54,15 +53,6 @@ pub enum PrewarmMode<Tx> {
     Skipped,
 }
 
-/// A wrapper for transactions that includes their index in the block.
-#[derive(Clone)]
-struct IndexedTransaction<Tx> {
-    /// The transaction index in the block.
-    index: usize,
-    /// The wrapped transaction.
-    tx: Tx,
-}
-
 /// A task that is responsible for caching and prewarming the cache by executing transactions
 /// individually in parallel.
 ///
@@ -122,82 +112,141 @@ where
         )
     }
 
-    /// Spawns all pending transactions as blocking tasks by first chunking them.
+    /// Streams pending transactions and executes them in parallel on the prewarming pool.
     ///
-    /// For Optimism chains, special handling is applied to the first transaction if it's a
-    /// deposit transaction (type 0x7E/126) which sets critical metadata that affects all
-    /// subsequent transactions in the block.
-    fn spawn_all<Tx>(
+    /// Kicks off EVM init on every pool thread, then uses `in_place_scope` to dispatch
+    /// transactions as they arrive and wait for all spawned tasks to complete before
+    /// clearing per-thread state. Workers that start via work-stealing lazily initialise
+    /// their EVM state on first access via [`get_or_init`](reth_tasks::pool::Worker::get_or_init).
+    fn spawn_txs_prewarm<Tx>(
         &self,
         pending: mpsc::Receiver<(usize, Tx)>,
         actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
     ) where
-        Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
+        Tx: ExecutableTxFor<Evm> + Send + 'static,
     {
         let executor = self.executor.clone();
         let ctx = self.ctx.clone();
         let span = Span::current();
 
-        self.executor.spawn_blocking_named("prewarm-spawn", move || {
-            let _enter = debug_span!(target: "engine::tree::payload_processor::prewarm", parent: span, "spawn_all").entered();
+        self.executor.spawn_blocking_named("prewarm-txs", move || {
+            let _enter = debug_span!(
+                target: "engine::tree::payload_processor::prewarm",
+                parent: span,
+                "prewarm_txs"
+            )
+            .entered();
 
-            let pool_threads = executor.prewarming_pool().current_num_threads();
-            // Don't spawn more workers than transactions. When transaction_count is 0
-            // (unknown), use all pool threads.
-            let workers_needed = if ctx.env.transaction_count > 0 {
-                ctx.env.transaction_count.min(pool_threads)
-            } else {
-                pool_threads
-            };
+            let ctx = &ctx;
+            let pool = executor.prewarming_pool();
 
-            let (done_tx, done_rx) = mpsc::sync_channel(workers_needed);
-
-            // Spawn workers
-            let tx_sender = ctx.clone().spawn_workers(workers_needed, &executor,  to_multi_proof.clone(), done_tx.clone());
-
-            // Distribute transactions to workers
             let mut tx_count = 0usize;
-            while let Ok((tx_index, tx)) = pending.recv() {
-                // Stop distributing if termination was requested
-                if ctx.terminate_execution.load(Ordering::Relaxed) {
-                    trace!(
-                        target: "engine::tree::payload_processor::prewarm",
-                        "Termination requested, stopping transaction distribution"
-                    );
-                    break;
+            let to_multi_proof = to_multi_proof.as_ref();
+            pool.in_place_scope(|s| {
+                s.spawn(|_| {
+                    pool.init::<PrewarmEvmState<Evm>>(|_| ctx.evm_for_ctx());
+                });
+
+                while let Ok((index, tx)) = pending.recv() {
+                    if ctx.terminate_execution.load(Ordering::Relaxed) {
+                        trace!(
+                            target: "engine::tree::payload_processor::prewarm",
+                            "Termination requested, stopping transaction distribution"
+                        );
+                        break;
+                    }
+
+                    tx_count += 1;
+                    let parent_span = Span::current();
+                    s.spawn(move |_| {
+                        let _enter = debug_span!(
+                            target: "engine::tree::payload_processor::prewarm",
+                            parent: parent_span,
+                            "prewarm_tx",
+                            i = index,
+                        )
+                        .entered();
+                        Self::transact_worker(ctx, index, tx, to_multi_proof);
+                    });
                 }
 
-                let indexed_tx = IndexedTransaction { index: tx_index, tx };
+                // Send withdrawal prefetch targets after all transactions dispatched
+                if let Some(to_multi_proof) = to_multi_proof &&
+                    let Some(withdrawals) = &ctx.env.withdrawals &&
+                    !withdrawals.is_empty()
+                {
+                    let targets = multiproof_targets_from_withdrawals(withdrawals);
+                    let _ = to_multi_proof.send(MultiProofMessage::PrefetchProofs(targets));
+                }
+            });
 
-                // Send transaction to the workers
-                // Ignore send errors: workers listen to terminate_execution and may
-                // exit early when signaled.
-                let _ = tx_sender.send(indexed_tx);
-
-                tx_count += 1;
-            }
-
-            // Send withdrawal prefetch targets after all transactions have been distributed
-            if let Some(to_multi_proof) = to_multi_proof
-                && let Some(withdrawals) = &ctx.env.withdrawals
-                && !withdrawals.is_empty()
-            {
-                let targets = multiproof_targets_from_withdrawals(withdrawals);
-                let _ = to_multi_proof
-                    .send(MultiProofMessage::PrefetchProofs(targets));
-            }
-
-            // drop sender and wait for all tasks to finish
-            drop(done_tx);
-            drop(tx_sender);
-            while done_rx.recv().is_ok() {}
+            // All tasks are done — clear per-thread EVM state for the next block.
+            pool.clear();
 
             let _ = actions_tx
                 .send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: tx_count });
         });
     }
 
+    /// Executes a single prewarm transaction on the current pool thread's EVM.
+    ///
+    /// Lazily initialises per-thread [`PrewarmEvmState`] via
+    /// [`get_or_init`](reth_tasks::pool::Worker::get_or_init) on first access.
+    fn transact_worker<Tx>(
+        ctx: &PrewarmContext<N, P, Evm>,
+        index: usize,
+        tx: Tx,
+        to_multi_proof: Option<&CrossbeamSender<MultiProofMessage>>,
+    ) where
+        Tx: ExecutableTxFor<Evm>,
+    {
+        WorkerPool::with_worker_mut(|worker| {
+            let Some((evm, metrics, terminate_execution)) =
+                worker.get_or_init::<PrewarmEvmState<Evm>>(|| ctx.evm_for_ctx()).as_mut()
+            else {
+                return;
+            };
+
+            if terminate_execution.load(Ordering::Relaxed) {
+                return;
+            }
+
+            let start = Instant::now();
+
+            let (tx_env, tx) = tx.into_parts();
+            let res = match evm.transact(tx_env) {
+                Ok(res) => res,
+                Err(err) => {
+                    trace!(
+                        target: "engine::tree::payload_processor::prewarm",
+                        %err,
+                        tx_hash=%tx.tx().tx_hash(),
+                        sender=%tx.signer(),
+                        "Error when executing prewarm transaction",
+                    );
+                    metrics.transaction_errors.increment(1);
+                    return;
+                }
+            };
+            metrics.execution_duration.record(start.elapsed());
+
+            if terminate_execution.load(Ordering::Relaxed) {
+                return;
+            }
+
+            if index > 0 {
+                let (targets, storage_targets) = multiproof_targets_from_state(res.state);
+                metrics.prefetch_storage_targets.record(storage_targets as f64);
+                if let Some(to_multi_proof) = to_multi_proof {
+                    let _ = to_multi_proof.send(MultiProofMessage::PrefetchProofs(targets));
+                }
+            }
+
+            metrics.total_runtime.record(start.elapsed());
+        });
+    }
+
     /// This method calls `ExecutionCache::update_with_guard` which requires exclusive access.
     /// It should only be called after ensuring that:
     /// 1. All prewarming tasks have completed execution
@@ -370,12 +419,12 @@ where
     )]
     pub fn run<Tx>(self, mode: PrewarmMode<Tx>, actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>)
     where
-        Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
+        Tx: ExecutableTxFor<Evm> + Send + 'static,
     {
         // Spawn execution tasks based on mode
         match mode {
             PrewarmMode::Transactions(pending) => {
-                self.spawn_all(pending, actions_tx, self.to_multi_proof.clone());
+                self.spawn_txs_prewarm(pending, actions_tx, self.to_multi_proof.clone());
             }
             PrewarmMode::BlockAccessList(bal) => {
                 self.run_bal_prewarm(bal, actions_tx);
@@ -454,27 +503,24 @@ where
     pub precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
 }
 
+/// Per-thread EVM state initialised by [`PrewarmContext::evm_for_ctx`] and stored in
+/// [`WorkerPool`] workers via [`Worker::get_or_init`](reth_tasks::pool::Worker::get_or_init).
+type PrewarmEvmState<Evm> = Option<(
+    EvmFor<Evm, StateProviderDatabase<reth_provider::StateProviderBox>>,
+    PrewarmMetrics,
+    Arc<AtomicBool>,
+)>;
+
 impl<N, P, Evm> PrewarmContext<N, P, Evm>
 where
     N: NodePrimitives,
     P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
     Evm: ConfigureEvm<Primitives = N> + 'static,
 {
-    /// Splits this context into an evm, metrics, and the atomic bool for terminating execution.
+    /// Creates a per-thread EVM, metrics handle, and termination flag for prewarming.
     #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
-    fn evm_for_ctx(self) -> Option<(EvmFor<Evm, impl Database>, PrewarmMetrics, Arc<AtomicBool>)> {
-        let Self {
-            env,
-            evm_config,
-            saved_cache,
-            provider,
-            metrics,
-            terminate_execution,
-            precompile_cache_disabled,
-            precompile_cache_map,
-        } = self;
-
-        let mut state_provider = match provider.build() {
+    fn evm_for_ctx(&self) -> PrewarmEvmState<Evm> {
+        let mut state_provider = match self.provider.build() {
             Ok(provider) => provider,
             Err(err) => {
                 trace!(
@@ -487,7 +533,7 @@ where
         };
 
         // Use the caches to create a new provider with caching
-        if let Some(saved_cache) = saved_cache {
+        if let Some(saved_cache) = &self.saved_cache {
             let caches = saved_cache.cache().clone();
             let cache_metrics = saved_cache.metrics().clone();
             state_provider =
@@ -496,7 +542,7 @@ where
 
         let state_provider = StateProviderDatabase::new(state_provider);
 
-        let mut evm_env = env.evm_env;
+        let mut evm_env = self.env.evm_env.clone();
 
         // we must disable the nonce check so that we can execute the transaction even if the nonce
         // doesn't match what's on chain.
@@ -508,130 +554,21 @@ where
 
         // create a new executor and disable nonce checks in the env
         let spec_id = *evm_env.spec_id();
-        let mut evm = evm_config.evm_with_env(state_provider, evm_env);
+        let mut evm = self.evm_config.evm_with_env(state_provider, evm_env);
 
-        if !precompile_cache_disabled {
+        if !self.precompile_cache_disabled {
             // Only cache pure precompiles to avoid issues with stateful precompiles
             evm.precompiles_mut().map_pure_precompiles(|address, precompile| {
                 CachedPrecompile::wrap(
                     precompile,
-                    precompile_cache_map.cache_for_address(*address),
+                    self.precompile_cache_map.cache_for_address(*address),
                     spec_id,
                     None, // No metrics for prewarm
                 )
             });
         }
 
-        Some((evm, metrics, terminate_execution))
-    }
-
-    /// Accepts a [`CrossbeamReceiver`] of transactions and a handle to prewarm task. Executes
-    /// transactions and streams [`MultiProofMessage::PrefetchProofs`] messages for each
-    /// transaction.
-    ///
-    /// This function processes transactions sequentially from the receiver and emits outcome events
-    /// via the provided sender. Execution errors are logged and tracked but do not stop the batch
-    /// processing unless the task is explicitly cancelled.
-    ///
-    /// Note: There are no ordering guarantees; this does not reflect the state produced by
-    /// sequential execution.
-    fn transact_batch<Tx>(
-        self,
-        txs: CrossbeamReceiver<IndexedTransaction<Tx>>,
-        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-        done_tx: SyncSender<()>,
-    ) where
-        Tx: ExecutableTxFor<Evm>,
-    {
-        let Some((mut evm, metrics, terminate_execution)) = self.evm_for_ctx() else { return };
-
-        while let Ok(IndexedTransaction { index, tx }) = txs.recv() {
-            let _enter = debug_span!(
-                target: "engine::tree::payload_processor::prewarm",
-                "prewarm tx",
-                i=index,
-            )
-            .entered();
-
-            // create the tx env
-            let start = Instant::now();
-
-            // If the task was cancelled, stop execution, and exit.
-            if terminate_execution.load(Ordering::Relaxed) {
-                break
-            }
-
-            let (tx_env, tx) = tx.into_parts();
-            let res = match evm.transact(tx_env) {
-                Ok(res) => res,
-                Err(err) => {
-                    trace!(
-                        target: "engine::tree::payload_processor::prewarm",
-                        %err,
-                        tx_hash=%tx.tx().tx_hash(),
-                        sender=%tx.signer(),
-                        "Error when executing prewarm transaction",
-                    );
-                    // Track transaction execution errors
-                    metrics.transaction_errors.increment(1);
-                    // skip error because we can ignore these errors and continue with the next tx
-                    continue
-                }
-            };
-            metrics.execution_duration.record(start.elapsed());
-
-            // If the task was cancelled, stop execution, and exit.
-            if terminate_execution.load(Ordering::Relaxed) {
-                break
-            }
-
-            // Only send outcome for transactions after the first txn
-            // as the main execution will be just as fast
-            if index > 0 {
-                let (targets, storage_targets) = multiproof_targets_from_state(res.state);
-                metrics.prefetch_storage_targets.record(storage_targets as f64);
-                if let Some(to_multi_proof) = &to_multi_proof {
-                    let _ = to_multi_proof.send(MultiProofMessage::PrefetchProofs(targets));
-                }
-            }
-
-            metrics.total_runtime.record(start.elapsed());
-        }
-
-        // send a message to the main task to flag that we're done
-        let _ = done_tx.send(());
-    }
-
-    /// Spawns worker tasks that pull transactions from a shared channel.
-    ///
-    /// Returns the sender for distributing transactions to workers.
-    fn spawn_workers<Tx>(
-        self,
-        workers_needed: usize,
-        task_executor: &Runtime,
-        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-        done_tx: SyncSender<()>,
-    ) -> CrossbeamSender<IndexedTransaction<Tx>>
-    where
-        Tx: ExecutableTxFor<Evm> + Send + 'static,
-    {
-        let (tx_sender, tx_receiver) = crossbeam_channel::unbounded();
-
-        // Spawn workers that all pull from the shared receiver
-        let span = Span::current();
-        for idx in 0..workers_needed {
-            let ctx = self.clone();
-            let to_multi_proof = to_multi_proof.clone();
-            let done_tx = done_tx.clone();
-            let rx = tx_receiver.clone();
-            let span = debug_span!(target: "engine::tree::payload_processor::prewarm", parent: &span, "prewarm_worker", idx);
-            task_executor.prewarming_pool().spawn(move || {
-                let _enter = span.entered();
-                ctx.transact_batch(rx, to_multi_proof, done_tx);
-            });
-        }
-
-        tx_sender
+        Some((evm, self.metrics.clone(), self.terminate_execution.clone()))
     }
 
     /// Prefetches a single account and all its storage slots from the BAL into the cache.

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

@@ -12,6 +12,7 @@ use crossbeam_channel::Receiver;
 use reth_primitives_traits::Receipt;
 use reth_trie_common::ordered_root::OrderedTrieRootEncodedBuilder;
 use tokio::sync::oneshot;
+use tracing::debug_span;
 
 /// Receipt with index, ready to be sent to the background task for encoding and trie building.
 #[derive(Debug, Clone)]
@@ -65,6 +66,13 @@ impl<R: Receipt> ReceiptRootTaskHandle<R> {
     /// * `receipts_len` - The total number of receipts expected. This is needed to correctly order
     ///   the trie keys according to RLP encoding rules.
     pub fn run(self, receipts_len: usize) {
+        let _span = debug_span!(
+            target: "engine::tree::payload_processor",
+            "receipt_root",
+            receipts_len,
+        )
+        .entered();
+
         let mut builder = OrderedTrieRootEncodedBuilder::new(receipts_len);
         let mut aggregated_bloom = Bloom::ZERO;
         let mut encode_buf = Vec::new();

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

@@ -1,5 +1,7 @@
 //! Sparse Trie task related functionality.
 
+use std::sync::Arc;
+
 use crate::tree::{
     multiproof::{
         dispatch_with_chunking, evm_state_to_hashed_post_state, MultiProofMessage,
@@ -31,7 +33,7 @@ use reth_trie_sparse::{
     SparseTrie,
 };
 use revm_primitives::{hash_map::Entry, B256Map};
-use tracing::{debug, debug_span, error, instrument};
+use tracing::{debug, debug_span, error, instrument, trace_span};
 
 /// Maximum number of pending/prewarm updates that we accumulate in memory before actually applying.
 const MAX_PENDING_UPDATES: usize = 100;
@@ -51,7 +53,7 @@ pub(super) struct SparseTrieCacheTask<A = ParallelSparseTrie, S = ParallelSparse
 
     /// The size of proof targets chunk to spawn in one calculation.
     /// If None, chunking is disabled and all targets are processed in a single proof.
-    chunk_size: Option<usize>,
+    chunk_size: usize,
     /// If this number is exceeded and chunking is enabled, then this will override whether or not
     /// there are any active workers and force chunking across workers. This is to prevent tasks
     /// which are very long from hitting a single worker.
@@ -90,8 +92,8 @@ pub(super) struct SparseTrieCacheTask<A = ParallelSparseTrie, S = ParallelSparse
     account_rlp_buf: Vec<u8>,
     /// Whether the last state update has been received.
     finished_state_updates: bool,
-    /// Pending targets to be dispatched to the proof workers.
-    pending_targets: MultiProofTargetsV2,
+    /// Pending proof targets queued for dispatch to proof workers.
+    pending_targets: PendingTargets,
     /// Number of pending execution/prewarming updates received but not yet passed to
     /// `update_leaves`.
     pending_updates: usize,
@@ -112,7 +114,7 @@ where
         proof_worker_handle: ProofWorkerHandle,
         metrics: MultiProofTaskMetrics,
         trie: SparseStateTrie<A, S>,
-        chunk_size: Option<usize>,
+        chunk_size: usize,
     ) -> Self {
         let (proof_result_tx, proof_result_rx) = crossbeam_channel::unbounded();
         let (hashed_state_tx, hashed_state_rx) = crossbeam_channel::unbounded();
@@ -192,8 +194,10 @@ where
         max_nodes_capacity: usize,
         max_values_capacity: usize,
         disable_pruning: bool,
+        updates: &TrieUpdates,
     ) -> (SparseStateTrie<A, S>, DeferredDrops) {
         let Self { mut trie, .. } = self;
+        trie.commit_updates(updates);
         if !disable_pruning {
             trie.prune(prune_depth, max_storage_tries);
             trie.shrink_to(max_nodes_capacity, max_values_capacity);
@@ -234,8 +238,13 @@ where
         let now = Instant::now();
 
         loop {
+            let mut t = Instant::now();
             crossbeam_channel::select_biased! {
                 recv(self.updates) -> message => {
+                    self.metrics
+                        .sparse_trie_channel_wait_duration_histogram
+                        .record(t.elapsed());
+
                     let update = match message {
                         Ok(m) => m,
                         Err(_) => {
@@ -249,17 +258,32 @@ where
                     self.pending_updates += 1;
                 }
                 recv(self.proof_result_rx) -> message => {
+                    let phase_end = Instant::now();
+                    self.metrics
+                        .sparse_trie_channel_wait_duration_histogram
+                        .record(phase_end.duration_since(t));
+                    t = phase_end;
+
                     let Ok(result) = message else {
                         unreachable!("we own the sender half")
                     };
-                    let mut result = result.result?;
 
+                    let mut result = result.result?;
                     while let Ok(next) = self.proof_result_rx.try_recv() {
                         let res = next.result?;
                         result.extend(res);
                     }
 
+                    let phase_end = Instant::now();
+                    self.metrics
+                        .sparse_trie_proof_coalesce_duration_histogram
+                        .record(phase_end.duration_since(t));
+                    t = phase_end;
+
                     self.on_proof_result(result)?;
+                    self.metrics
+                        .sparse_trie_reveal_multiproof_duration_histogram
+                        .record(t.elapsed());
                 },
             }
 
@@ -267,8 +291,10 @@ where
                 // If we don't have any pending messages, we can spend some time on computing
                 // storage roots and promoting account updates.
                 self.dispatch_pending_targets();
+                t = Instant::now();
                 self.process_new_updates()?;
                 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() &&
@@ -281,9 +307,11 @@ where
             } 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,
+                t = Instant::now();
                 self.process_new_updates()?;
+                self.metrics.sparse_trie_process_updates_duration_histogram.record(t.elapsed());
                 self.dispatch_pending_targets();
-            } else if self.pending_targets.chunking_length() > self.chunk_size.unwrap_or_default() {
+            } else if self.pending_targets.len() > self.chunk_size {
                 // Make sure to dispatch targets if we've accumulated a lot of them.
                 self.dispatch_pending_targets();
             }
@@ -306,7 +334,7 @@ where
 
         Ok(StateRootComputeOutcome {
             state_root,
-            trie_updates,
+            trie_updates: Arc::new(trie_updates),
             #[cfg(feature = "trie-debug")]
             debug_recorders,
         })
@@ -404,12 +432,12 @@ where
         })
     }
 
-    #[instrument(
-        level = "debug",
-        target = "engine::tree::payload_processor::sparse_trie",
-        skip_all
-    )]
     fn process_new_updates(&mut self) -> SparseTrieResult<()> {
+        if self.pending_updates == 0 {
+            return Ok(());
+        }
+
+        let _span = debug_span!("process_new_updates").entered();
         self.pending_updates = 0;
 
         // Firstly apply all new storage and account updates to the tries.
@@ -465,50 +493,38 @@ where
         let storage_updates =
             if new { &mut self.new_storage_updates } else { &mut self.storage_updates };
 
-        // Process all storage updates in parallel, skipping tries with no pending updates.
-        let span = tracing::Span::current();
-        let storage_results = storage_updates
-            .iter_mut()
-            .filter(|(_, updates)| !updates.is_empty())
-            .map(|(address, updates)| {
-                let trie = self.trie.take_or_create_storage_trie(address);
-                let fetched = self.fetched_storage_targets.remove(address).unwrap_or_default();
+        // Process all storage updates, skipping tries with no pending updates.
+        let span = debug_span!("process_storage_leaf_updates").entered();
+        for (address, updates) in storage_updates {
+            if updates.is_empty() {
+                continue;
+            }
+            let _enter = trace_span!(target: "engine::tree::payload_processor::sparse_trie", parent: &span, "storage_trie_leaf_updates", a=%address).entered();
 
-                (address, updates, fetched, trie)
-            })
-            .par_bridge_buffered()
-            .map(|(address, updates, mut fetched, mut trie)| {
-                let _enter = debug_span!(target: "engine::tree::payload_processor::sparse_trie", parent: &span, "storage_trie_leaf_updates", a=%address).entered();
-                let mut targets = Vec::new();
+            let trie = self.trie.get_or_create_storage_trie_mut(*address);
+            let fetched = self.fetched_storage_targets.entry(*address).or_default();
+            let mut targets = Vec::new();
 
-                trie.update_leaves(updates, |path, min_len| match fetched.entry(path) {
-                    Entry::Occupied(mut entry) => {
-                        if min_len < *entry.get() {
-                            entry.insert(min_len);
-                            targets.push(Target::new(path).with_min_len(min_len));
-                        }
-                    }
-                    Entry::Vacant(entry) => {
+            trie.update_leaves(updates, |path, min_len| match fetched.entry(path) {
+                Entry::Occupied(mut entry) => {
+                    if min_len < *entry.get() {
                         entry.insert(min_len);
                         targets.push(Target::new(path).with_min_len(min_len));
                     }
-                })?;
-
-                SparseTrieResult::Ok((address, targets, fetched, trie))
-            })
-            .collect::<Result<Vec<_>, _>>()?;
-
-        drop(span);
-
-        for (address, targets, fetched, trie) in storage_results {
-            self.fetched_storage_targets.insert(*address, fetched);
-            self.trie.insert_storage_trie(*address, trie);
+                }
+                Entry::Vacant(entry) => {
+                    entry.insert(min_len);
+                    targets.push(Target::new(path).with_min_len(min_len));
+                }
+            })?;
 
             if !targets.is_empty() {
-                self.pending_targets.storage_targets.entry(*address).or_default().extend(targets);
+                self.pending_targets.extend_storage_targets(address, targets);
             }
         }
 
+        drop(span);
+
         // Process account trie updates and fill the account targets.
         self.process_account_leaf_updates(new)?;
 
@@ -535,15 +551,13 @@ where
                     if min_len < *entry.get() {
                         entry.insert(min_len);
                         self.pending_targets
-                            .account_targets
-                            .push(Target::new(target).with_min_len(min_len));
+                            .push_account_target(Target::new(target).with_min_len(min_len));
                     }
                 }
                 Entry::Vacant(entry) => {
                     entry.insert(min_len);
                     self.pending_targets
-                        .account_targets
-                        .push(Target::new(target).with_min_len(min_len));
+                        .push_account_target(Target::new(target).with_min_len(min_len));
                 }
             }
         })?;
@@ -646,39 +660,36 @@ where
         Ok(())
     }
 
-    #[instrument(
-        level = "debug",
-        target = "engine::tree::payload_processor::sparse_trie",
-        skip_all
-    )]
     fn dispatch_pending_targets(&mut self) {
-        if !self.pending_targets.is_empty() {
-            let chunking_length = self.pending_targets.chunking_length();
-            dispatch_with_chunking(
-                std::mem::take(&mut self.pending_targets),
-                chunking_length,
-                self.chunk_size,
-                self.max_targets_for_chunking,
-                self.proof_worker_handle.available_account_workers(),
-                self.proof_worker_handle.available_storage_workers(),
-                MultiProofTargetsV2::chunks,
-                |proof_targets| {
-                    if let Err(e) = self.proof_worker_handle.dispatch_account_multiproof(
-                        AccountMultiproofInput {
-                            targets: proof_targets,
-                            proof_result_sender: ProofResultContext::new(
-                                self.proof_result_tx.clone(),
-                                0,
-                                HashedPostState::default(),
-                                Instant::now(),
-                            ),
-                        },
-                    ) {
-                        error!("failed to dispatch account multiproof: {e:?}");
-                    }
-                },
-            );
+        if self.pending_targets.is_empty() {
+            return;
         }
+
+        let _span = debug_span!("dispatch_pending_targets").entered();
+        let (targets, chunking_length) = self.pending_targets.take();
+        dispatch_with_chunking(
+            targets,
+            chunking_length,
+            self.chunk_size,
+            self.max_targets_for_chunking,
+            self.proof_worker_handle.available_account_workers(),
+            self.proof_worker_handle.available_storage_workers(),
+            MultiProofTargetsV2::chunks,
+            |proof_targets| {
+                if let Err(e) =
+                    self.proof_worker_handle.dispatch_account_multiproof(AccountMultiproofInput {
+                        targets: proof_targets,
+                        proof_result_sender: ProofResultContext::new(
+                            self.proof_result_tx.clone(),
+                            HashedPostState::default(),
+                            Instant::now(),
+                        ),
+                    })
+                {
+                    error!("failed to dispatch account multiproof: {e:?}");
+                }
+            },
+        );
     }
 }
 
@@ -697,6 +708,44 @@ fn encode_account_leaf_value(
     account_rlp_buf.clone()
 }
 
+/// Pending proof targets queued for dispatch to proof workers, along with their count.
+#[derive(Default)]
+struct PendingTargets {
+    /// The proof targets.
+    targets: MultiProofTargetsV2,
+    /// Number of account + storage proof targets currently queued.
+    len: usize,
+}
+
+impl PendingTargets {
+    /// Returns the number of pending targets.
+    const fn len(&self) -> usize {
+        self.len
+    }
+
+    /// Returns `true` if there are no pending targets.
+    const fn is_empty(&self) -> bool {
+        self.len == 0
+    }
+
+    /// Takes the pending targets, replacing with empty defaults.
+    fn take(&mut self) -> (MultiProofTargetsV2, usize) {
+        (std::mem::take(&mut self.targets), std::mem::take(&mut self.len))
+    }
+
+    /// Adds a target to the account targets.
+    fn push_account_target(&mut self, target: Target) {
+        self.targets.account_targets.push(target);
+        self.len += 1;
+    }
+
+    /// Extends storage targets for the given address.
+    fn extend_storage_targets(&mut self, address: &B256, targets: Vec<Target>) {
+        self.len += targets.len();
+        self.targets.storage_targets.entry(*address).or_default().extend(targets);
+    }
+}
+
 /// Message type for the sparse trie task.
 enum SparseTrieTaskMessage {
     /// A hashed state update ready to be processed.
@@ -709,12 +758,12 @@ enum SparseTrieTaskMessage {
 
 /// Outcome of the state root computation, including the state root itself with
 /// the trie updates.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct StateRootComputeOutcome {
     /// The state root.
     pub state_root: B256,
     /// The trie updates.
-    pub trie_updates: TrieUpdates,
+    pub trie_updates: Arc<TrieUpdates>,
     /// Debug recorders taken from the sparse tries, keyed by `None` for account trie
     /// and `Some(address)` for storage tries.
     #[cfg(feature = "trie-debug")]

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

@@ -52,7 +52,7 @@ use std::{
     panic::{self, AssertUnwindSafe},
     sync::{mpsc::RecvTimeoutError, Arc},
 };
-use tracing::{debug, debug_span, error, info, instrument, trace, warn};
+use tracing::{debug, debug_span, error, info, instrument, trace, warn, Span};
 
 /// Handle to a [`HashedPostState`] computed on a background thread.
 type LazyHashedPostState = reth_tasks::LazyHandle<HashedPostState>;
@@ -292,7 +292,7 @@ where
         let block = self.convert_to_block(input)?;
 
         // Validate block consensus rules which includes header validation
-        if let Err(consensus_err) = self.validate_block_inner(&block) {
+        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())
         }
@@ -334,9 +334,10 @@ where
         V: PayloadValidator<T, Block = N::Block> + Clone,
         Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
     {
-        // Spawn block conversion on a background thread so it runs concurrently with the
+        // Spawn payload conversion on a background thread so it runs concurrently with the
         // rest of the function (setup + execution). For payloads this overlaps the cost of
-        // RLP decoding + header hashing; for already-converted blocks this is a no-op.
+        // RLP decoding + header hashing.
+        let is_payload = matches!(&input, BlockOrPayload::Payload(_));
         let convert_to_block = match &input {
             BlockOrPayload::Payload(_) => {
                 let payload_clone = input.clone();
@@ -527,18 +528,44 @@ where
                 hashed_state_provider.hashed_post_state(&hashed_state_output.state)
             });
 
-        let block = convert_to_block(input)?.with_senders(senders);
+        let block = convert_to_block(input)?;
+        let transaction_root = is_payload.then(|| {
+            let block = block.clone();
+            let parent_span = Span::current();
+            let num_hash = block.num_hash();
+            self.payload_processor.executor().spawn_blocking_named("payload-tx-root", move || {
+                let _span =
+                    debug_span!(target: "engine::tree::payload_validator", parent: parent_span, "payload_tx_root", block = ?num_hash)
+                        .entered();
+                block.body().calculate_tx_root()
+            })
+        });
+        let block = block.with_senders(senders);
 
         // Wait for the receipt root computation to complete.
-        let receipt_root_bloom = receipt_root_rx
-            .blocking_recv()
-            .inspect_err(|_| {
-                tracing::error!(
-                    target: "engine::tree::payload_validator",
-                    "Receipt root task dropped sender without result, receipt root calculation likely aborted"
-                );
-            })
-            .ok();
+        let receipt_root_bloom = {
+            let _enter = debug_span!(
+                target: "engine::tree::payload_validator",
+                "wait_receipt_root",
+            )
+            .entered();
+
+            receipt_root_rx
+                .blocking_recv()
+                .inspect_err(|_| {
+                    tracing::error!(
+                        target: "engine::tree::payload_validator",
+                        "Receipt root task dropped sender without result, receipt root calculation likely aborted"
+                    );
+                })
+                .ok()
+        };
+        let transaction_root = transaction_root.map(|handle| {
+            let _span =
+                debug_span!(target: "engine::tree::payload_validator", "wait_payload_tx_root")
+                    .entered();
+            handle.try_into_inner().expect("sole handle")
+        });
 
         let hashed_state = ensure_ok_post_block!(
             self.validate_post_execution(
@@ -546,6 +573,7 @@ where
                 &parent_block,
                 &output,
                 &mut ctx,
+                transaction_root,
                 receipt_root_bloom,
                 hashed_state,
             ),
@@ -591,7 +619,7 @@ where
                             let _has_diff = self.compare_trie_updates_with_serial(
                                 overlay_factory.clone(),
                                 &hashed_state,
-                                trie_updates.clone(),
+                                trie_updates.as_ref().clone(),
                             );
                             #[cfg(feature = "trie-debug")]
                             if _has_diff {
@@ -637,7 +665,7 @@ where
                             ?elapsed,
                             "Regular root task finished"
                         );
-                        maybe_state_root = Some((result.0, result.1, elapsed));
+                        maybe_state_root = Some((result.0, Arc::new(result.1), elapsed));
                     }
                     Err(error) => {
                         debug!(target: "engine::tree::payload_validator", %error, "Parallel state root computation failed");
@@ -672,7 +700,7 @@ where
                 self.metrics.block_validation.state_root_task_fallback_success_total.increment(1);
             }
 
-            (root, updates, root_time.elapsed())
+            (root, Arc::new(updates), root_time.elapsed())
         };
 
         self.metrics.block_validation.record_state_root(&trie_output, root_elapsed.as_secs_f64());
@@ -737,13 +765,19 @@ where
     /// Validate if block is correct and satisfies all the consensus rules that concern the header
     /// and block body itself.
     #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
-    fn validate_block_inner(&self, block: &SealedBlock<N::Block>) -> Result<(), ConsensusError> {
+    fn validate_block_inner(
+        &self,
+        block: &SealedBlock<N::Block>,
+        transaction_root: Option<B256>,
+    ) -> 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)
         }
 
-        if let Err(e) = self.consensus.validate_block_pre_execution(block) {
+        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)
         }
@@ -1070,7 +1104,7 @@ where
                             let (state_root, trie_updates) = result?;
                             return Ok(Ok(StateRootComputeOutcome {
                                 state_root,
-                                trie_updates,
+                                trie_updates: Arc::new(trie_updates),
                                 #[cfg(feature = "trie-debug")]
                                 debug_recorders: Vec::new(),
                             }));
@@ -1087,7 +1121,7 @@ where
                         let (state_root, trie_updates) = result?;
                         return Ok(Ok(StateRootComputeOutcome {
                             state_root,
-                            trie_updates,
+                            trie_updates: Arc::new(trie_updates),
                             #[cfg(feature = "trie-debug")]
                             debug_recorders: Vec::new(),
                         }));
@@ -1208,12 +1242,14 @@ where
     ///
     /// The `hashed_state` handle wraps the background hashed post state computation.
     #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
+    #[expect(clippy::too_many_arguments)]
     fn validate_post_execution<T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
         &self,
         block: &RecoveredBlock<N::Block>,
         parent_block: &SealedHeader<N::BlockHeader>,
         output: &BlockExecutionOutput<N::Receipt>,
         ctx: &mut TreeCtx<'_, N>,
+        transaction_root: Option<B256>,
         receipt_root_bloom: Option<ReceiptRootBloom>,
         hashed_state: LazyHashedPostState,
     ) -> Result<LazyHashedPostState, InsertBlockErrorKind>
@@ -1224,7 +1260,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) {
+        if let Err(e) = self.validate_block_inner(block, transaction_root) {
             return Err(e.into())
         }
 
@@ -1251,10 +1287,15 @@ where
         }
         drop(_enter);
 
+        // Wait for the background keccak256 hashing task to complete. This blocks until
+        // all changed addresses and storage slots have been hashed.
+        let hashed_state_ref =
+            debug_span!(target: "engine::tree::payload_validator", "wait_hashed_post_state")
+                .in_scope(|| hashed_state.get());
+
         let _enter = debug_span!(target: "engine::tree::payload_validator", "validate_block_post_execution_with_hashed_state").entered();
-        if let Err(err) = self
-            .validator
-            .validate_block_post_execution_with_hashed_state(hashed_state.get(), block)
+        if let Err(err) =
+            self.validator.validate_block_post_execution_with_hashed_state(hashed_state_ref, block)
         {
             // call post-block hook
             self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
@@ -1479,7 +1520,7 @@ where
         execution_outcome: Arc<BlockExecutionOutput<N::Receipt>>,
         ctx: &TreeCtx<'_, N>,
         hashed_state: LazyHashedPostState,
-        trie_output: TrieUpdates,
+        trie_output: Arc<TrieUpdates>,
         overlay_factory: OverlayStateProviderFactory<P>,
     ) -> ExecutedBlock<N> {
         // Capture parent hash and ancestor overlays for deferred trie input construction.
@@ -1502,7 +1543,7 @@ where
             Err(handle) => Arc::new(handle.get().clone()),
         };
         let deferred_trie_data =
-            DeferredTrieData::pending(hashed_state, Arc::new(trie_output), anchor_hash, ancestors);
+            DeferredTrieData::pending(hashed_state, trie_output, anchor_hash, ancestors);
         let deferred_handle_task = deferred_trie_data.clone();
         let block_validation_metrics = self.metrics.block_validation.clone();
 

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

@@ -1,6 +1,9 @@
 //! Contains a precompile cache backed by `schnellru::LruMap` (LRU by length).
 
-use alloy_primitives::Bytes;
+use alloy_primitives::{
+    map::{DefaultHashBuilder, FbBuildHasher},
+    Bytes,
+};
 use moka::policy::EvictionPolicy;
 use reth_evm::precompiles::{DynPrecompile, Precompile, PrecompileInput};
 use reth_primitives_traits::dashmap::DashMap;
@@ -13,7 +16,7 @@ const MAX_CACHE_SIZE: u32 = 10_000;
 
 /// Stores caches for each precompile.
 #[derive(Debug, Clone, Default)]
-pub struct PrecompileCacheMap<S>(Arc<DashMap<Address, PrecompileCache<S>>>)
+pub struct PrecompileCacheMap<S>(Arc<DashMap<Address, PrecompileCache<S>, FbBuildHasher<20>>>)
 where
     S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static;
 
@@ -37,9 +40,7 @@ where
 
 /// Cache for precompiles, for each input stores the result.
 #[derive(Debug, Clone)]
-pub struct PrecompileCache<S>(
-    moka::sync::Cache<Bytes, CacheEntry<S>, alloy_primitives::map::DefaultHashBuilder>,
-)
+pub struct PrecompileCache<S>(moka::sync::Cache<Bytes, CacheEntry<S>, DefaultHashBuilder>)
 where
     S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static;
 

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

@@ -139,7 +139,7 @@ impl<N: NodePrimitives> TreeState<N> {
     ///
     /// Both parent hash and anchor hash must match to ensure the overlay is valid.
     /// This prevents using a stale overlay after persistence has advanced the anchor.
-    pub(crate) fn get_cached_overlay(
+    pub fn get_cached_overlay(
         &self,
         parent_hash: B256,
         expected_anchor: B256,

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

@@ -2041,3 +2041,126 @@ mod forkchoice_updated_tests {
         assert_eq!(last_persisted_number, canonical_tip);
     }
 }
+
+/// Tests that `on_valid_downloaded_block` triggers a download for the actual head block when
+/// the block matches a non-head sync target (safe or finalized).
+///
+/// This exercises the exact code path fixed in `on_downloaded_block`: after `insert_block`
+/// returns `Inserted(Valid)`, `on_valid_downloaded_block` checks `sync_target.contains()`.
+/// If the block is NOT the head, it should make canonical inline and emit a `Download`
+/// event for the head — rather than returning `MakeCanonical` which would stop the download
+/// pipeline.
+///
+/// Reproduces the hive test failure:
+///   "Sync after 2 blocks - Withdrawals on Block 2 - Multiple Withdrawal Accounts -
+///    No Transactions: Timeout while waiting for secondary client to sync"
+#[test]
+fn test_on_valid_downloaded_non_head_sync_target_continues_to_head() {
+    reth_tracing::init_test_tracing();
+
+    let chain_spec = MAINNET.clone();
+    let mut test_harness = TestHarness::new(chain_spec);
+
+    // Build blocks: genesis (0) and safe block (1).
+    let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..2).collect();
+    let genesis = &blocks[0];
+    let safe_block = &blocks[1];
+
+    // Insert genesis and safe block into the tree. The safe block must be in the tree
+    // for `make_canonical` to succeed inside `on_valid_downloaded_block`.
+    test_harness = test_harness.with_blocks(vec![genesis.clone(), safe_block.clone()]);
+
+    let genesis_hash = genesis.recovered_block().hash();
+    let safe_hash = safe_block.recovered_block().hash();
+    let head_hash = B256::random(); // head block is unknown — hasn't been downloaded yet
+
+    // Reset canonical head to genesis so the safe block is in tree but not yet canonical.
+    test_harness.tree.state.tree_state.set_canonical_head(genesis.recovered_block().num_hash());
+
+    // Set the forkchoice tracker to SYNCING with head != safe.
+    let fcu_state = ForkchoiceState {
+        head_block_hash: head_hash,
+        safe_block_hash: safe_hash,
+        finalized_block_hash: genesis_hash,
+    };
+    test_harness
+        .tree
+        .state
+        .forkchoice_state_tracker
+        .set_latest(fcu_state, ForkchoiceStatus::Syncing);
+
+    // Call on_valid_downloaded_block — this is called by on_downloaded_block after
+    // insert_block returns Inserted(Valid).
+    let safe_num_hash = safe_block.recovered_block().num_hash();
+    let result = test_harness.tree.on_valid_downloaded_block(safe_num_hash).unwrap();
+
+    // With the fix: the engine makes safe canonical inline, then emits Download for head.
+    // Without the fix: it would return MakeCanonical{safe_hash} and never download head.
+    match result {
+        Some(TreeEvent::Download(DownloadRequest::BlockSet(hashes))) => {
+            assert!(
+                hashes.contains(&head_hash),
+                "Expected download for head block {head_hash}, got {hashes:?}"
+            );
+        }
+        Some(TreeEvent::TreeAction(TreeAction::MakeCanonical { sync_target_head })) => {
+            panic!(
+                "BUG: returned MakeCanonical for non-head block {sync_target_head} \
+                 instead of downloading the actual head {head_hash}"
+            );
+        }
+        other => panic!("Expected Download event for head block, got: {other:?}"),
+    }
+
+    // Verify the safe block was made canonical.
+    assert_eq!(
+        test_harness.tree.state.tree_state.canonical_block_hash(),
+        safe_hash,
+        "Safe block should be canonical after on_valid_downloaded_block"
+    );
+}
+
+/// Tests that `on_valid_downloaded_block` returns `MakeCanonical` when the downloaded block
+/// IS the sync target head (the normal non-buggy path).
+#[test]
+fn test_on_valid_downloaded_head_sync_target_returns_make_canonical() {
+    reth_tracing::init_test_tracing();
+
+    let chain_spec = MAINNET.clone();
+    let mut test_harness = TestHarness::new(chain_spec);
+
+    let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..2).collect();
+    let genesis = &blocks[0];
+    let head_block = &blocks[1];
+
+    test_harness = test_harness.with_blocks(vec![genesis.clone(), head_block.clone()]);
+
+    let genesis_hash = genesis.recovered_block().hash();
+    let head_hash = head_block.recovered_block().hash();
+
+    // Reset canonical head to genesis.
+    test_harness.tree.state.tree_state.set_canonical_head(genesis.recovered_block().num_hash());
+
+    // Set the forkchoice tracker: head == the downloaded block.
+    let fcu_state = ForkchoiceState {
+        head_block_hash: head_hash,
+        safe_block_hash: head_hash,
+        finalized_block_hash: genesis_hash,
+    };
+    test_harness
+        .tree
+        .state
+        .forkchoice_state_tracker
+        .set_latest(fcu_state, ForkchoiceStatus::Syncing);
+
+    let head_num_hash = head_block.recovered_block().num_hash();
+    let result = test_harness.tree.on_valid_downloaded_block(head_num_hash).unwrap();
+
+    // When the downloaded block IS the head, should return MakeCanonical.
+    match result {
+        Some(TreeEvent::TreeAction(TreeAction::MakeCanonical { sync_target_head })) => {
+            assert_eq!(sync_target_head, head_hash);
+        }
+        other => panic!("Expected MakeCanonical for head block, got: {other:?}"),
+    }
+}

crates/engine/util/src/engine_store.rs

@@ -108,7 +108,7 @@ impl EngineMessageStore {
                 tracing::warn!(target: "engine::store", ?filename, "Skipping non json file");
             }
         }
-        Ok(filenames_by_ts.into_iter().flat_map(|(_, paths)| paths))
+        Ok(filenames_by_ts.into_values().flatten())
     }
 }
 

crates/ethereum/cli/src/app.rs

@@ -97,7 +97,21 @@ where
     {
         let runner = match self.runner.take() {
             Some(runner) => runner,
-            None => CliRunner::try_default_runtime()?,
+            None => {
+                let runtime_config = match &self.cli.command {
+                    Commands::Node(command) => {
+                        reth_tasks::RuntimeConfig::default().with_rayon(RayonConfig {
+                            reserved_cpu_cores: command.engine.reserved_cpu_cores,
+                            proof_storage_worker_threads: command.engine.storage_worker_count,
+                            proof_account_worker_threads: command.engine.account_worker_count,
+                            prewarming_threads: command.engine.prewarming_threads,
+                            ..Default::default()
+                        })
+                    }
+                    _ => reth_tasks::RuntimeConfig::default(),
+                };
+                CliRunner::try_with_runtime_config(runtime_config)?
+            }
         };
 
         // Add network name if available to the logs dir
@@ -123,8 +137,8 @@ where
     ///
     /// See [`Cli::init_tracing`] for more information.
     pub fn init_tracing(&mut self, runner: &CliRunner) -> Result<()> {
-        if self.guard.is_none() {
-            self.guard = self.cli.init_tracing(runner, self.layers.take().unwrap_or_default())?;
+        if let Some(layers) = self.layers.take() {
+            self.guard = self.cli.init_tracing(runner, layers)?;
         }
 
         Ok(())
@@ -161,17 +175,6 @@ where
                 Rpc::validate_selection(ws_api, "ws.api").map_err(|e| eyre!("{e}"))?;
             }
 
-            let rayon_config = RayonConfig {
-                reserved_cpu_cores: command.engine.reserved_cpu_cores,
-                proof_storage_worker_threads: command.engine.storage_worker_count,
-                proof_account_worker_threads: command.engine.account_worker_count,
-                prewarming_threads: command.engine.prewarming_threads,
-                ..Default::default()
-            };
-            let runner = CliRunner::try_with_runtime_config(
-                reth_tasks::RuntimeConfig::default().with_rayon(rayon_config),
-            )?;
-
             runner.run_command_until_exit(|ctx| {
                 command.execute(ctx, FnLauncher::new::<C, Ext>(launcher))
             })

crates/ethereum/cli/src/interface.rs

@@ -1,11 +1,8 @@
 //! CLI definition and entrypoint to executable
 
-use crate::{
-    app::{run_commands_with, CliApp},
-    chainspec::EthereumChainSpecParser,
-};
+use crate::{app::CliApp, chainspec::EthereumChainSpecParser};
 use clap::{Parser, Subcommand};
-use reth_chainspec::{ChainSpec, EthChainSpec, Hardforks};
+use reth_chainspec::{ChainSpec, Hardforks};
 use reth_cli::chainspec::ChainSpecParser;
 use reth_cli_commands::{
     common::{CliComponentsBuilder, CliNodeTypes, HeaderMut},
@@ -22,7 +19,6 @@ use reth_node_core::{
     args::{LogArgs, OtlpInitStatus, OtlpLogsStatus, TraceArgs},
     version::version_metadata,
 };
-use reth_node_metrics::recorder::install_prometheus_recorder;
 use reth_rpc_server_types::{DefaultRpcModuleValidator, RpcModuleValidator};
 use reth_tracing::{FileWorkerGuard, Layers};
 use std::{ffi::OsString, fmt, future::Future, marker::PhantomData, sync::Arc};
@@ -135,7 +131,8 @@ impl<
         Fut: Future<Output = eyre::Result<()>>,
         C: ChainSpecParser<ChainSpec = ChainSpec>,
     {
-        self.with_runner(CliRunner::try_default_runtime()?, launcher)
+        self.configure()
+            .run(FnLauncher::new::<C, Ext>(async move |builder, ext| launcher(builder, ext).await))
     }
 
     /// Execute the configured cli command with the provided [`CliComponentsBuilder`].
@@ -156,7 +153,7 @@ impl<
         N: CliNodeTypes<Primitives: NodePrimitives<BlockHeader: HeaderMut>, ChainSpec: Hardforks>,
         C: ChainSpecParser<ChainSpec = N::ChainSpec>,
     {
-        self.with_runner_and_components(CliRunner::try_default_runtime()?, components, launcher)
+        self.configure().run_with_components(components, launcher)
     }
 
     /// Execute the configured cli command with the provided [`CliRunner`].
@@ -192,7 +189,7 @@ impl<
     /// Execute the configured cli command with the provided [`CliRunner`] and
     /// [`CliComponentsBuilder`].
     pub fn with_runner_and_components<N>(
-        mut self,
+        self,
         runner: CliRunner,
         components: impl CliComponentsBuilder<N>,
         launcher: impl AsyncFnOnce(
@@ -204,24 +201,9 @@ impl<
         N: CliNodeTypes<Primitives: NodePrimitives<BlockHeader: HeaderMut>, ChainSpec: Hardforks>,
         C: ChainSpecParser<ChainSpec = N::ChainSpec>,
     {
-        // Add network name if available to the logs dir
-        if let Some(chain_spec) = self.command.chain_spec() {
-            self.logs.log_file_directory =
-                self.logs.log_file_directory.join(chain_spec.chain().to_string());
-        }
-
-        // Apply node-specific log defaults before initializing tracing
-        if matches!(self.command, Commands::Node(_)) {
-            self.logs.apply_node_defaults();
-        }
-
-        let _guard = self.init_tracing(&runner, Layers::new())?;
-
-        // Install the prometheus recorder to be sure to record all metrics
-        install_prometheus_recorder();
-
-        // Use the shared standalone function to avoid duplication
-        run_commands_with::<C, Ext, Rpc, N, SubCmd>(self, runner, components, launcher)
+        let mut app = self.configure();
+        app.set_runner(runner);
+        app.run_with_components(components, launcher)
     }
 
     /// Initializes tracing with the configured options.
@@ -368,7 +350,7 @@ mod tests {
     use super::*;
     use crate::chainspec::SUPPORTED_CHAINS;
     use clap::CommandFactory;
-    use reth_chainspec::SEPOLIA;
+    use reth_chainspec::{EthChainSpec, SEPOLIA};
     use reth_node_core::args::ColorMode;
 
     #[test]

crates/ethereum/consensus/src/lib.rs

@@ -15,13 +15,16 @@ use alloc::{fmt::Debug, sync::Arc};
 use alloy_consensus::{constants::MAXIMUM_EXTRA_DATA_SIZE, EMPTY_OMMER_ROOT_HASH};
 use alloy_eips::eip7840::BlobParams;
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
-use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
+use reth_consensus::{
+    Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom, TransactionRoot,
+};
 use reth_consensus_common::validation::{
     validate_4844_header_standalone, validate_against_parent_4844,
     validate_against_parent_eip1559_base_fee, validate_against_parent_gas_limit,
     validate_against_parent_hash_number, validate_against_parent_timestamp,
-    validate_block_pre_execution, validate_body_against_header, validate_header_base_fee,
-    validate_header_extra_data, validate_header_gas,
+    validate_block_pre_execution, validate_block_pre_execution_with_tx_root,
+    validate_body_against_header, validate_header_base_fee, validate_header_extra_data,
+    validate_header_gas,
 };
 use reth_execution_types::BlockExecutionResult;
 use reth_primitives_traits::{
@@ -102,6 +105,14 @@ where
     fn validate_block_pre_execution(&self, block: &SealedBlock<B>) -> Result<(), ConsensusError> {
         validate_block_pre_execution(block, &self.chain_spec)
     }
+
+    fn validate_block_pre_execution_with_tx_root(
+        &self,
+        block: &SealedBlock<B>,
+        transaction_root: Option<TransactionRoot>,
+    ) -> Result<(), ConsensusError> {
+        validate_block_pre_execution_with_tx_root(block, &self.chain_spec, transaction_root)
+    }
 }
 
 impl<H, ChainSpec> HeaderValidator<H> for EthBeaconConsensus<ChainSpec>
@@ -231,7 +242,7 @@ mod tests {
     #[test]
     fn test_valid_gas_limit_increase() {
         let parent = header_with_gas_limit(GAS_LIMIT_BOUND_DIVISOR * 10);
-        let child = header_with_gas_limit((parent.gas_limit + 5) as u64);
+        let child = header_with_gas_limit(parent.gas_limit + 5);
 
         assert!(validate_against_parent_gas_limit(
             &child,
@@ -249,7 +260,7 @@ mod tests {
         assert!(matches!(
             validate_against_parent_gas_limit(&child, &parent, &ChainSpec::<Header>::default()).unwrap_err(),
             ConsensusError::GasLimitInvalidMinimum { child_gas_limit }
-                if child_gas_limit == child.gas_limit as u64
+                if child_gas_limit == child.gas_limit
         ));
     }
 

crates/ethereum/payload/src/lib.rs

@@ -253,8 +253,8 @@ where
         // There's only limited amount of blob space available per block, so we need to check if
         // the EIP-4844 can still fit in the block
         let mut blob_tx_sidecar = None;
-        if let Some(blob_tx) = tx.as_eip4844() {
-            let tx_blob_count = blob_tx.tx().blob_versioned_hashes.len() as u64;
+        if let Some(blob_hashes) = tx.blob_versioned_hashes() {
+            let tx_blob_count = blob_hashes.len() as u64;
 
             if block_blob_count + tx_blob_count > max_blob_count {
                 // we can't fit this _blob_ transaction into the block, so we mark it as
@@ -329,8 +329,8 @@ where
         };
 
         // add to the total blob gas used if the transaction successfully executed
-        if let Some(blob_tx) = tx.as_eip4844() {
-            block_blob_count += blob_tx.tx().blob_versioned_hashes.len() as u64;
+        if let Some(blob_hashes) = tx.blob_versioned_hashes() {
+            block_blob_count += blob_hashes.len() as u64;
 
             // if we've reached the max blob count, we can skip blob txs entirely
             if block_blob_count == max_blob_count {

crates/evm/execution-errors/src/trie.rs

@@ -165,13 +165,8 @@ pub enum SparseTrieErrorKind {
     #[error("sparse trie is blind")]
     Blind,
     /// Encountered blinded node on update.
-    #[error("attempted to update blind node at {path:?}: {hash}")]
-    BlindedNode {
-        /// Blind node path.
-        path: Nibbles,
-        /// Node hash
-        hash: B256,
-    },
+    #[error("attempted to update blind node at {0:?}")]
+    BlindedNode(Nibbles),
     /// Encountered unexpected node at path when revealing.
     #[error("encountered an invalid node at path {path:?} when revealing: {node:?}")]
     Reveal {

crates/evm/execution-types/src/chain.rs

@@ -202,6 +202,18 @@ impl<N: NodePrimitives> Chain<N> {
         self.blocks().iter().map(|block| block.1)
     }
 
+    /// Returns an iterator over all transactions in the chain.
+    pub fn transactions_iter(&self) -> impl Iterator<Item = &N::SignedTx> + '_ {
+        self.blocks_iter().flat_map(|block| block.body().transactions())
+    }
+
+    /// Returns an iterator over all [`Recovered`] transaction references in the chain.
+    pub fn transactions_recovered_iter(
+        &self,
+    ) -> impl Iterator<Item = Recovered<&N::SignedTx>> + '_ {
+        self.blocks_iter().flat_map(|block| block.transactions_recovered())
+    }
+
     /// Returns an iterator over all blocks and their receipts in the chain.
     pub fn blocks_and_receipts(
         &self,

crates/exex/exex/src/backfill/job.rs

@@ -376,7 +376,7 @@ mod tests {
         );
 
         for (i, ((pipeline_block, pipeline_output), (backfill_block, mut backfill_output))) in
-            pipeline_results.iter().zip(backfill_results.into_iter()).enumerate()
+            pipeline_results.iter().zip(backfill_results).enumerate()
         {
             backfill_output.state.reverts.sort();
 

crates/exex/exex/src/manager.rs

@@ -505,9 +505,6 @@ where
         }
         let buffer_full = this.buffer.len() >= this.max_capacity;
 
-        // Update capacity
-        this.update_capacity();
-
         // Advance all poll senders
         let mut min_id = usize::MAX;
         for idx in (0..this.exex_handles.len()).rev() {

crates/net/discv4/src/lib.rs

@@ -955,10 +955,8 @@ impl Discv4Service {
 
         // Check if ENR was updated
         match (last_enr_seq, old_enr) {
-            (Some(new), Some(old)) => {
-                if new > old {
-                    self.send_enr_request(record);
-                }
+            (Some(new), Some(old)) if new > old => {
+                self.send_enr_request(record);
             }
             (Some(_), None) => {
                 // got an ENR
@@ -1195,10 +1193,8 @@ impl Discv4Service {
         } else {
             // Request ENR if included in the ping
             match (ping.enr_sq, old_enr) {
-                (Some(new), Some(old)) => {
-                    if new > old {
-                        self.send_enr_request(record);
-                    }
+                (Some(new), Some(old)) if new > old => {
+                    self.send_enr_request(record);
                 }
                 (Some(_), None) => {
                     self.send_enr_request(record);
@@ -1355,10 +1351,8 @@ impl Discv4Service {
                     _ => return,
                 };
                 match (fork_id, old_fork_id) {
-                    (Some(new), Some(old)) => {
-                        if new != old {
-                            self.notify(DiscoveryUpdate::EnrForkId(record, new))
-                        }
+                    (Some(new), Some(old)) if new != old => {
+                        self.notify(DiscoveryUpdate::EnrForkId(record, new))
                     }
                     (Some(new), None) => self.notify(DiscoveryUpdate::EnrForkId(record, new)),
                     _ => {}

crates/net/downloaders/src/bodies/queue.rs

@@ -52,6 +52,7 @@ where
     pub(crate) fn clear(&mut self) {
         self.inner.clear();
         self.last_requested_block_number.take();
+        self.metrics.clear();
     }
     /// Add new request to the queue.
     /// Expects a sorted list of headers.

crates/net/downloaders/src/metrics.rs

@@ -60,6 +60,15 @@ impl BodyDownloaderMetrics {
             _error => self.unexpected_errors.increment(1),
         }
     }
+
+    /// Clear all gauge metrics by setting them to 0.
+    pub fn clear(&self) {
+        self.in_flight_requests.set(0);
+        self.buffered_responses.set(0);
+        self.buffered_blocks.set(0);
+        self.buffered_blocks_size_bytes.set(0);
+        self.queued_blocks.set(0);
+    }
 }
 
 /// Metrics for an individual response, i.e. the size in bytes, and length (number of bodies) in the

crates/net/network-types/src/lib.rs

@@ -30,6 +30,6 @@ pub use peers::{
         DEFAULT_REPUTATION,
     },
     state::PeerConnectionState,
-    ConnectionsConfig, Peer, PeersConfig,
+    ConnectionsConfig, Peer, PeersConfig, PersistedPeerInfo,
 };
 pub use session::{SessionLimits, SessionsConfig};

crates/net/network-types/src/peers/config.rs

@@ -11,7 +11,7 @@ use reth_net_banlist::{BanList, IpFilter};
 use reth_network_peers::{NodeRecord, TrustedPeer};
 use tracing::info;
 
-use crate::{BackoffKind, ReputationChangeWeights};
+use crate::{peers::PersistedPeerInfo, BackoffKind, ReputationChangeWeights};
 
 /// Maximum number of available slots for outbound sessions.
 pub const DEFAULT_MAX_COUNT_PEERS_OUTBOUND: u32 = 100;
@@ -147,6 +147,9 @@ pub struct PeersConfig {
     /// Basic nodes to connect to.
     #[cfg_attr(feature = "serde", serde(skip))]
     pub basic_nodes: HashSet<NodeRecord>,
+    /// Peers restored from a previous run, containing richer metadata than basic nodes.
+    #[cfg_attr(feature = "serde", serde(skip))]
+    pub persisted_peers: Vec<PersistedPeerInfo>,
     /// How long to ban bad peers.
     #[cfg_attr(feature = "serde", serde(with = "humantime_serde"))]
     pub ban_duration: Duration,
@@ -193,6 +196,7 @@ impl Default for PeersConfig {
             trusted_nodes_only: false,
             trusted_nodes_resolution_interval: Duration::from_secs(60 * 60),
             basic_nodes: Default::default(),
+            persisted_peers: Default::default(),
             max_backoff_count: 5,
             incoming_ip_throttle_duration: INBOUND_IP_THROTTLE_DURATION,
             ip_filter: IpFilter::default(),
@@ -298,20 +302,39 @@ impl PeersConfig {
         self.connection_info.max_outbound + self.connection_info.max_inbound
     }
 
-    /// Read from file nodes available at launch. Ignored if None.
+    /// Read persisted peers from file at launch.
+    ///
+    /// Supports both the current [`PersistedPeerInfo`] format and the legacy `Vec<NodeRecord>`
+    /// format. Legacy entries are converted to [`PersistedPeerInfo`] with default metadata.
+    ///
+    /// Ignored if `optional_file` is `None` or the file does not exist.
+    #[cfg(feature = "serde")]
     pub fn with_basic_nodes_from_file(
-        self,
+        mut self,
         optional_file: Option<impl AsRef<Path>>,
     ) -> Result<Self, io::Error> {
         let Some(file_path) = optional_file else { return Ok(self) };
-        let reader = match std::fs::File::open(file_path.as_ref()) {
-            Ok(file) => io::BufReader::new(file),
+        let raw = match std::fs::read_to_string(file_path.as_ref()) {
+            Ok(contents) => contents,
             Err(e) if e.kind() == ErrorKind::NotFound => return Ok(self),
-            Err(e) => Err(e)?,
+            Err(e) => return Err(e),
         };
+
         info!(target: "net::peers", file = %file_path.as_ref().display(), "Loading saved peers");
-        let nodes: HashSet<NodeRecord> = serde_json::from_reader(reader)?;
-        Ok(self.with_basic_nodes(nodes))
+
+        // Try the new format first, fall back to legacy Vec<NodeRecord>
+        let peers: Vec<PersistedPeerInfo> = serde_json::from_str(&raw)
+            .or_else(|_| {
+                let nodes: HashSet<NodeRecord> = serde_json::from_str(&raw)?;
+                Ok::<_, serde_json::Error>(
+                    nodes.into_iter().map(PersistedPeerInfo::from_node_record).collect(),
+                )
+            })
+            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
+
+        info!(target: "net::peers", count = peers.len(), "Loaded persisted peers");
+        self.persisted_peers = peers;
+        Ok(self)
     }
 
     /// Configure the IP filter for restricting network connections to specific IP ranges.

crates/net/network-types/src/peers/kind.rs

@@ -2,6 +2,8 @@
 
 /// Represents the kind of peer
 #[derive(Debug, Clone, Copy, Default, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "serde", serde(rename_all = "lowercase"))]
 pub enum PeerKind {
     /// Basic peer kind.
     #[default]

crates/net/network-types/src/peers/mod.rs

@@ -8,6 +8,7 @@ pub use config::{ConnectionsConfig, PeersConfig};
 pub use reputation::{Reputation, ReputationChange, ReputationChangeKind, ReputationChangeWeights};
 
 use alloy_eip2124::ForkId;
+use reth_network_peers::{NodeRecord, PeerId};
 use tracing::trace;
 
 use crate::{
@@ -140,3 +141,33 @@ impl Peer {
         matches!(self.kind, PeerKind::Static)
     }
 }
+
+/// Peer info persisted to disk.
+///
+/// Contains richer metadata than a plain [`NodeRecord`], preserving the peer's kind, fork ID,
+/// and reputation across restarts.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+pub struct PersistedPeerInfo {
+    /// The node record (id, address, ports).
+    pub record: NodeRecord,
+    /// The kind of peer.
+    pub kind: PeerKind,
+    /// The [`ForkId`] that the peer announced via discovery.
+    #[cfg_attr(feature = "serde", serde(default, skip_serializing_if = "Option::is_none"))]
+    pub fork_id: Option<ForkId>,
+    /// The peer's reputation at the time of persisting.
+    pub reputation: i32,
+}
+
+impl PersistedPeerInfo {
+    /// Returns the peer id.
+    pub const fn peer_id(&self) -> PeerId {
+        self.record.id
+    }
+
+    /// Converts a legacy [`NodeRecord`] into a [`PersistedPeerInfo`] with default metadata.
+    pub const fn from_node_record(record: NodeRecord) -> Self {
+        Self { record, kind: PeerKind::Basic, fork_id: None, reputation: DEFAULT_REPUTATION }
+    }
+}

crates/net/network/Cargo.toml

@@ -33,7 +33,7 @@ reth-storage-api.workspace = true
 reth-tokio-util.workspace = true
 reth-consensus.workspace = true
 reth-network-peers = { workspace = true, features = ["net"] }
-reth-network-types.workspace = true
+reth-network-types = { workspace = true, features = ["serde"] }
 
 # ethereum
 alloy-consensus.workspace = true
@@ -105,7 +105,6 @@ serde = [
     "dep:serde",
     "secp256k1/serde",
     "enr/serde",
-    "reth-network-types/serde",
     "reth-dns-discovery/serde",
     "reth-eth-wire/serde",
     "reth-eth-wire-types/serde",
@@ -125,6 +124,7 @@ serde = [
     "reth-network-api/serde",
     "rand_08/serde",
     "reth-storage-api/serde",
+    "reth-network-types/serde",
 ]
 test-utils = [
     "reth-transaction-pool/test-utils",

crates/net/network/src/discovery.rs

@@ -301,6 +301,20 @@ impl Discovery {
     }
 }
 
+impl Drop for Discovery {
+    fn drop(&mut self) {
+        if let Some(discv4) = &self.discv4 {
+            discv4.terminate();
+        }
+        if let Some(handle) = self._discv4_service.take() {
+            handle.abort();
+        }
+        if let Some(handle) = self._dns_disc_service.take() {
+            handle.abort();
+        }
+    }
+}
+
 impl Stream for Discovery {
     type Item = DiscoveryEvent;
 

crates/net/network/src/manager.rs

@@ -431,27 +431,30 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
 
     /// Returns an iterator over all peers in the peer set.
     pub fn all_peers(&self) -> impl Iterator<Item = NodeRecord> + '_ {
-        self.swarm.state().peers().iter_peers()
+        self.swarm.peers().iter_peers()
     }
 
     /// Returns the number of peers in the peer set.
     pub fn num_known_peers(&self) -> usize {
-        self.swarm.state().peers().num_known_peers()
+        self.swarm.peers().num_known_peers()
     }
 
     /// Returns a new [`PeersHandle`] that can be cloned and shared.
     ///
     /// The [`PeersHandle`] can be used to interact with the network's peer set.
     pub fn peers_handle(&self) -> PeersHandle {
-        self.swarm.state().peers().handle()
+        self.swarm.peers().handle()
     }
 
     /// Collect the peers from the [`NetworkManager`] and write them to the given
     /// `persistent_peers_file`.
+    ///
+    /// Only persists peers that are not currently backed off or banned. Includes metadata like
+    /// peer kind, fork ID, and reputation.
     pub fn write_peers_to_file(&self, persistent_peers_file: &Path) -> Result<(), FsPathError> {
-        let known_peers = self.all_peers().collect::<Vec<_>>();
+        let peers = self.swarm.peers().persistable_peers().collect::<Vec<_>>();
         persistent_peers_file.parent().map(fs::create_dir_all).transpose()?;
-        reth_fs_util::write_json_file(persistent_peers_file, &known_peers)?;
+        reth_fs_util::write_json_file(persistent_peers_file, &peers)?;
         Ok(())
     }
 
@@ -727,10 +730,10 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 let _ = tx.send(());
             }
             NetworkHandleMessage::ReputationChange(peer_id, kind) => {
-                self.swarm.state_mut().peers_mut().apply_reputation_change(&peer_id, kind);
+                self.swarm.peers_mut().apply_reputation_change(&peer_id, kind);
             }
             NetworkHandleMessage::GetReputationById(peer_id, tx) => {
-                let _ = tx.send(self.swarm.state_mut().peers().get_reputation(&peer_id));
+                let _ = tx.send(self.swarm.peers().get_reputation(&peer_id));
             }
             NetworkHandleMessage::FetchClient(tx) => {
                 let _ = tx.send(self.fetch_client());
@@ -753,7 +756,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 let _ = tx.send(self.get_peer_infos_by_ids(peer_ids));
             }
             NetworkHandleMessage::GetPeerInfosByPeerKind(kind, tx) => {
-                let peer_ids = self.swarm.state().peers().peers_by_kind(kind);
+                let peer_ids = self.swarm.peers().peers_by_kind(kind);
                 let _ = tx.send(self.get_peer_infos_by_ids(peer_ids));
             }
             NetworkHandleMessage::AddRlpxSubProtocol(proto) => self.add_rlpx_sub_protocol(proto),
@@ -788,7 +791,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 self.metrics.total_incoming_connections.increment(1);
                 self.metrics
                     .incoming_connections
-                    .set(self.swarm.state().peers().num_inbound_connections() as f64);
+                    .set(self.swarm.peers().num_inbound_connections() as f64);
             }
             SwarmEvent::OutgoingTcpConnection { remote_addr, peer_id } => {
                 trace!(target: "net", ?remote_addr, ?peer_id, "Starting outbound connection.");
@@ -826,7 +829,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 }
 
                 if direction.is_outgoing() {
-                    self.swarm.state_mut().peers_mut().on_active_outgoing_established(peer_id);
+                    self.swarm.peers_mut().on_active_outgoing_established(peer_id);
                 }
 
                 self.update_active_connection_metrics();
@@ -854,12 +857,12 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
             SwarmEvent::PeerAdded(peer_id) => {
                 trace!(target: "net", ?peer_id, "Peer added");
                 self.event_sender.notify(NetworkEvent::Peer(PeerEvent::PeerAdded(peer_id)));
-                self.metrics.tracked_peers.set(self.swarm.state().peers().num_known_peers() as f64);
+                self.metrics.tracked_peers.set(self.swarm.peers().num_known_peers() as f64);
             }
             SwarmEvent::PeerRemoved(peer_id) => {
                 trace!(target: "net", ?peer_id, "Peer dropped");
                 self.event_sender.notify(NetworkEvent::Peer(PeerEvent::PeerRemoved(peer_id)));
-                self.metrics.tracked_peers.set(self.swarm.state().peers().num_known_peers() as f64);
+                self.metrics.tracked_peers.set(self.swarm.peers().num_known_peers() as f64);
             }
             SwarmEvent::SessionClosed { peer_id, remote_addr, error } => {
                 let total_active = self.num_active_peers.fetch_sub(1, Ordering::Relaxed) - 1;
@@ -874,23 +877,19 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 );
 
                 // Capture direction before state is reset to Idle
-                let is_inbound = self.swarm.state().peers().is_inbound_peer(&peer_id);
+                let is_inbound = self.swarm.peers().is_inbound_peer(&peer_id);
 
                 let reason = if let Some(ref err) = error {
                     // If the connection was closed due to an error, we report
                     // the peer
-                    self.swarm.state_mut().peers_mut().on_active_session_dropped(
-                        &remote_addr,
-                        &peer_id,
-                        err,
-                    );
+                    self.swarm.peers_mut().on_active_session_dropped(&remote_addr, &peer_id, err);
                     self.backed_off_peers_metrics.increment_for_reason(
                         BackoffReason::from_disconnect(err.as_disconnected()),
                     );
                     err.as_disconnected()
                 } else {
                     // Gracefully disconnected
-                    self.swarm.state_mut().peers_mut().on_active_session_gracefully_closed(peer_id);
+                    self.swarm.peers_mut().on_active_session_gracefully_closed(peer_id);
                     self.backed_off_peers_metrics
                         .increment_for_reason(BackoffReason::GracefulClose);
                     None
@@ -905,9 +904,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                         self.disconnect_metrics.increment_outbound(reason);
                     }
                 }
-                self.metrics
-                    .backed_off_peers
-                    .set(self.swarm.state().peers().num_backed_off_peers() as f64);
+                self.metrics.backed_off_peers.set(self.swarm.peers().num_backed_off_peers() as f64);
                 self.event_sender
                     .notify(NetworkEvent::Peer(PeerEvent::SessionClosed { peer_id, reason }));
             }
@@ -937,7 +934,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 self.closed_sessions_metrics.incoming_pending.increment(1);
                 self.metrics
                     .incoming_connections
-                    .set(self.swarm.state().peers().num_inbound_connections() as f64);
+                    .set(self.swarm.peers().num_inbound_connections() as f64);
             }
             SwarmEvent::OutgoingPendingSessionClosed { remote_addr, peer_id, error } => {
                 trace!(
@@ -949,7 +946,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 );
 
                 if let Some(ref err) = error {
-                    self.swarm.state_mut().peers_mut().on_outgoing_pending_session_dropped(
+                    self.swarm.peers_mut().on_outgoing_pending_session_dropped(
                         &remote_addr,
                         &peer_id,
                         err,
@@ -969,9 +966,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 }
                 self.closed_sessions_metrics.outgoing_pending.increment(1);
                 self.update_pending_connection_metrics();
-                self.metrics
-                    .backed_off_peers
-                    .set(self.swarm.state().peers().num_backed_off_peers() as f64);
+                self.metrics.backed_off_peers.set(self.swarm.peers().num_backed_off_peers() as f64);
             }
             SwarmEvent::OutgoingConnectionError { remote_addr, peer_id, error } => {
                 trace!(
@@ -982,16 +977,14 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                     "Outgoing connection error"
                 );
 
-                self.swarm.state_mut().peers_mut().on_outgoing_connection_failure(
+                self.swarm.peers_mut().on_outgoing_connection_failure(
                     &remote_addr,
                     &peer_id,
                     &error,
                 );
 
                 self.backed_off_peers_metrics.increment_for_reason(BackoffReason::ConnectionError);
-                self.metrics
-                    .backed_off_peers
-                    .set(self.swarm.state().peers().num_backed_off_peers() as f64);
+                self.metrics.backed_off_peers.set(self.swarm.peers().num_backed_off_peers() as f64);
                 self.update_pending_connection_metrics();
             }
             SwarmEvent::BadMessage { peer_id } => {
@@ -1049,12 +1042,8 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
     /// Updates the metrics for active,established connections
     #[inline]
     fn update_active_connection_metrics(&self) {
-        self.metrics
-            .incoming_connections
-            .set(self.swarm.state().peers().num_inbound_connections() as f64);
-        self.metrics
-            .outgoing_connections
-            .set(self.swarm.state().peers().num_outbound_connections() as f64);
+        self.metrics.incoming_connections.set(self.swarm.peers().num_inbound_connections() as f64);
+        self.metrics.outgoing_connections.set(self.swarm.peers().num_outbound_connections() as f64);
     }
 
     /// Updates the metrics for pending connections
@@ -1062,7 +1051,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
     fn update_pending_connection_metrics(&self) {
         self.metrics
             .pending_outgoing_connections
-            .set(self.swarm.state().peers().num_pending_outbound_connections() as f64);
+            .set(self.swarm.peers().num_pending_outbound_connections() as f64);
         self.metrics
             .total_pending_connections
             .set(self.swarm.sessions().num_pending_connections() as f64);

crates/net/network/src/peers.rs

@@ -20,7 +20,7 @@ use reth_network_types::{
         reputation::{DEFAULT_REPUTATION, MAX_TRUSTED_PEER_REPUTATION_CHANGE},
     },
     ConnectionsConfig, Peer, PeerAddr, PeerConnectionState, PeerKind, PeersConfig,
-    ReputationChangeKind, ReputationChangeOutcome, ReputationChangeWeights,
+    PersistedPeerInfo, ReputationChangeKind, ReputationChangeOutcome, ReputationChangeWeights,
 };
 use std::{
     collections::{hash_map::Entry, HashMap, HashSet, VecDeque},
@@ -111,6 +111,7 @@ impl PeersManager {
             trusted_nodes_only,
             trusted_nodes_resolution_interval,
             basic_nodes,
+            persisted_peers,
             max_backoff_count,
             incoming_ip_throttle_duration,
             ip_filter,
@@ -122,7 +123,8 @@ impl PeersManager {
         // We use half of the interval to decrease the max duration to `150%` in worst case
         let unban_interval = ban_duration.min(backoff_durations.low) / 2;
 
-        let mut peers = HashMap::with_capacity(trusted_nodes.len() + basic_nodes.len());
+        let mut peers =
+            HashMap::with_capacity(trusted_nodes.len() + basic_nodes.len() + persisted_peers.len());
         let mut trusted_peer_ids = HashSet::with_capacity(trusted_nodes.len());
 
         for trusted_peer in &trusted_nodes {
@@ -139,6 +141,19 @@ impl PeersManager {
             }
         }
 
+        for PersistedPeerInfo { record, kind, fork_id, reputation } in persisted_peers {
+            let NodeRecord { address, tcp_port, udp_port, id } = record;
+            peers.entry(id).or_insert_with(|| {
+                let mut peer = Peer::with_kind(
+                    PeerAddr::new_with_ports(address, tcp_port, Some(udp_port)),
+                    kind,
+                );
+                peer.fork_id = fork_id.map(Box::new);
+                peer.reputation = reputation;
+                peer
+            });
+        }
+
         for NodeRecord { address, tcp_port, udp_port, id } in basic_nodes {
             peers.entry(id).or_insert_with(|| {
                 Peer::new(PeerAddr::new_with_ports(address, tcp_port, Some(udp_port)))
@@ -191,7 +206,7 @@ impl PeersManager {
         self.peers.len()
     }
 
-    /// Returns an iterator over all peers
+    /// Returns an iterator over all peers as [`NodeRecord`]s.
     pub(crate) fn iter_peers(&self) -> impl Iterator<Item = NodeRecord> + '_ {
         self.peers.iter().map(|(peer_id, v)| {
             NodeRecord::new_with_ports(
@@ -203,6 +218,26 @@ impl PeersManager {
         })
     }
 
+    /// Returns an iterator over peers suitable for persisting to disk.
+    ///
+    /// Filters out backed-off and banned peers, and includes metadata like kind, fork ID, and
+    /// reputation.
+    pub(crate) fn persistable_peers(&self) -> impl Iterator<Item = PersistedPeerInfo> + '_ {
+        self.peers.iter().filter(|(_, peer)| !peer.is_backed_off() && !peer.is_banned()).map(
+            |(peer_id, peer)| PersistedPeerInfo {
+                record: NodeRecord::new_with_ports(
+                    peer.addr.tcp().ip(),
+                    peer.addr.tcp().port(),
+                    peer.addr.udp().map(|addr| addr.port()),
+                    *peer_id,
+                ),
+                kind: peer.kind,
+                fork_id: peer.fork_id.as_deref().copied(),
+                reputation: peer.reputation,
+            },
+        )
+    }
+
     /// Returns the `NodeRecord` and `PeerKind` for the given peer id
     pub(crate) fn peer_by_id(&self, peer_id: PeerId) -> Option<(NodeRecord, PeerKind)> {
         self.peers.get(&peer_id).map(|v| {
@@ -827,7 +862,7 @@ impl PeersManager {
             }
         }
 
-        if kind.filter(|kind| kind.is_trusted()).is_some() {
+        if kind.is_some_and(|kind| kind.is_trusted()) {
             // also track the peer in the peer id set
             self.trusted_peer_ids.insert(peer_id);
         }
@@ -939,11 +974,14 @@ impl PeersManager {
         self.trusted_peer_ids.remove(&peer_id);
     }
 
-    /// Returns the idle peer with the highest reputation.
+    /// Returns the best idle peer to connect to.
     ///
     /// Peers that are `trusted` or `static`, see [`PeerKind`], are prioritized as long as they're
     /// not currently marked as banned or backed off.
     ///
+    /// Among remaining peers, the one with the highest reputation is selected. When reputation is
+    /// equal, a peer with a discovered `fork_id` is preferred since it indicates a compatible fork.
+    ///
     /// If `trusted_nodes_only` is enabled, see [`PeersConfig`], then this will only consider
     /// `trusted` peers.
     ///
@@ -969,9 +1007,15 @@ impl PeersManager {
                 return Some((*maybe_better.0, maybe_better.1))
             }
 
-            // otherwise we keep track of the best peer using the reputation
-            if maybe_better.1.reputation > best_peer.1.reputation {
-                best_peer = maybe_better;
+            // prefer higher reputation, break ties by fork_id presence
+            match maybe_better.1.reputation.cmp(&best_peer.1.reputation) {
+                std::cmp::Ordering::Greater => best_peer = maybe_better,
+                std::cmp::Ordering::Equal
+                    if maybe_better.1.fork_id.is_some() && best_peer.1.fork_id.is_none() =>
+                {
+                    best_peer = maybe_better
+                }
+                _ => {}
             }
         }
         Some((*best_peer.0, best_peer.1))
@@ -1294,6 +1338,7 @@ mod tests {
         errors::{EthHandshakeError, EthStreamError, P2PHandshakeError, P2PStreamError},
         DisconnectReason,
     };
+    use reth_ethereum_forks::{ForkHash, ForkId};
     use reth_net_banlist::BanList;
     use reth_network_api::Direction;
     use reth_network_peers::{PeerId, TrustedPeer};
@@ -3238,4 +3283,23 @@ mod tests {
         assert!(peers.on_incoming_pending_session(ip2).is_ok());
         assert!(peers.on_incoming_pending_session(ip3).is_ok());
     }
+
+    #[tokio::test]
+    async fn test_best_unconnected_prefers_fork_id_as_tiebreaker() {
+        let mut peers = PeersManager::default();
+        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8008);
+
+        let fork_id = ForkId { hash: ForkHash([0xaa, 0xbb, 0xcc, 0xdd]), next: 0 };
+
+        // add two peers with equal reputation, only one has a fork_id
+        let no_fork = PeerId::random();
+        peers.add_peer(no_fork, PeerAddr::from_tcp(addr), None);
+
+        let with_fork = PeerId::random();
+        peers.add_peer(with_fork, PeerAddr::from_tcp(addr), None);
+        peers.peers.get_mut(&with_fork).unwrap().fork_id = Some(Box::new(fork_id));
+
+        let (best_id, _) = peers.best_unconnected().unwrap();
+        assert_eq!(best_id, with_fork, "fork_id should break tie when reputation is equal");
+    }
 }

crates/net/network/src/swarm.rs

@@ -1,7 +1,7 @@
 use crate::{
     listener::{ConnectionListener, ListenerEvent},
     message::PeerMessage,
-    peers::InboundConnectionError,
+    peers::{InboundConnectionError, PeersManager},
     protocol::IntoRlpxSubProtocol,
     session::{Direction, PendingSessionHandshakeError, SessionEvent, SessionId, SessionManager},
     state::{NetworkState, StateAction},
@@ -98,6 +98,16 @@ impl<N: NetworkPrimitives> Swarm<N> {
     pub(crate) const fn sessions_mut(&mut self) -> &mut SessionManager<N> {
         &mut self.sessions
     }
+
+    /// Access to the [`PeersManager`].
+    pub(crate) const fn peers(&self) -> &PeersManager {
+        self.state.peers()
+    }
+
+    /// Mutable access to the [`PeersManager`].
+    pub(crate) const fn peers_mut(&mut self) -> &mut PeersManager {
+        self.state.peers_mut()
+    }
 }
 
 impl<N: NetworkPrimitives> Swarm<N> {
@@ -190,9 +200,7 @@ impl<N: NetworkPrimitives> Swarm<N> {
                     return None
                 }
                 // ensure we can handle an incoming connection from this address
-                if let Err(err) =
-                    self.state_mut().peers_mut().on_incoming_pending_session(remote_addr.ip())
-                {
+                if let Err(err) = self.peers_mut().on_incoming_pending_session(remote_addr.ip()) {
                     match err {
                         InboundConnectionError::IpBanned => {
                             trace!(target: "net", ?remote_addr, "The incoming ip address is in the ban list");
@@ -256,21 +264,21 @@ impl<N: NetworkPrimitives> Swarm<N> {
                 //
                 // When disabled (default), peers without a fork ID are admitted immediately.
                 // Peers that *do* carry a fork ID are always validated against ours.
-                let enforce = self.state().peers().enforce_enr_fork_id();
+                let enforce = self.peers().enforce_enr_fork_id();
                 let allow = match fork_id {
                     Some(f) => self.sessions.is_valid_fork_id(f),
                     None => !enforce,
                 };
                 if allow {
-                    self.state_mut().peers_mut().add_peer(peer_id, addr, fork_id);
+                    self.peers_mut().add_peer(peer_id, addr, fork_id);
                 }
             }
             StateAction::DiscoveredEnrForkId { peer_id, addr, fork_id } => {
                 if self.sessions.is_valid_fork_id(fork_id) {
-                    self.state_mut().peers_mut().add_peer(peer_id, addr, Some(fork_id));
+                    self.peers_mut().add_peer(peer_id, addr, Some(fork_id));
                 } else {
                     trace!(target: "net", ?peer_id, remote_fork_id=?fork_id, our_fork_id=?self.sessions.fork_id(), "fork id mismatch, removing peer");
-                    self.state_mut().peers_mut().remove_peer(peer_id);
+                    self.peers_mut().remove_peer(peer_id);
                 }
             }
         }
@@ -279,18 +287,18 @@ impl<N: NetworkPrimitives> Swarm<N> {
 
     /// Set network connection state to `ShuttingDown`
     pub(crate) const fn on_shutdown_requested(&mut self) {
-        self.state_mut().peers_mut().on_shutdown();
+        self.peers_mut().on_shutdown();
     }
 
     /// Checks if the node's network connection state is '`ShuttingDown`'
     #[inline]
     pub(crate) const fn is_shutting_down(&self) -> bool {
-        self.state().peers().connection_state().is_shutting_down()
+        self.peers().connection_state().is_shutting_down()
     }
 
     /// Set network connection state to `Hibernate` or `Active`
     pub(crate) const fn on_network_state_change(&mut self, network_state: NetworkConnectionState) {
-        self.state_mut().peers_mut().on_network_state_change(network_state);
+        self.peers_mut().on_network_state_change(network_state);
     }
 }
 

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

@@ -904,7 +904,7 @@ where
         // send hashes if any
         if let Some(new_pooled_hashes) = pooled {
             for hash in new_pooled_hashes.iter_hashes().copied() {
-                propagated.0.entry(hash).or_default().push(PropagateKind::Hash(peer_id));
+                propagated.record(hash, PropagateKind::Hash(peer_id));
                 // mark transaction as seen by peer
                 peer.seen_transactions.insert(hash);
             }
@@ -916,7 +916,7 @@ where
         // send full transactions, if any
         if let Some(new_full_transactions) = full {
             for tx in &new_full_transactions {
-                propagated.0.entry(*tx.tx_hash()).or_default().push(PropagateKind::Full(peer_id));
+                propagated.record(*tx.tx_hash(), PropagateKind::Full(peer_id));
                 // mark transaction as seen by peer
                 peer.seen_transactions.insert(*tx.tx_hash());
             }
@@ -926,7 +926,7 @@ where
         }
 
         // Update propagated transactions metrics
-        self.metrics.propagated_transactions.increment(propagated.0.len() as u64);
+        self.metrics.propagated_transactions.increment(propagated.len() as u64);
 
         Some(propagated)
     }
@@ -977,7 +977,7 @@ where
             }
 
             for hash in new_pooled_hashes.iter_hashes().copied() {
-                propagated.0.entry(hash).or_default().push(PropagateKind::Hash(peer_id));
+                propagated.record(hash, PropagateKind::Hash(peer_id));
             }
 
             trace!(target: "net::tx::propagation", ?peer_id, ?new_pooled_hashes, "Propagating transactions to peer");
@@ -986,7 +986,7 @@ where
             self.network.send_transactions_hashes(peer_id, new_pooled_hashes);
 
             // Update propagated transactions metrics
-            self.metrics.propagated_transactions.increment(propagated.0.len() as u64);
+            self.metrics.propagated_transactions.increment(propagated.len() as u64);
 
             propagated
         };
@@ -1057,7 +1057,7 @@ where
                     .truncate(SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE);
 
                 for hash in new_pooled_hashes.iter_hashes().copied() {
-                    propagated.0.entry(hash).or_default().push(PropagateKind::Hash(*peer_id));
+                    propagated.record(hash, PropagateKind::Hash(*peer_id));
                     // mark transaction as seen by peer
                     peer.seen_transactions.insert(hash);
                 }
@@ -1071,11 +1071,7 @@ where
             // send full transactions, if any
             if let Some(new_full_transactions) = full {
                 for tx in &new_full_transactions {
-                    propagated
-                        .0
-                        .entry(*tx.tx_hash())
-                        .or_default()
-                        .push(PropagateKind::Full(*peer_id));
+                    propagated.record(*tx.tx_hash(), PropagateKind::Full(*peer_id));
                     // mark transaction as seen by peer
                     peer.seen_transactions.insert(*tx.tx_hash());
                 }
@@ -1088,7 +1084,7 @@ where
         }
 
         // Update propagated transactions metrics
-        self.metrics.propagated_transactions.increment(propagated.0.len() as u64);
+        self.metrics.propagated_transactions.increment(propagated.len() as u64);
 
         propagated
     }
@@ -1236,7 +1232,7 @@ where
             msg_builder.push_pooled(pooled_tx);
         }
 
-        debug!(target: "net::tx", ?peer_id, tx_count = msg_builder.is_empty(), "Broadcasting transaction hashes");
+        debug!(target: "net::tx", ?peer_id, tx_count = msg_builder.len(), "Broadcasting transaction hashes");
         let msg = msg_builder.build();
         self.network.send_transactions_hashes(peer_id, msg);
     }
@@ -1924,6 +1920,14 @@ impl PooledTransactionsHashesBuilder {
         }
     }
 
+    /// Returns the number of transactions in the builder.
+    fn len(&self) -> usize {
+        match self {
+            Self::Eth66(hashes) => hashes.len(),
+            Self::Eth68(hashes) => hashes.len(),
+        }
+    }
+
     /// Appends all hashes
     fn extend<T: SignedTransaction>(
         &mut self,
@@ -2903,12 +2907,12 @@ mod tests {
 
         let propagated =
             tx_manager.propagate_transactions(propagate.clone(), PropagationMode::Basic);
-        assert_eq!(propagated.0.len(), 2);
-        let prop_txs = propagated.0.get(eip1559_tx.transaction.hash()).unwrap();
+        assert_eq!(propagated.len(), 2);
+        let prop_txs = propagated.get(eip1559_tx.transaction.hash()).unwrap();
         assert_eq!(prop_txs.len(), 1);
         assert!(prop_txs[0].is_full());
 
-        let prop_txs = propagated.0.get(eip4844_tx.transaction.hash()).unwrap();
+        let prop_txs = propagated.get(eip4844_tx.transaction.hash()).unwrap();
         assert_eq!(prop_txs.len(), 1);
         assert!(prop_txs[0].is_hash());
 
@@ -2919,7 +2923,7 @@ mod tests {
 
         // propagate again
         let propagated = tx_manager.propagate_transactions(propagate, PropagationMode::Basic);
-        assert!(propagated.0.is_empty());
+        assert!(propagated.is_empty());
     }
 
     #[tokio::test]

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

@@ -9,7 +9,7 @@ use crate::{
     NodeBuilderWithComponents, NodeComponents, NodeComponentsBuilder, NodeHandle, NodeTypesAdapter,
 };
 use alloy_consensus::BlockHeader;
-use futures::{stream_select, FutureExt, StreamExt};
+use futures::{stream::FusedStream, stream_select, FutureExt, StreamExt};
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_engine_tree::{
     chain::{ChainEvent, FromOrchestrator},
@@ -356,7 +356,7 @@ impl EngineNodeLauncher {
                             }
                         }
                     }
-                    payload = built_payloads.select_next_some() => {
+                    payload = built_payloads.select_next_some(), if !built_payloads.is_terminated() => {
                         if let Some(executed_block) = payload.executed_block() {
                             debug!(target: "reth::cli", block=?executed_block.recovered_block.num_hash(),  "inserting built payload");
                             orchestrator.handler_mut().handler_mut().on_event(EngineApiRequest::InsertExecutedBlock(executed_block.into_executed_payload()).into());

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

@@ -24,6 +24,8 @@ pub struct ExExLauncher<Node: FullNodeComponents> {
     config_container: WithConfigs<<Node::Types as NodeTypes>::ChainSpec>,
     /// The threshold for the number of blocks in the WAL before emitting a warning.
     wal_blocks_warning: usize,
+    /// The max notification buffer capacity for the ExEx manager.
+    capacity: usize,
 }
 
 impl<Node: FullNodeComponents + Clone> ExExLauncher<Node> {
@@ -40,6 +42,7 @@ impl<Node: FullNodeComponents + Clone> ExExLauncher<Node> {
             components,
             config_container,
             wal_blocks_warning: DEFAULT_WAL_BLOCKS_WARNING,
+            capacity: DEFAULT_EXEX_MANAGER_CAPACITY,
         }
     }
 
@@ -53,6 +56,12 @@ impl<Node: FullNodeComponents + Clone> ExExLauncher<Node> {
         self
     }
 
+    /// Sets the max notification buffer capacity for the [`ExExManager`].
+    pub const fn with_capacity(mut self, capacity: usize) -> Self {
+        self.capacity = capacity;
+        self
+    }
+
     /// Launches all execution extensions.
     ///
     /// Spawns all extensions and returns the handle to the exex manager if any extensions are
@@ -60,7 +69,8 @@ impl<Node: FullNodeComponents + Clone> ExExLauncher<Node> {
     pub async fn launch(
         self,
     ) -> eyre::Result<Option<ExExManagerHandle<PrimitivesTy<Node::Types>>>> {
-        let Self { head, extensions, components, config_container, wal_blocks_warning } = self;
+        let Self { head, extensions, components, config_container, wal_blocks_warning, capacity } =
+            self;
         let head = BlockNumHash::new(head.number, head.hash);
 
         if extensions.is_empty() {
@@ -134,7 +144,7 @@ impl<Node: FullNodeComponents + Clone> ExExLauncher<Node> {
         let exex_manager = ExExManager::new(
             components.provider().clone(),
             exex_handles,
-            DEFAULT_EXEX_MANAGER_CAPACITY,
+            capacity,
             exex_wal,
             components.provider().finalized_block_stream(),
         )

crates/node/builder/src/rpc.rs

@@ -1020,7 +1020,13 @@ where
             .with_executor(node.task_executor().clone())
             .with_evm_config(node.evm_config().clone())
             .with_consensus(node.consensus().clone())
-            .build_with_auth_server(module_config, engine_api, eth_api, engine_events.clone());
+            .build_with_auth_server(
+                module_config,
+                engine_api,
+                eth_api,
+                engine_events.clone(),
+                beacon_engine_handle.clone(),
+            );
 
         // in dev mode we generate 20 random dev-signer accounts
         if config.dev.dev {

crates/node/core/Cargo.toml

@@ -59,7 +59,6 @@ url.workspace = true
 ipnet.workspace = true
 # io
 dirs-next.workspace = true
-shellexpand.workspace = true
 
 # obs
 tracing.workspace = true

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

@@ -78,6 +78,10 @@ pub struct BenchmarkArgs {
     /// and returns server-side timing breakdowns (latency, persistence wait, cache wait).
     #[arg(long, default_value = "false", verbatim_doc_comment)]
     pub reth_new_payload: bool,
+
+    /// Fetch and replay RLP-encoded blocks. Implies `reth_new_payload`.
+    #[arg(long, default_value = "false", verbatim_doc_comment)]
+    pub rlp_blocks: bool,
 }
 
 #[cfg(test)]

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

@@ -29,7 +29,6 @@ pub struct DefaultEngineValues {
     cross_block_cache_size: usize,
     state_root_task_compare_updates: bool,
     accept_execution_requests_hash: bool,
-    multiproof_chunking_enabled: bool,
     multiproof_chunk_size: usize,
     reserved_cpu_cores: usize,
     precompile_cache_disabled: bool,
@@ -111,12 +110,6 @@ impl DefaultEngineValues {
         self
     }
 
-    /// Set whether to enable multiproof chunking by default
-    pub const fn with_multiproof_chunking_enabled(mut self, v: bool) -> Self {
-        self.multiproof_chunking_enabled = v;
-        self
-    }
-
     /// Set the default multiproof chunk size
     pub const fn with_multiproof_chunk_size(mut self, v: usize) -> Self {
         self.multiproof_chunk_size = v;
@@ -217,7 +210,6 @@ impl Default for DefaultEngineValues {
             cross_block_cache_size: DEFAULT_CROSS_BLOCK_CACHE_SIZE_MB,
             state_root_task_compare_updates: false,
             accept_execution_requests_hash: false,
-            multiproof_chunking_enabled: true,
             multiproof_chunk_size: DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE,
             reserved_cpu_cores: DEFAULT_RESERVED_CPU_CORES,
             precompile_cache_disabled: false,
@@ -300,10 +292,6 @@ pub struct EngineArgs {
     #[arg(long = "engine.accept-execution-requests-hash", default_value_t = DefaultEngineValues::get_global().accept_execution_requests_hash)]
     pub accept_execution_requests_hash: bool,
 
-    /// Whether multiproof task should chunk proof targets.
-    #[arg(long = "engine.multiproof-chunking", default_value_t = DefaultEngineValues::get_global().multiproof_chunking_enabled)]
-    pub multiproof_chunking_enabled: bool,
-
     /// Multiproof task chunk size for proof targets.
     #[arg(long = "engine.multiproof-chunk-size", default_value_t = DefaultEngineValues::get_global().multiproof_chunk_size)]
     pub multiproof_chunk_size: usize,
@@ -404,7 +392,6 @@ impl Default for EngineArgs {
             cross_block_cache_size,
             state_root_task_compare_updates,
             accept_execution_requests_hash,
-            multiproof_chunking_enabled,
             multiproof_chunk_size,
             reserved_cpu_cores,
             precompile_cache_disabled,
@@ -433,7 +420,6 @@ impl Default for EngineArgs {
             state_provider_metrics,
             cross_block_cache_size,
             accept_execution_requests_hash,
-            multiproof_chunking_enabled,
             multiproof_chunk_size,
             reserved_cpu_cores,
             precompile_cache_enabled: true,
@@ -467,7 +453,6 @@ impl EngineArgs {
             .with_state_provider_metrics(self.state_provider_metrics)
             .with_always_compare_trie_updates(self.state_root_task_compare_updates)
             .with_cross_block_cache_size(self.cross_block_cache_size * 1024 * 1024)
-            .with_multiproof_chunking_enabled(self.multiproof_chunking_enabled)
             .with_multiproof_chunk_size(self.multiproof_chunk_size)
             .with_reserved_cpu_cores(self.reserved_cpu_cores)
             .without_precompile_cache(self.precompile_cache_disabled)
@@ -476,8 +461,6 @@ impl EngineArgs {
                 self.always_process_payload_attributes_on_canonical_head,
             )
             .with_unwind_canonical_header(self.allow_unwind_canonical_header)
-            .with_storage_worker_count_opt(self.storage_worker_count)
-            .with_account_worker_count_opt(self.account_worker_count)
             .without_cache_metrics(self.cache_metrics_disabled)
             .with_sparse_trie_prune_depth(self.sparse_trie_prune_depth)
             .with_sparse_trie_max_storage_tries(self.sparse_trie_max_storage_tries)
@@ -521,7 +504,6 @@ mod tests {
             cross_block_cache_size: 256,
             state_root_task_compare_updates: true,
             accept_execution_requests_hash: true,
-            multiproof_chunking_enabled: true,
             multiproof_chunk_size: 512,
             reserved_cpu_cores: 4,
             precompile_cache_enabled: true,
@@ -553,7 +535,6 @@ mod tests {
             "256",
             "--engine.state-root-task-compare-updates",
             "--engine.accept-execution-requests-hash",
-            "--engine.multiproof-chunking",
             "--engine.multiproof-chunk-size",
             "512",
             "--engine.reserved-cpu-cores",

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

@@ -1104,9 +1104,9 @@ mod tests {
 
         let net_cfg = builder.build_with_noop_provider(MAINNET.clone());
 
-        // Assert basic_nodes contains our node
+        // Assert persisted_peers contains our node (legacy format is auto-converted)
         let node: NodeRecord = enode.parse().unwrap();
-        assert!(net_cfg.peers_config.basic_nodes.contains(&node));
+        assert!(net_cfg.peers_config.persisted_peers.iter().any(|p| p.record == node));
 
         // Cleanup
         let _ = fs::remove_file(&peers_file);

crates/node/core/src/dirs.rs

@@ -3,7 +3,7 @@
 use crate::{args::DatadirArgs, utils::parse_path};
 use reth_chainspec::Chain;
 use std::{
-    env::VarError,
+    convert::Infallible,
     fmt::{Debug, Display, Formatter},
     path::{Path, PathBuf},
     str::FromStr,
@@ -86,8 +86,7 @@ pub trait XdgPath {
 /// A wrapper type that either parses a user-given path or defaults to an
 /// OS-specific path.
 ///
-/// The [`FromStr`] implementation supports shell expansions and common patterns such as `~` for the
-/// home directory.
+/// The [`FromStr`] implementation parses a string into a path.
 ///
 /// # Example
 ///
@@ -127,10 +126,10 @@ impl<D: XdgPath> Default for PlatformPath<D> {
 }
 
 impl<D> FromStr for PlatformPath<D> {
-    type Err = shellexpand::LookupError<VarError>;
+    type Err = Infallible;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        Ok(Self(parse_path(s)?, std::marker::PhantomData))
+        Ok(Self(parse_path(s), std::marker::PhantomData))
     }
 }
 
@@ -235,7 +234,7 @@ impl<D> Default for MaybePlatformPath<D> {
 }
 
 impl<D> FromStr for MaybePlatformPath<D> {
-    type Err = shellexpand::LookupError<VarError>;
+    type Err = Infallible;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         let p = match s {

crates/node/core/src/utils.rs

@@ -10,16 +10,12 @@ use reth_network_p2p::{
     bodies::client::BodiesClient, headers::client::HeadersClient, priority::Priority,
 };
 use reth_primitives_traits::{Block, SealedBlock, SealedHeader};
-use std::{
-    env::VarError,
-    path::{Path, PathBuf},
-};
+use std::path::{Path, PathBuf};
 use tracing::{debug, info};
 
-/// Parses a user-specified path with support for environment variables and common shorthands (e.g.
-/// ~ for the user's home directory).
-pub fn parse_path(value: &str) -> Result<PathBuf, shellexpand::LookupError<VarError>> {
-    shellexpand::full(value).map(|path| PathBuf::from(path.into_owned()))
+/// Parses a user-specified path into a [`PathBuf`].
+pub fn parse_path(value: &str) -> PathBuf {
+    PathBuf::from(value)
 }
 
 /// Attempts to retrieve or create a JWT secret from the specified path.

crates/primitives-traits/src/lib.rs

@@ -168,7 +168,7 @@ pub use alloy_primitives::{logs_bloom, Log, LogData};
 pub mod proofs;
 
 mod storage;
-pub use storage::{StorageEntry, StorageSlotKey, ValueWithSubKey};
+pub use storage::{StorageEntry, ValueWithSubKey};
 
 pub mod sync;
 

crates/primitives-traits/src/serde_bincode_compat.rs

@@ -145,9 +145,10 @@ impl<T: RlpBincode + 'static> SerdeBincodeCompat for T {
 mod block_bincode {
     use crate::serde_bincode_compat::SerdeBincodeCompat;
     use alloc::{borrow::Cow, vec::Vec};
-    use alloy_consensus::TxEip4844;
+    use alloy_consensus::TxTy;
     use alloy_eips::eip4895::Withdrawals;
-    use serde::{Deserialize, Deserializer, Serialize, Serializer};
+    use core::fmt::Debug;
+    use serde::{de::DeserializeOwned, Deserialize, Deserializer, Serialize, Serializer};
     use serde_with::{DeserializeAs, SerializeAs};
 
     /// Bincode-compatible [`alloy_consensus::Block`] serde implementation.
@@ -319,9 +320,11 @@ mod block_bincode {
         }
     }
 
-    impl super::SerdeBincodeCompat for alloy_consensus::EthereumTxEnvelope<TxEip4844> {
+    impl<T: Clone + Serialize + DeserializeOwned + Debug + 'static> super::SerdeBincodeCompat
+        for alloy_consensus::EthereumTxEnvelope<T>
+    {
         type BincodeRepr<'a> =
-            alloy_consensus::serde_bincode_compat::transaction::EthereumTxEnvelope<'a>;
+            alloy_consensus::serde_bincode_compat::transaction::EthereumTxEnvelope<'a, T>;
 
         fn as_repr(&self) -> Self::BincodeRepr<'_> {
             self.into()
@@ -346,8 +349,10 @@ mod block_bincode {
         }
     }
 
-    impl super::SerdeBincodeCompat for alloy_consensus::EthereumReceipt {
-        type BincodeRepr<'a> = alloy_consensus::serde_bincode_compat::EthereumReceipt<'a>;
+    impl<T: TxTy + Serialize + DeserializeOwned> super::SerdeBincodeCompat
+        for alloy_consensus::EthereumReceipt<T>
+    {
+        type BincodeRepr<'a> = alloy_consensus::serde_bincode_compat::EthereumReceipt<'a, T>;
 
         fn as_repr(&self) -> Self::BincodeRepr<'_> {
             self.into()

crates/primitives-traits/src/storage.rs

@@ -1,4 +1,4 @@
-use alloy_primitives::{keccak256, B256, U256};
+use alloy_primitives::{B256, U256};
 
 /// Trait for `DupSort` table values that contain a subkey.
 ///
@@ -12,117 +12,6 @@ pub trait ValueWithSubKey {
     fn get_subkey(&self) -> Self::SubKey;
 }
 
-/// A storage slot key that tracks whether it holds a plain (unhashed) EVM slot
-/// or a keccak256-hashed slot.
-///
-/// This enum replaces the `use_hashed_state: bool` parameter pattern by carrying
-/// provenance with the key itself. Once tagged at a read/write boundary, downstream
-/// code can call [`Self::to_hashed`] without risk of double-hashing — hashing a
-/// [`StorageSlotKey::Hashed`] is a no-op.
-///
-/// The on-disk encoding is unchanged (raw 32-byte [`B256`]). The variant is set
-/// by the code that knows the context (which table, which storage mode).
-#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub enum StorageSlotKey {
-    /// An unhashed EVM storage slot, as produced by REVM execution.
-    Plain(B256),
-    /// A keccak256-hashed storage slot, as stored in `HashedStorages` and
-    /// in v2-mode `StorageChangeSets`.
-    Hashed(B256),
-}
-
-impl Default for StorageSlotKey {
-    fn default() -> Self {
-        Self::Plain(B256::ZERO)
-    }
-}
-
-impl StorageSlotKey {
-    /// Create a plain slot key from a REVM [`U256`] storage index.
-    pub const fn from_u256(slot: U256) -> Self {
-        Self::Plain(B256::new(slot.to_be_bytes()))
-    }
-
-    /// Create a plain slot key from a raw [`B256`].
-    pub const fn plain(key: B256) -> Self {
-        Self::Plain(key)
-    }
-
-    /// Create a hashed slot key from a raw [`B256`].
-    pub const fn hashed(key: B256) -> Self {
-        Self::Hashed(key)
-    }
-
-    /// Tag a raw [`B256`] based on the storage mode.
-    ///
-    /// When `use_hashed_state` is true the key is assumed already hashed.
-    /// When false it is assumed to be a plain slot.
-    pub const fn from_raw(key: B256, use_hashed_state: bool) -> Self {
-        if use_hashed_state {
-            Self::Hashed(key)
-        } else {
-            Self::Plain(key)
-        }
-    }
-
-    /// Returns the raw [`B256`] regardless of variant.
-    pub const fn as_b256(&self) -> B256 {
-        match *self {
-            Self::Plain(b) | Self::Hashed(b) => b,
-        }
-    }
-
-    /// Returns `true` if this key is already hashed.
-    pub const fn is_hashed(&self) -> bool {
-        matches!(self, Self::Hashed(_))
-    }
-
-    /// Returns `true` if this key is plain (unhashed).
-    pub const fn is_plain(&self) -> bool {
-        matches!(self, Self::Plain(_))
-    }
-
-    /// Produce the keccak256-hashed form of this slot key.
-    ///
-    /// - If already [`Hashed`](Self::Hashed), returns the inner value as-is (no double-hash).
-    /// - If [`Plain`](Self::Plain), applies keccak256 and returns the result.
-    pub fn to_hashed(&self) -> B256 {
-        match *self {
-            Self::Hashed(b) => b,
-            Self::Plain(b) => keccak256(b),
-        }
-    }
-
-    /// Convert a plain slot to its changeset representation.
-    ///
-    /// In v2 mode (`use_hashed_state = true`), the changeset stores hashed keys,
-    /// so the plain key is hashed. In v1 mode, the plain key is stored as-is.
-    ///
-    /// Panics (debug) if called on an already-hashed key.
-    pub fn to_changeset_key(self, use_hashed_state: bool) -> B256 {
-        debug_assert!(self.is_plain(), "to_changeset_key called on already-hashed key");
-        if use_hashed_state {
-            self.to_hashed()
-        } else {
-            self.as_b256()
-        }
-    }
-
-    /// Like [`to_changeset_key`](Self::to_changeset_key) but returns a tagged
-    /// [`StorageSlotKey`] instead of a raw [`B256`].
-    ///
-    /// Panics (debug) if called on an already-hashed key.
-    pub fn to_changeset(self, use_hashed_state: bool) -> Self {
-        Self::from_raw(self.to_changeset_key(use_hashed_state), use_hashed_state)
-    }
-}
-
-impl From<StorageSlotKey> for B256 {
-    fn from(key: StorageSlotKey) -> Self {
-        key.as_b256()
-    }
-}
-
 /// Account storage entry.
 ///
 /// `key` is the subkey when used as a value in the `StorageChangeSets` table.
@@ -142,14 +31,6 @@ impl StorageEntry {
     pub const fn new(key: B256, value: U256) -> Self {
         Self { key, value }
     }
-
-    /// Tag this entry's key as a [`StorageSlotKey`] based on the storage mode.
-    ///
-    /// When `use_hashed_state` is true, the key is tagged as already-hashed.
-    /// When false, it is tagged as plain.
-    pub const fn slot_key(&self, use_hashed_state: bool) -> StorageSlotKey {
-        StorageSlotKey::from_raw(self.key, use_hashed_state)
-    }
 }
 
 impl ValueWithSubKey for StorageEntry {