feat(payload): prototype cached payload builder using engine-tree caches

Adds EthereumPayloadBuilder2 that demonstrates using the engine-tree's three caches (execution cache, precompile cache, sparse trie) inside the payload builder for faster block building. Changes: - New builder2.rs with EthereumPayloadBuilder2 and cached_ethereum_payload() - Make payload_processor types public (SharedPreservedSparseTrie, PreservedSparseTrie, PreservedTrieGuard, SparseTrie, PayloadExecutionCache) - SparseTrieStateProvider wrapper delegating all StateProvider traits - Sparse trie state root is a stub (falls through to standard computation)
2026-04-30 03:01:58 -04:00 · 2026-03-19 06:16:58 +00:00
37 changed files with 1757 additions and 670 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -290,17 +290,21 @@ dependencies = [
 [[package]]
 name = "alloy-evm"
 version = "0.29.2"
-source = "git+https://github.com/alloy-rs/evm?rev=b0eb7e6#b0eb7e617f964f7090c504f21a5977cc440117f7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3cb6ba2dafd6327f78f2b59ae539bd5c39c57a01dc76763e92942619d934a7bb"
 dependencies = [
 "alloy-consensus",
 "alloy-eips",
 "alloy-hardforks 0.4.7",
+ "alloy-op-hardforks",
 "alloy-primitives",
 "alloy-rpc-types-engine",
 "alloy-rpc-types-eth",
 "alloy-sol-types",
 "auto_impl",
 "derive_more",
+ "op-alloy",
+ "op-revm",
 "revm",
 "thiserror 2.0.18",
 "tracing",
@@ -436,6 +440,18 @@ dependencies = [
 "url",
 ]

+[[package]]
+name = "alloy-op-hardforks"
+version = "0.4.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6472c610150c4c4c15be9e1b964c9b78068f933bda25fb9cdf09b9ac2bb66f36"
+dependencies = [
+ "alloy-chains",
+ "alloy-hardforks 0.4.7",
+ "alloy-primitives",
+ "auto_impl",
+]
+
 [[package]]
 name = "alloy-primitives"
 version = "1.5.7"
@@ -2954,7 +2970,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ab67060fc6b8ef687992d439ca0fa36e7ed17e9a0b16b25b601e8757df720de"
 dependencies = [
 "data-encoding",
- "syn 2.0.117",
+ "syn 1.0.109",
 ]

 [[package]]
@@ -3456,7 +3472,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
 "libc",
- "windows-sys 0.59.0",
+ "windows-sys 0.52.0",
 ]

 [[package]]
@@ -4271,7 +4287,7 @@ dependencies = [
 "log",
 "rustversion",
 "windows-link 0.1.3",
- "windows-result 0.3.4",
+ "windows-result 0.4.1",
 ]

 [[package]]
@@ -4809,7 +4825,7 @@ dependencies = [
 "libc",
 "percent-encoding",
 "pin-project-lite",
- "socket2 0.6.3",
+ "socket2 0.5.10",
 "tokio",
 "tower-service",
 "tracing",
@@ -4827,7 +4843,7 @@ dependencies = [
 "js-sys",
 "log",
 "wasm-bindgen",
- "windows-core 0.61.2",
+ "windows-core 0.62.2",
 ]

 [[package]]
@@ -5179,7 +5195,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
 dependencies = [
 "hermit-abi",
 "libc",
- "windows-sys 0.59.0",
+ "windows-sys 0.52.0",
 ]

 [[package]]
@@ -6339,6 +6355,19 @@ version = "11.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"

+[[package]]
+name = "op-alloy"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a95dd0974d5e60ffe9342a70cc0033d299244fab01cb16a958eb7352ddba1fa7"
+dependencies = [
+ "op-alloy-consensus",
+ "op-alloy-network",
+ "op-alloy-provider",
+ "op-alloy-rpc-types",
+ "op-alloy-rpc-types-engine",
+]
+
 [[package]]
 name = "op-alloy-consensus"
 version = "0.24.0"
@@ -6359,6 +6388,37 @@ dependencies = [
 "thiserror 2.0.18",
 ]

+[[package]]
+name = "op-alloy-network"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c8ea44162d493219cc678aaca1253d46c3aa73aa361326dfa9d406f086dfa135"
+dependencies = [
+ "alloy-consensus",
+ "alloy-network",
+ "alloy-primitives",
+ "alloy-provider",
+ "alloy-rpc-types-eth",
+ "alloy-signer",
+ "op-alloy-consensus",
+ "op-alloy-rpc-types",
+]
+
+[[package]]
+name = "op-alloy-provider"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "83aa8dc34bdf077c8e6d48ff75beff4ac14b428d982c9722483ccd7473c0e114"
+dependencies = [
+ "alloy-network",
+ "alloy-primitives",
+ "alloy-provider",
+ "alloy-rpc-types-engine",
+ "alloy-transport",
+ "async-trait",
+ "op-alloy-rpc-types-engine",
+]
+
 [[package]]
 name = "op-alloy-rpc-types"
 version = "0.24.0"
@@ -6400,6 +6460,17 @@ dependencies = [
 "thiserror 2.0.18",
 ]

+[[package]]
+name = "op-revm"
+version = "17.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57a98f3a512a7e02a1dcf1242b57302d83657b265a665d50ad98d0b158efaf2c"
+dependencies = [
+ "auto_impl",
+ "revm",
+ "serde",
+]
+
 [[package]]
 name = "opaque-debug"
 version = "0.3.1"
@@ -7065,7 +7136,7 @@ dependencies = [
 "quinn-udp",
 "rustc-hash",
 "rustls",
- "socket2 0.6.3",
+ "socket2 0.5.10",
 "thiserror 2.0.18",
 "tokio",
 "tracing",
@@ -7102,7 +7173,7 @@ dependencies = [
 "cfg_aliases",
 "libc",
 "once_cell",
- "socket2 0.6.3",
+ "socket2 0.5.10",
 "tracing",
 "windows-sys 0.60.2",
 ]
@@ -8656,6 +8727,7 @@ dependencies = [
 "reth-basic-payload-builder",
 "reth-chainspec",
 "reth-consensus-common",
+ "reth-engine-tree",
 "reth-errors",
 "reth-ethereum-primitives",
 "reth-evm",
@@ -8668,6 +8740,7 @@ dependencies = [
 "reth-revm",
 "reth-storage-api",
 "reth-transaction-pool",
+ "reth-trie",
 "revm",
 "tracing",
 ]
@@ -10592,7 +10665,8 @@ dependencies = [
 [[package]]
 name = "revm"
 version = "36.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b0abc15d09cd211e9e73410ada10134069c794d4bcdb787dfc16a1bf0939849c"
 dependencies = [
 "revm-bytecode",
 "revm-context",
@@ -10610,7 +10684,8 @@ dependencies = [
 [[package]]
 name = "revm-bytecode"
 version = "9.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e86e468df3cf5cf59fa7ef71a3e9ccabb76bb336401ea2c0674f563104cf3c5e"
 dependencies = [
 "bitvec",
 "phf",
@@ -10621,7 +10696,8 @@ dependencies = [
 [[package]]
 name = "revm-context"
 version = "15.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9eb1f0a76b14d684a444fc52f7bf6b7564bf882599d91ee62e76d602e7a187c7"
 dependencies = [
 "bitvec",
 "cfg-if",
@@ -10637,7 +10713,8 @@ dependencies = [
 [[package]]
 name = "revm-context-interface"
 version = "16.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc256b27743e2912ca16899568e6652a372eb5d1d573e6edb16c7836b16cf487"
 dependencies = [
 "alloy-eip2930",
 "alloy-eip7702",
@@ -10652,7 +10729,8 @@ dependencies = [
 [[package]]
 name = "revm-database"
 version = "12.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c0a7d6da41061f2c50f99a2632571026b23684b5449ff319914151f4449b6c8"
 dependencies = [
 "alloy-eips",
 "revm-bytecode",
@@ -10665,7 +10743,8 @@ dependencies = [
 [[package]]
 name = "revm-database-interface"
 version = "10.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bd497a38a79057b94a049552cb1f925ad15078bc1a479c132aeeebd1d2ccc768"
 dependencies = [
 "auto_impl",
 "either",
@@ -10678,7 +10757,8 @@ dependencies = [
 [[package]]
 name = "revm-handler"
 version = "17.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f1eed729ca9b228ae98688f352235871e9b8be3d568d488e4070f64c56e9d3d"
 dependencies = [
 "auto_impl",
 "derive-where",
@@ -10696,7 +10776,8 @@ dependencies = [
 [[package]]
 name = "revm-inspector"
 version = "17.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cbf5102391706513689f91cb3cb3d97b5f13a02e8647e6e9cb7620877ef84847"
 dependencies = [
 "auto_impl",
 "either",
@@ -10712,8 +10793,9 @@ dependencies = [

 [[package]]
 name = "revm-inspectors"
-version = "0.36.1"
-source = "git+https://github.com/paradigmxyz/revm-inspectors?rev=24becc3#24becc35973c6c1d4e1c1475fa51a83d36d50d48"
+version = "0.36.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cfb0f462c8a3d9989d3dbc62d7cca4dfecd7072cfa5d563ab90ced60590ed1da"
 dependencies = [
 "alloy-primitives",
 "alloy-rpc-types-eth",
@@ -10732,7 +10814,8 @@ dependencies = [
 [[package]]
 name = "revm-interpreter"
 version = "34.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf22f80612bb8f58fd1f578750281f2afadb6c93835b14ae6a4d6b75ca26f445"
 dependencies = [
 "revm-bytecode",
 "revm-context-interface",
@@ -10744,7 +10827,8 @@ dependencies = [
 [[package]]
 name = "revm-precompile"
 version = "32.1.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2ec11f45deec71e4945e1809736bb20d454285f9167ab53c5159dae1deb603f"
 dependencies = [
 "ark-bls12-381",
 "ark-bn254",
@@ -10758,7 +10842,6 @@ dependencies = [
 "cfg-if",
 "k256",
 "p256",
- "revm-context-interface",
 "revm-primitives",
 "ripemd",
 "secp256k1 0.31.1",
@@ -10768,7 +10851,8 @@ dependencies = [
 [[package]]
 name = "revm-primitives"
 version = "22.1.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4bcfb5ce6cf18b118932bcdb7da05cd9c250f2cb9f64131396b55f3fe3537c35"
 dependencies = [
 "alloy-primitives",
 "num_enum",
@@ -10779,7 +10863,8 @@ dependencies = [
 [[package]]
 name = "revm-state"
 version = "10.0.0"
-source = "git+https://github.com/bluealloy/revm?rev=712dac7a#712dac7a3461f5f80682627ee8fc9032c337c51b"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d29404707763da607e5d6e4771cb203998c28159279c2f64cc32de08d2814651"
 dependencies = [
 "alloy-eip7928",
 "bitflags 2.11.0",
@@ -11022,7 +11107,7 @@ dependencies = [
 "errno",
 "libc",
 "linux-raw-sys",
- "windows-sys 0.59.0",
+ "windows-sys 0.52.0",
 ]

 [[package]]
@@ -11613,8 +11698,7 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 [[package]]
 name = "slotmap"
 version = "1.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bdd58c3c93c3d278ca835519292445cb4b0d4dc59ccfdf7ceadaab3f8aeb4038"
+source = "git+https://github.com/DaniPopes/slotmap.git?branch=dani%2Fshrink-methods#09fbd360968f9ecef19fc9559037cf869d33858b"
 dependencies = [
 "version_check",
 ]
@@ -11880,7 +11964,7 @@ dependencies = [
 "getrandom 0.4.2",
 "once_cell",
 "rustix",
- "windows-sys 0.59.0",
+ "windows-sys 0.52.0",
 ]

 [[package]]
@@ -12605,7 +12689,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c5f7c95348f20c1c913d72157b3c6dee6ea3e30b3d19502c5a7f6d3f160dacbf"
 dependencies = [
 "cc",
- "windows-targets 0.48.5",
+ "windows-targets 0.52.6",
 ]

 [[package]]
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -399,9 +399,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, features = [
-    "__internal",
-] }
+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" }
@@ -446,22 +444,18 @@ revm-state = { version = "10.0.0", default-features = false }
 revm-primitives = { version = "22.1.0", default-features = false }
 revm-interpreter = { version = "34.0.0", default-features = false }
 revm-database-interface = { version = "10.0.0", default-features = false }
-revm-inspectors = "0.36.1"
+revm-inspectors = "0.36.0"

 # eth
 alloy-dyn-abi = "1.5.6"
-alloy-primitives = { version = "1.5.6", default-features = false, features = [
-    "map-foldhash",
-] }
+alloy-primitives = { version = "1.5.6", default-features = false, features = ["map-foldhash"] }
 alloy-sol-types = { version = "1.5.6", default-features = false }

 alloy-chains = { version = "0.2.5", default-features = false }
 alloy-eip2124 = { version = "0.2.0", default-features = false }
 alloy-eip7928 = { version = "0.3.0", default-features = false }
 alloy-evm = { version = "0.29.2", default-features = false }
-alloy-rlp = { version = "0.3.13", default-features = false, features = [
-    "core-net",
-] }
+alloy-rlp = { version = "0.3.13", default-features = false, features = ["core-net"] }
 alloy-trie = { version = "0.9.4", default-features = false }

 alloy-hardforks = "0.4.5"
@@ -473,15 +467,10 @@ alloy-genesis = { version = "1.7.3", default-features = false }
 alloy-json-rpc = { version = "1.7.3", default-features = false }
 alloy-network = { version = "1.7.3", default-features = false }
 alloy-network-primitives = { version = "1.7.3", default-features = false }
-alloy-provider = { version = "1.7.3", features = [
-    "reqwest",
-    "debug-api",
-], default-features = false }
+alloy-provider = { version = "1.7.3", features = ["reqwest", "debug-api"], default-features = false }
 alloy-pubsub = { version = "1.7.3", default-features = false }
 alloy-rpc-client = { version = "1.7.3", default-features = false }
-alloy-rpc-types = { version = "1.7.3", features = [
-    "eth",
-], default-features = false }
+alloy-rpc-types = { version = "1.7.3", features = ["eth"], default-features = false }
 alloy-rpc-types-admin = { version = "1.7.3", default-features = false }
 alloy-rpc-types-anvil = { version = "1.7.3", default-features = false }
 alloy-rpc-types-beacon = { version = "1.7.3", default-features = false }
@@ -495,9 +484,7 @@ alloy-serde = { version = "1.7.3", default-features = false }
 alloy-signer = { version = "1.7.3", default-features = false }
 alloy-signer-local = { version = "1.7.3", default-features = false }
 alloy-transport = { version = "1.7.3" }
-alloy-transport-http = { version = "1.7.3", features = [
-    "reqwest-rustls-tls",
-], default-features = false }
+alloy-transport-http = { version = "1.7.3", features = ["reqwest-rustls-tls"], default-features = false }
 alloy-transport-ipc = { version = "1.7.3", default-features = false }
 alloy-transport-ws = { version = "1.7.3", default-features = false }

@@ -511,10 +498,7 @@ either = { version = "1.15.0", default-features = false }
 arrayvec = { version = "0.7.6", default-features = false }
 aquamarine = "0.6"
 auto_impl = "1"
-backon = { version = "1.2", default-features = false, features = [
-    "std-blocking-sleep",
-    "tokio-sleep",
-] }
+backon = { version = "1.2", default-features = false, features = ["std-blocking-sleep", "tokio-sleep"] }
 bincode = "1.3"
 bitflags = "2.4"
 boyer-moore-magiclen = "0.2.16"
@@ -538,13 +522,9 @@ linked_hash_set = "0.1"
 libc = "0.2"
 lz4 = "1.28.1"
 modular-bitfield = "0.13.1"
-notify = { version = "8.0.0", default-features = false, features = [
-    "macos_fsevent",
-] }
+notify = { version = "8.0.0", default-features = false, features = ["macos_fsevent"] }
 nybbles = { version = "0.4.8", default-features = false }
-once_cell = { version = "1.19", default-features = false, features = [
-    "critical-section",
-] }
+once_cell = { version = "1.19", default-features = false, features = ["critical-section"] }
 parking_lot = "0.12"
 quanta = "0.12"
 paste = "1.0"
@@ -558,16 +538,15 @@ 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 }
 shlex = "1.3"
-slotmap = "1"
+# https://github.com/orlp/slotmap/pull/148
+slotmap = { git = "https://github.com/DaniPopes/slotmap.git", branch = "dani/shrink-methods" }
 smallvec = "1"
 strum = { version = "0.27", default-features = false }
 strum_macros = "0.27"
 syn = "2.0"
 thiserror = { version = "2.0.0", default-features = false }
 tar = "0.4.44"
-tracing = { version = "0.1.0", default-features = false, features = [
-    "attributes",
-] }
+tracing = { version = "0.1.0", default-features = false, features = ["attributes"] }
 tracing-appender = "0.2"
 url = { version = "2.3", default-features = false }
 zstd = "0.13"
@@ -605,11 +584,7 @@ futures-util = { version = "0.3", default-features = false }
 hyper = "1.3"
 hyper-util = "0.1.5"
 pin-project = "1.0.12"
-reqwest = { version = "0.12", default-features = false, features = [
-    "rustls-tls",
-    "rustls-tls-native-roots",
-    "stream",
-] }
+reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "rustls-tls-native-roots", "stream"] }
 tracing-futures = "0.2"
 tower = "0.5"
 tower-http = "0.6"
@@ -634,10 +609,7 @@ proptest-arbitrary-interop = "0.1.0"
 # crypto
 enr = { version = "0.13", default-features = false }
 k256 = { version = "0.13", default-features = false, features = ["ecdsa"] }
-secp256k1 = { version = "0.30", default-features = false, features = [
-    "global-context",
-    "recovery",
-] }
+secp256k1 = { version = "0.30", default-features = false, features = ["global-context", "recovery"] }
 # rand 8 for secp256k1
 rand_08 = { package = "rand", version = "0.8" }

@@ -780,15 +752,6 @@ ipnet = "2.11"
 # jsonrpsee-http-client = { git = "https://github.com/paradigmxyz/jsonrpsee", branch = "matt/make-rpc-service-pub" }
 # jsonrpsee-types = { git = "https://github.com/paradigmxyz/jsonrpsee", branch = "matt/make-rpc-service-pub" }

-alloy-evm = { git = "https://github.com/alloy-rs/evm", rev = "b0eb7e6" }
+# alloy-evm = { git = "https://github.com/alloy-rs/evm", rev = "9bc2dba" }

-revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", rev = "24becc3" }
-
-# revm from rakita/state-gas branch
-revm = { git = "https://github.com/bluealloy/revm", rev = "712dac7a" }
-revm-bytecode = { git = "https://github.com/bluealloy/revm", rev = "712dac7a" }
-revm-database = { git = "https://github.com/bluealloy/revm", rev = "712dac7a" }
-revm-state = { git = "https://github.com/bluealloy/revm", rev = "712dac7a" }
-revm-primitives = { git = "https://github.com/bluealloy/revm", rev = "712dac7a" }
-revm-interpreter = { git = "https://github.com/bluealloy/revm", rev = "712dac7a" }
-revm-database-interface = { git = "https://github.com/bluealloy/revm", rev = "712dac7a" }
+# revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", rev = "3020ea8" }
--- a/crates/engine/local/Cargo.toml
+++ b/crates/engine/local/Cargo.toml
@@ -38,8 +38,8 @@ op-alloy-rpc-types-engine = { workspace = true, optional = true }
 workspace = true

 [features]
-# op = [
-#     "dep:op-alloy-rpc-types-engine",
-#     "reth-payload-primitives/op",
-#     "reth-primitives-traits/op",
-# ]
+op = [
+    "dep:op-alloy-rpc-types-engine",
+    "reth-payload-primitives/op",
+    "reth-primitives-traits/op",
+]
--- a/crates/engine/tree/src/tree/payload_processor/mod.rs
+++ b/crates/engine/tree/src/tree/payload_processor/mod.rs
@@ -55,12 +55,14 @@ use tracing::{debug, debug_span, instrument, warn, Span};

 pub mod bal;
 pub mod multiproof;
-mod preserved_sparse_trie;
+pub mod preserved_sparse_trie;
 pub mod prewarm;
 pub mod receipt_root_task;
 pub mod sparse_trie;

-use preserved_sparse_trie::{PreservedSparseTrie, SharedPreservedSparseTrie};
+pub use preserved_sparse_trie::{
+    PreservedSparseTrie, PreservedTrieGuard, SharedPreservedSparseTrie, SparseTrie,
+};

 /// Default parallelism thresholds to use with the [`ParallelSparseTrie`].
 ///
@@ -1024,7 +1026,7 @@ impl PayloadExecutionCache {
    /// - It exists and matches the requested parent hash
    /// - No other tasks are currently using it (checked via Arc reference count)
    #[instrument(level = "debug", target = "engine::tree::payload_processor", skip(self))]
-    pub(crate) fn get_cache_for(&self, parent_hash: B256) -> Option<SavedCache> {
+    pub fn get_cache_for(&self, parent_hash: B256) -> Option<SavedCache> {
        let start = Instant::now();
        let mut cache = self.inner.write();

--- a/crates/engine/tree/src/tree/payload_processor/preserved_sparse_trie.rs
+++ b/crates/engine/tree/src/tree/payload_processor/preserved_sparse_trie.rs
@@ -7,25 +7,25 @@ use std::{sync::Arc, time::Instant};
 use tracing::debug;

 /// Type alias for the sparse trie type used in preservation.
-pub(super) type SparseTrie = SparseStateTrie<ConfigurableSparseTrie, ConfigurableSparseTrie>;
+pub type SparseTrie = SparseStateTrie<ConfigurableSparseTrie, ConfigurableSparseTrie>;

 /// Shared handle to a preserved sparse trie that can be reused across payload validations.
 ///
 /// This is stored in [`PayloadProcessor`](super::PayloadProcessor) and cloned to pass to
 /// [`SparseTrieCacheTask`](super::sparse_trie::SparseTrieCacheTask) for trie reuse.
 #[derive(Debug, Default, Clone)]
-pub(super) struct SharedPreservedSparseTrie(Arc<Mutex<Option<PreservedSparseTrie>>>);
+pub struct SharedPreservedSparseTrie(Arc<Mutex<Option<PreservedSparseTrie>>>);

 impl SharedPreservedSparseTrie {
    /// Takes the preserved trie if present, leaving `None` in its place.
-    pub(super) fn take(&self) -> Option<PreservedSparseTrie> {
+    pub fn take(&self) -> Option<PreservedSparseTrie> {
        self.0.lock().take()
    }

    /// Acquires a guard that blocks `take()` until dropped.
    /// Use this before sending the state root result to ensure the next block
    /// waits for the trie to be stored.
-    pub(super) fn lock(&self) -> PreservedTrieGuard<'_> {
+    pub fn lock(&self) -> PreservedTrieGuard<'_> {
        PreservedTrieGuard(self.0.lock())
    }

@@ -36,7 +36,7 @@ impl SharedPreservedSparseTrie {
    /// before starting payload processing.
    ///
    /// Returns the time spent waiting for the lock.
-    pub(super) fn wait_for_availability(&self) -> std::time::Duration {
+    pub fn wait_for_availability(&self) -> std::time::Duration {
        let start = Instant::now();
        let _guard = self.0.lock();
        let elapsed = start.elapsed();
@@ -53,11 +53,11 @@ impl SharedPreservedSparseTrie {

 /// Guard that holds the lock on the preserved trie.
 /// While held, `take()` will block. Call `store()` to save the trie before dropping.
-pub(super) struct PreservedTrieGuard<'a>(parking_lot::MutexGuard<'a, Option<PreservedSparseTrie>>);
+pub struct PreservedTrieGuard<'a>(parking_lot::MutexGuard<'a, Option<PreservedSparseTrie>>);

 impl PreservedTrieGuard<'_> {
    /// Stores a preserved trie for later reuse.
-    pub(super) fn store(&mut self, trie: PreservedSparseTrie) {
+    pub fn store(&mut self, trie: PreservedSparseTrie) {
        self.0.replace(trie);
    }
 }
@@ -69,7 +69,7 @@ impl PreservedTrieGuard<'_> {
 ///   matches the anchor.
 /// - **Cleared**: Trie data has been cleared but allocations are preserved for reuse.
 #[derive(Debug)]
-pub(super) enum PreservedSparseTrie {
+pub enum PreservedSparseTrie {
    /// Trie with a computed state root that can be reused for continuation payloads.
    Anchored {
        /// The sparse state trie (pruned after root computation).
@@ -90,12 +90,12 @@ impl PreservedSparseTrie {
    ///
    /// The `state_root` is the computed state root from the trie, which becomes the
    /// anchor for determining if subsequent payloads can reuse this trie.
-    pub(super) const fn anchored(trie: SparseTrie, state_root: B256) -> Self {
+    pub const fn anchored(trie: SparseTrie, state_root: B256) -> Self {
        Self::Anchored { trie, state_root }
    }

    /// Creates a cleared preserved trie (allocations preserved, data cleared).
-    pub(super) const fn cleared(trie: SparseTrie) -> Self {
+    pub const fn cleared(trie: SparseTrie) -> Self {
        Self::Cleared { trie }
    }

@@ -104,7 +104,7 @@ impl PreservedSparseTrie {
    /// If the preserved trie is anchored and the parent state root matches, the pruned
    /// trie structure is reused directly. Otherwise, the trie is cleared but allocations
    /// are preserved to reduce memory overhead.
-    pub(super) fn into_trie_for(self, parent_state_root: B256) -> SparseTrie {
+    pub fn into_trie_for(self, parent_state_root: B256) -> SparseTrie {
        match self {
            Self::Anchored { trie, state_root } if state_root == parent_state_root => {
                debug!(
--- a/crates/engine/tree/src/tree/precompile_cache.rs
+++ b/crates/engine/tree/src/tree/precompile_cache.rs
@@ -81,8 +81,8 @@ pub struct CacheEntry<S> {
 }

 impl<S> CacheEntry<S> {
-    const fn regular_gas_used(&self) -> u64 {
-        self.output.gas.limit() - self.output.gas.remaining()
+    const fn gas_used(&self) -> u64 {
+        self.output.gas_used
    }

    fn to_precompile_result(&self) -> PrecompileResult {
@@ -170,10 +170,10 @@ where

    fn call(&self, input: PrecompileInput<'_>) -> PrecompileResult {
        if let Some(entry) = &self.cache.get(input.data, self.spec_id.clone()) &&
-            input.gas >= entry.regular_gas_used()
+            input.gas >= entry.gas_used()
        {
            self.increment_by_one_precompile_cache_hits();
-            return entry.to_precompile_result();
+            return entry.to_precompile_result()
        }

        let calldata = input.data;
@@ -228,14 +228,15 @@ mod tests {
    use super::*;
    use reth_evm::{EthEvmFactory, Evm, EvmEnv, EvmFactory};
    use reth_revm::db::EmptyDB;
-    use revm::{context::TxEnv, interpreter::gas::GasTracker, precompile::PrecompileOutput};
+    use revm::{context::TxEnv, precompile::PrecompileOutput};
    use revm_primitives::hardfork::SpecId;

    #[test]
    fn test_precompile_cache_basic() {
        let dyn_precompile: DynPrecompile = (|_input: PrecompileInput<'_>| -> PrecompileResult {
            Ok(PrecompileOutput {
-                gas: GasTracker::new(0, 0, 0),
+                gas_used: 0,
+                gas_refunded: 0,
                bytes: Bytes::default(),
                reverted: false,
            })
@@ -246,7 +247,8 @@ mod tests {
            CachedPrecompile::new(dyn_precompile, PrecompileCache::default(), SpecId::PRAGUE, None);

        let output = PrecompileOutput {
-            gas: GasTracker::new(50, 0, 0),
+            gas_used: 50,
+            gas_refunded: 0,
            bytes: alloy_primitives::Bytes::copy_from_slice(b"cached_result"),
            reverted: false,
        };
@@ -277,7 +279,8 @@ mod tests {
                assert_eq!(input.data, input_data);

                Ok(PrecompileOutput {
-                    gas: GasTracker::new(5000, 0, 0),
+                    gas_used: 5000,
+                    gas_refunded: 0,
                    bytes: alloy_primitives::Bytes::copy_from_slice(b"output_from_precompile_1"),
                    reverted: false,
                })
@@ -291,7 +294,8 @@ mod tests {
                assert_eq!(input.data, input_data);

                Ok(PrecompileOutput {
-                    gas: GasTracker::new(7000, 0, 0),
+                    gas_used: 7000,
+                    gas_refunded: 0,
                    bytes: alloy_primitives::Bytes::copy_from_slice(b"output_from_precompile_2"),
                    reverted: false,
                })
--- a/crates/ethereum/payload/Cargo.toml
+++ b/crates/ethereum/payload/Cargo.toml
@@ -27,7 +27,9 @@ reth-evm.workspace = true
 reth-evm-ethereum = { workspace = true, features = ["std"] }
 reth-errors.workspace = true
 reth-chainspec.workspace = true
+reth-engine-tree.workspace = true
 reth-payload-validator.workspace = true
+reth-trie.workspace = true

 # ethereum
 alloy-rlp.workspace = true
--- a/crates/ethereum/payload/src/builder2.rs
+++ b/crates/ethereum/payload/src/builder2.rs
@@ -0,0 +1,593 @@
+//! Prototype payload builder that uses engine-tree caches (execution cache, precompile cache,
+//! sparse trie) to speed up block building.
+//!
+//! This is NOT wired into the node builder -- it exists to prototype the API surface needed
+//! to share engine-tree caches with the payload builder.
+
+use alloy_consensus::Transaction;
+use alloy_primitives::U256;
+use alloy_rlp::Encodable;
+use reth_basic_payload_builder::{
+    is_better_payload, BuildArguments, BuildOutcome, MissingPayloadBehaviour, PayloadBuilder,
+    PayloadConfig,
+};
+use reth_chainspec::{ChainSpecProvider, EthChainSpec, EthereumHardforks};
+use reth_consensus_common::validation::MAX_RLP_BLOCK_SIZE;
+use reth_engine_tree::tree::{
+    precompile_cache::{CachedPrecompile, PrecompileCacheMap},
+    CachedStateMetrics, CachedStateProvider, ExecutionCache, PayloadExecutionCache,
+    PreservedSparseTrie, SharedPreservedSparseTrie,
+};
+use reth_errors::{BlockExecutionError, BlockValidationError, ConsensusError};
+use reth_ethereum_primitives::{EthPrimitives, TransactionSigned};
+use reth_evm::{
+    execute::{BlockBuilder, BlockBuilderOutcome},
+    ConfigureEvm, Evm, NextBlockEnvAttributes, SpecFor,
+};
+use reth_evm_ethereum::EthEvmConfig;
+use reth_payload_builder::{BlobSidecars, EthBuiltPayload, EthPayloadBuilderAttributes};
+use reth_payload_builder_primitives::PayloadBuilderError;
+use reth_payload_primitives::PayloadBuilderAttributes;
+use reth_primitives_traits::transaction::error::InvalidTransactionError;
+use reth_revm::{database::StateProviderDatabase, db::State};
+use reth_storage_api::{StateProvider, StateProviderFactory};
+use reth_trie::{updates::TrieUpdates, HashedPostState, HashedStorage, MultiProof, TrieInput};
+use reth_transaction_pool::{
+    error::{Eip4844PoolTransactionError, InvalidPoolTransactionError},
+    BestTransactions, BestTransactionsAttributes, PoolTransaction, TransactionPool,
+    ValidPoolTransaction,
+};
+use revm::context_interface::Block as _;
+use std::sync::Arc;
+use tracing::{debug, trace, warn};
+
+use crate::EthereumBuilderConfig;
+
+/// Default cross-block cache size (256 MB) used when no engine cache is available.
+const DEFAULT_CACHE_SIZE: usize = 256 * 1024 * 1024;
+
+type BestTransactionsIter<Pool> = Box<
+    dyn BestTransactions<Item = Arc<ValidPoolTransaction<<Pool as TransactionPool>::Transaction>>>,
+>;
+
+/// Prototype Ethereum payload builder that leverages engine-tree caches.
+///
+/// Holds references to the three engine caches:
+/// - **Execution cache**: warm account/storage/bytecode data from prior block execution
+/// - **Precompile cache**: cached precompile results across blocks
+/// - **Sparse trie**: preserved sparse trie for faster state root computation
+#[derive(Debug, Clone)]
+pub struct EthereumPayloadBuilder2<Pool, Client, EvmConfig = EthEvmConfig>
+where
+    EvmConfig: ConfigureEvm,
+{
+    /// Client providing access to node state.
+    client: Client,
+    /// Transaction pool.
+    pool: Pool,
+    /// The type responsible for creating the evm.
+    evm_config: EvmConfig,
+    /// Payload builder configuration.
+    builder_config: EthereumBuilderConfig,
+    /// Engine execution cache (Arc-backed, cheap to clone).
+    execution_cache: PayloadExecutionCache,
+    /// Engine precompile cache map (Arc-backed, cheap to clone).
+    precompile_cache_map: PrecompileCacheMap<SpecFor<EvmConfig>>,
+    /// Engine sparse trie (Arc-backed, cheap to clone).
+    sparse_trie: SharedPreservedSparseTrie,
+}
+
+impl<Pool, Client, EvmConfig> EthereumPayloadBuilder2<Pool, Client, EvmConfig>
+where
+    EvmConfig: ConfigureEvm,
+{
+    /// Creates a new `EthereumPayloadBuilder2`.
+    pub fn new(
+        client: Client,
+        pool: Pool,
+        evm_config: EvmConfig,
+        builder_config: EthereumBuilderConfig,
+        execution_cache: PayloadExecutionCache,
+        precompile_cache_map: PrecompileCacheMap<SpecFor<EvmConfig>>,
+        sparse_trie: SharedPreservedSparseTrie,
+    ) -> Self {
+        Self {
+            client,
+            pool,
+            evm_config,
+            builder_config,
+            execution_cache,
+            precompile_cache_map,
+            sparse_trie,
+        }
+    }
+}
+
+impl<Pool, Client, EvmConfig> PayloadBuilder for EthereumPayloadBuilder2<Pool, Client, EvmConfig>
+where
+    EvmConfig: ConfigureEvm<Primitives = EthPrimitives, NextBlockEnvCtx = NextBlockEnvAttributes>,
+    Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthereumHardforks> + Clone,
+    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TransactionSigned>>,
+{
+    type Attributes = EthPayloadBuilderAttributes;
+    type BuiltPayload = EthBuiltPayload;
+
+    fn try_build(
+        &self,
+        args: BuildArguments<EthPayloadBuilderAttributes, EthBuiltPayload>,
+    ) -> Result<BuildOutcome<EthBuiltPayload>, PayloadBuilderError> {
+        cached_ethereum_payload(
+            self.evm_config.clone(),
+            self.client.clone(),
+            self.pool.clone(),
+            self.builder_config.clone(),
+            args,
+            |attributes| self.pool.best_transactions_with_attributes(attributes),
+            &self.execution_cache,
+            &self.precompile_cache_map,
+            &self.sparse_trie,
+        )
+    }
+
+    fn on_missing_payload(
+        &self,
+        _args: BuildArguments<Self::Attributes, Self::BuiltPayload>,
+    ) -> MissingPayloadBehaviour<Self::BuiltPayload> {
+        if self.builder_config.await_payload_on_missing {
+            MissingPayloadBehaviour::AwaitInProgress
+        } else {
+            MissingPayloadBehaviour::RaceEmptyPayload
+        }
+    }
+
+    fn build_empty_payload(
+        &self,
+        config: PayloadConfig<Self::Attributes>,
+    ) -> Result<EthBuiltPayload, PayloadBuilderError> {
+        let args = BuildArguments::new(Default::default(), config, Default::default(), None);
+
+        cached_ethereum_payload(
+            self.evm_config.clone(),
+            self.client.clone(),
+            self.pool.clone(),
+            self.builder_config.clone(),
+            args,
+            |attributes| self.pool.best_transactions_with_attributes(attributes),
+            &self.execution_cache,
+            &self.precompile_cache_map,
+            &self.sparse_trie,
+        )?
+        .into_payload()
+        .ok_or_else(|| PayloadBuilderError::MissingPayload)
+    }
+}
+
+/// Constructs an Ethereum transaction payload using engine-tree caches.
+///
+/// This is identical to [`default_ethereum_payload`](crate::default_ethereum_payload) except:
+///
+/// **Phase 1** - Execution cache + precompile cache:
+/// - Uses `CachedStateProvider` wrapping the state provider with the engine's execution cache
+/// - Wraps EVM precompiles with `CachedPrecompile` using the engine's precompile cache
+///
+/// **Phase 2** - Sparse trie state root:
+/// - Takes the preserved sparse trie before building
+/// - Uses it to compute the state root instead of the slow `state_root_with_updates()`
+#[allow(clippy::too_many_arguments)]
+pub fn cached_ethereum_payload<EvmConfig, Client, Pool, F>(
+    evm_config: EvmConfig,
+    client: Client,
+    pool: Pool,
+    builder_config: EthereumBuilderConfig,
+    args: BuildArguments<EthPayloadBuilderAttributes, EthBuiltPayload>,
+    best_txs: F,
+    execution_cache: &PayloadExecutionCache,
+    precompile_cache_map: &PrecompileCacheMap<SpecFor<EvmConfig>>,
+    sparse_trie: &SharedPreservedSparseTrie,
+) -> Result<BuildOutcome<EthBuiltPayload>, PayloadBuilderError>
+where
+    EvmConfig: ConfigureEvm<Primitives = EthPrimitives, NextBlockEnvCtx = NextBlockEnvAttributes>,
+    Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthereumHardforks>,
+    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TransactionSigned>>,
+    F: FnOnce(BestTransactionsAttributes) -> BestTransactionsIter<Pool>,
+{
+    let BuildArguments { mut cached_reads, config, cancel, best_payload } = args;
+    let PayloadConfig { parent_header, attributes } = config;
+
+    let state_provider = client.state_by_block_hash(parent_header.hash())?;
+
+    // --- Phase 1: Execution cache ---
+    // Try to get a warm cache from the engine's execution cache for the parent block.
+    // If unavailable, create a fresh (empty) cache — `CachedStateProvider` will still
+    // function correctly, just with no pre-warmed data.
+    let (caches, metrics) = if let Some(saved) = execution_cache.get_cache_for(parent_header.hash())
+    {
+        debug!(target: "payload_builder", "using engine execution cache for parent");
+        (saved.cache().clone(), saved.metrics().clone())
+    } else {
+        debug!(target: "payload_builder", "no engine execution cache available, using fresh cache");
+        (ExecutionCache::new(DEFAULT_CACHE_SIZE), CachedStateMetrics::zeroed())
+    };
+
+    let cached_state = CachedStateProvider::new(state_provider.as_ref(), caches, metrics);
+    let state = StateProviderDatabase::new(&cached_state);
+    let mut db =
+        State::builder().with_database(cached_reads.as_db_mut(state)).with_bundle_update().build();
+
+    let next_block_attrs = NextBlockEnvAttributes {
+        timestamp: attributes.timestamp(),
+        suggested_fee_recipient: attributes.suggested_fee_recipient(),
+        prev_randao: attributes.prev_randao(),
+        gas_limit: builder_config.gas_limit(parent_header.gas_limit),
+        parent_beacon_block_root: attributes.parent_beacon_block_root(),
+        withdrawals: Some(attributes.withdrawals().clone()),
+        extra_data: builder_config.extra_data,
+    };
+
+    // --- Phase 1: Precompile cache ---
+    // Get spec_id before creating the builder, to properly key cached precompile results.
+    let spec_id = *evm_config
+        .next_evm_env(&parent_header, &next_block_attrs)
+        .map_err(PayloadBuilderError::other)?
+        .spec_id();
+
+    let mut builder = evm_config
+        .builder_for_next_block(&mut db, &parent_header, next_block_attrs)
+        .map_err(PayloadBuilderError::other)?;
+
+    builder.evm_mut().precompiles_mut().map_cacheable_precompiles(|address, precompile| {
+        CachedPrecompile::wrap(
+            precompile,
+            precompile_cache_map.cache_for_address(*address),
+            spec_id,
+            None,
+        )
+    });
+
+    let chain_spec = client.chain_spec();
+
+    debug!(target: "payload_builder", id=%attributes.id, parent_header = ?parent_header.hash(), parent_number = parent_header.number, "building new payload (cached)");
+    let mut cumulative_gas_used = 0;
+    let block_gas_limit: u64 = builder.evm_mut().block().gas_limit();
+    let base_fee = builder.evm_mut().block().basefee();
+
+    let mut best_txs = best_txs(BestTransactionsAttributes::new(
+        base_fee,
+        builder.evm_mut().block().blob_gasprice().map(|gasprice| gasprice as u64),
+    ));
+    let mut total_fees = U256::ZERO;
+
+    builder.apply_pre_execution_changes().map_err(|err| {
+        warn!(target: "payload_builder", %err, "failed to apply pre-execution changes");
+        PayloadBuilderError::Internal(err.into())
+    })?;
+
+    let mut blob_sidecars = BlobSidecars::Empty;
+    let mut block_blob_count = 0;
+    let mut block_transactions_rlp_length = 0;
+
+    let blob_params = chain_spec.blob_params_at_timestamp(attributes.timestamp);
+    let protocol_max_blob_count =
+        blob_params.as_ref().map(|params| params.max_blob_count).unwrap_or_else(Default::default);
+
+    let max_blob_count = builder_config
+        .max_blobs_per_block
+        .map(|user_limit| std::cmp::min(user_limit, protocol_max_blob_count).max(1))
+        .unwrap_or(protocol_max_blob_count);
+
+    let is_osaka = chain_spec.is_osaka_active_at_timestamp(attributes.timestamp);
+    let withdrawals_rlp_length = attributes.withdrawals().length();
+
+    // --- Transaction execution loop (identical to default_ethereum_payload) ---
+    while let Some(pool_tx) = best_txs.next() {
+        if cumulative_gas_used + pool_tx.gas_limit() > block_gas_limit {
+            best_txs.mark_invalid(
+                &pool_tx,
+                &InvalidPoolTransactionError::ExceedsGasLimit(pool_tx.gas_limit(), block_gas_limit),
+            );
+            continue
+        }
+
+        if cancel.is_cancelled() {
+            return Ok(BuildOutcome::Cancelled)
+        }
+
+        let tx = pool_tx.to_consensus();
+        let tx_rlp_len = tx.inner().length();
+
+        let estimated_block_size_with_tx =
+            block_transactions_rlp_length + tx_rlp_len + withdrawals_rlp_length + 1024;
+
+        if is_osaka && estimated_block_size_with_tx > MAX_RLP_BLOCK_SIZE {
+            best_txs.mark_invalid(
+                &pool_tx,
+                &InvalidPoolTransactionError::OversizedData {
+                    size: estimated_block_size_with_tx,
+                    limit: MAX_RLP_BLOCK_SIZE,
+                },
+            );
+            continue
+        }
+
+        let mut blob_tx_sidecar = None;
+        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 {
+                trace!(target: "payload_builder", tx=?tx.hash(), ?block_blob_count, "skipping blob transaction because it would exceed the max blob count per block");
+                best_txs.mark_invalid(
+                    &pool_tx,
+                    &InvalidPoolTransactionError::Eip4844(
+                        Eip4844PoolTransactionError::TooManyEip4844Blobs {
+                            have: block_blob_count + tx_blob_count,
+                            permitted: max_blob_count,
+                        },
+                    ),
+                );
+                continue
+            }
+
+            let blob_sidecar_result = 'sidecar: {
+                let Some(sidecar) =
+                    pool.get_blob(*tx.hash()).map_err(PayloadBuilderError::other)?
+                else {
+                    break 'sidecar Err(Eip4844PoolTransactionError::MissingEip4844BlobSidecar)
+                };
+
+                if is_osaka {
+                    if sidecar.is_eip7594() {
+                        Ok(sidecar)
+                    } else {
+                        Err(Eip4844PoolTransactionError::UnexpectedEip4844SidecarAfterOsaka)
+                    }
+                } else if sidecar.is_eip4844() {
+                    Ok(sidecar)
+                } else {
+                    Err(Eip4844PoolTransactionError::UnexpectedEip7594SidecarBeforeOsaka)
+                }
+            };
+
+            blob_tx_sidecar = match blob_sidecar_result {
+                Ok(sidecar) => Some(sidecar),
+                Err(error) => {
+                    best_txs.mark_invalid(&pool_tx, &InvalidPoolTransactionError::Eip4844(error));
+                    continue
+                }
+            };
+        }
+
+        let gas_used = match builder.execute_transaction(tx.clone()) {
+            Ok(gas_used) => gas_used,
+            Err(BlockExecutionError::Validation(BlockValidationError::InvalidTx {
+                error, ..
+            })) => {
+                if error.is_nonce_too_low() {
+                    trace!(target: "payload_builder", %error, ?tx, "skipping nonce too low transaction");
+                } else {
+                    trace!(target: "payload_builder", %error, ?tx, "skipping invalid transaction and its descendants");
+                    best_txs.mark_invalid(
+                        &pool_tx,
+                        &InvalidPoolTransactionError::Consensus(
+                            InvalidTransactionError::TxTypeNotSupported,
+                        ),
+                    );
+                }
+                continue
+            }
+            Err(err) => return Err(PayloadBuilderError::evm(err)),
+        };
+
+        if let Some(blob_hashes) = tx.blob_versioned_hashes() {
+            block_blob_count += blob_hashes.len() as u64;
+            if block_blob_count == max_blob_count {
+                best_txs.skip_blobs();
+            }
+        }
+
+        block_transactions_rlp_length += tx_rlp_len;
+
+        let miner_fee =
+            tx.effective_tip_per_gas(base_fee).expect("fee is always valid; execution succeeded");
+        total_fees += U256::from(miner_fee) * U256::from(gas_used);
+        cumulative_gas_used += gas_used;
+
+        if let Some(sidecar) = blob_tx_sidecar {
+            blob_sidecars.push_sidecar_variant(sidecar.as_ref().clone());
+        }
+    }
+
+    // check if we have a better block
+    if !is_better_payload(best_payload.as_ref(), total_fees) {
+        drop(builder);
+        return Ok(BuildOutcome::Aborted { fees: total_fees, cached_reads })
+    }
+
+    // --- Phase 2: Sparse trie state root ---
+    // Take the preserved sparse trie before finishing. The wrapper's
+    // `state_root_with_updates` will use it instead of the slow full trie computation.
+    let preserved = sparse_trie.take();
+    let wrapper = SparseTrieStateProvider { inner: state_provider.as_ref(), preserved };
+
+    let BlockBuilderOutcome { execution_result, block, .. } = builder.finish(wrapper)?;
+
+    let requests = chain_spec
+        .is_prague_active_at_timestamp(attributes.timestamp)
+        .then_some(execution_result.requests);
+
+    let sealed_block = Arc::new(block.into_sealed_block());
+    debug!(target: "payload_builder", id=%attributes.id, sealed_block_header = ?sealed_block.sealed_header(), "sealed built block (cached)");
+
+    if is_osaka && sealed_block.rlp_length() > MAX_RLP_BLOCK_SIZE {
+        return Err(PayloadBuilderError::other(ConsensusError::BlockTooLarge {
+            rlp_length: sealed_block.rlp_length(),
+            max_rlp_length: MAX_RLP_BLOCK_SIZE,
+        }));
+    }
+
+    let payload = EthBuiltPayload::new(attributes.id, sealed_block, total_fees, requests)
+        .with_sidecars(blob_sidecars);
+
+    Ok(BuildOutcome::Better { payload, cached_reads })
+}
+
+/// A state provider wrapper that holds a preserved sparse trie for
+/// faster state root computation.
+///
+/// All `StateProvider` trait methods delegate to the inner provider. The
+/// `state_root_with_updates` method is the hook point for sparse trie integration.
+struct SparseTrieStateProvider<'a> {
+    inner: &'a dyn StateProvider,
+    /// The preserved sparse trie, taken from the shared handle before building.
+    #[allow(dead_code)]
+    preserved: Option<PreservedSparseTrie>,
+}
+
+// --- Delegate all StateProvider trait methods to inner ---
+
+impl reth_storage_api::AccountReader for SparseTrieStateProvider<'_> {
+    fn basic_account(
+        &self,
+        address: &alloy_primitives::Address,
+    ) -> reth_errors::ProviderResult<Option<reth_primitives_traits::Account>> {
+        self.inner.basic_account(address)
+    }
+}
+
+impl reth_storage_api::BlockHashReader for SparseTrieStateProvider<'_> {
+    fn block_hash(
+        &self,
+        number: alloy_primitives::BlockNumber,
+    ) -> reth_errors::ProviderResult<Option<alloy_primitives::B256>> {
+        self.inner.block_hash(number)
+    }
+
+    fn canonical_hashes_range(
+        &self,
+        start: alloy_primitives::BlockNumber,
+        end: alloy_primitives::BlockNumber,
+    ) -> reth_errors::ProviderResult<Vec<alloy_primitives::B256>> {
+        self.inner.canonical_hashes_range(start, end)
+    }
+}
+
+impl reth_storage_api::BytecodeReader for SparseTrieStateProvider<'_> {
+    fn bytecode_by_hash(
+        &self,
+        code_hash: &alloy_primitives::B256,
+    ) -> reth_errors::ProviderResult<Option<reth_primitives_traits::Bytecode>> {
+        self.inner.bytecode_by_hash(code_hash)
+    }
+}
+
+impl reth_storage_api::StateRootProvider for SparseTrieStateProvider<'_> {
+    fn state_root(
+        &self,
+        hashed_state: HashedPostState,
+    ) -> reth_errors::ProviderResult<alloy_primitives::B256> {
+        self.inner.state_root(hashed_state)
+    }
+
+    fn state_root_from_nodes(
+        &self,
+        input: TrieInput,
+    ) -> reth_errors::ProviderResult<alloy_primitives::B256> {
+        self.inner.state_root_from_nodes(input)
+    }
+
+    fn state_root_with_updates(
+        &self,
+        hashed_state: HashedPostState,
+    ) -> reth_errors::ProviderResult<(alloy_primitives::B256, TrieUpdates)> {
+        // Phase 2: Hook point for sparse trie state root computation.
+        //
+        // Currently falls through to the standard computation. The sparse trie integration
+        // requires the multiproof pipeline which is complex to wire synchronously.
+        //
+        // Future implementation:
+        //   1. let targets = hashed_state.multi_proof_targets();
+        //   2. let multiproof = self.inner.multiproof(TrieInput::default(), targets)?;
+        //   3. sparse_trie.reveal_multiproof(multiproof)?;
+        //   4. // update account/storage leaves from hashed_state
+        //   5. let (root, updates) = sparse_trie.root_with_updates(provider)?;
+        //   6. return Ok((root, updates));
+        self.inner.state_root_with_updates(hashed_state)
+    }
+
+    fn state_root_from_nodes_with_updates(
+        &self,
+        input: TrieInput,
+    ) -> reth_errors::ProviderResult<(alloy_primitives::B256, TrieUpdates)> {
+        self.inner.state_root_from_nodes_with_updates(input)
+    }
+}
+
+impl reth_storage_api::StorageRootProvider for SparseTrieStateProvider<'_> {
+    fn storage_root(
+        &self,
+        address: alloy_primitives::Address,
+        hashed_storage: HashedStorage,
+    ) -> reth_errors::ProviderResult<alloy_primitives::B256> {
+        self.inner.storage_root(address, hashed_storage)
+    }
+
+    fn storage_proof(
+        &self,
+        address: alloy_primitives::Address,
+        slot: alloy_primitives::B256,
+        hashed_storage: HashedStorage,
+    ) -> reth_errors::ProviderResult<reth_trie::StorageProof> {
+        self.inner.storage_proof(address, slot, hashed_storage)
+    }
+
+    fn storage_multiproof(
+        &self,
+        address: alloy_primitives::Address,
+        slots: &[alloy_primitives::B256],
+        hashed_storage: HashedStorage,
+    ) -> reth_errors::ProviderResult<reth_trie::StorageMultiProof> {
+        self.inner.storage_multiproof(address, slots, hashed_storage)
+    }
+}
+
+impl reth_storage_api::StateProofProvider for SparseTrieStateProvider<'_> {
+    fn proof(
+        &self,
+        input: TrieInput,
+        address: alloy_primitives::Address,
+        slots: &[alloy_primitives::B256],
+    ) -> reth_errors::ProviderResult<reth_trie::AccountProof> {
+        self.inner.proof(input, address, slots)
+    }
+
+    fn multiproof(
+        &self,
+        input: TrieInput,
+        targets: reth_trie::MultiProofTargets,
+    ) -> reth_errors::ProviderResult<MultiProof> {
+        self.inner.multiproof(input, targets)
+    }
+
+    fn witness(
+        &self,
+        input: TrieInput,
+        target: HashedPostState,
+    ) -> reth_errors::ProviderResult<Vec<alloy_primitives::Bytes>> {
+        self.inner.witness(input, target)
+    }
+}
+
+impl reth_storage_api::HashedPostStateProvider for SparseTrieStateProvider<'_> {
+    fn hashed_post_state(&self, bundle_state: &reth_revm::db::BundleState) -> HashedPostState {
+        self.inner.hashed_post_state(bundle_state)
+    }
+}
+
+impl StateProvider for SparseTrieStateProvider<'_> {
+    fn storage(
+        &self,
+        account: alloy_primitives::Address,
+        storage_key: alloy_primitives::StorageKey,
+    ) -> reth_errors::ProviderResult<Option<alloy_primitives::StorageValue>> {
+        self.inner.storage(account, storage_key)
+    }
+}
--- a/crates/ethereum/payload/src/lib.rs
+++ b/crates/ethereum/payload/src/lib.rs
@@ -40,6 +40,8 @@ use revm::context_interface::Block as _;
 use std::sync::Arc;
 use tracing::{debug, trace, warn};

+pub mod builder2;
+
 mod config;
 pub use config::*;

--- a/crates/evm/evm/Cargo.toml
+++ b/crates/evm/evm/Cargo.toml
@@ -62,4 +62,4 @@ test-utils = [
    "reth-trie-common/test-utils",
    "reth-ethereum-primitives/test-utils",
 ]
-# op = ["alloy-evm/op", "reth-primitives-traits/op"]
+op = ["alloy-evm/op", "reth-primitives-traits/op"]
--- a/crates/evm/evm/src/execute.rs
+++ b/crates/evm/evm/src/execute.rs
@@ -468,7 +468,7 @@ where
            self.executor.execute_transaction_with_commit_condition((tx_env, &tx), f)?
        {
            self.transactions.push(tx);
-            Ok(Some(gas_used.into()))
+            Ok(Some(gas_used))
        } else {
            Ok(None)
        }
--- a/crates/node/builder/Cargo.toml
+++ b/crates/node/builder/Cargo.toml
@@ -122,10 +122,10 @@ test-utils = [
    "reth-tasks/test-utils",
 ]
 trie-debug = ["reth-engine-tree/trie-debug"]
-# op = [
-#     "reth-db/op",
-#     "reth-db-api/op",
-#     "reth-engine-local/op",
-#     "reth-evm/op",
-#     "reth-primitives-traits/op",
-# ]
+op = [
+    "reth-db/op",
+    "reth-db-api/op",
+    "reth-engine-local/op",
+    "reth-evm/op",
+    "reth-primitives-traits/op",
+]
--- a/crates/payload/primitives/Cargo.toml
+++ b/crates/payload/primitives/Cargo.toml
@@ -53,7 +53,7 @@ std = [
    "either/std",
    "alloy-consensus/std",
 ]
-# op = [
-#     "dep:op-alloy-rpc-types-engine",
-#     "reth-primitives-traits/op",
-# ]
+op = [
+    "dep:op-alloy-rpc-types-engine",
+    "reth-primitives-traits/op",
+]
--- a/crates/rpc/rpc-convert/Cargo.toml
+++ b/crates/rpc/rpc-convert/Cargo.toml
@@ -43,10 +43,10 @@ serde_json.workspace = true

 [features]
 default = []
-#op = [
-#    "dep:op-alloy-consensus",
-#    "dep:op-alloy-rpc-types",
-#    "reth-evm/op",
-#    "reth-primitives-traits/op",
-#    "alloy-evm/op",
-#]
+op = [
+    "dep:op-alloy-consensus",
+    "dep:op-alloy-rpc-types",
+    "reth-evm/op",
+    "reth-primitives-traits/op",
+    "alloy-evm/op",
+]
--- a/crates/rpc/rpc-eth-api/Cargo.toml
+++ b/crates/rpc/rpc-eth-api/Cargo.toml
@@ -62,9 +62,9 @@ tracing.workspace = true
 [features]
 js-tracer = ["revm-inspectors/js-tracer", "reth-rpc-eth-types/js-tracer"]
 client = ["jsonrpsee/client", "jsonrpsee/async-client"]
-# op = [
-#     "reth-evm/op",
-#     "reth-primitives-traits/op",
-#     "reth-rpc-convert/op",
-#     "alloy-evm/op",
-# ]
+op = [
+    "reth-evm/op",
+    "reth-primitives-traits/op",
+    "reth-rpc-convert/op",
+    "alloy-evm/op",
+]
--- a/crates/rpc/rpc-eth-api/src/helpers/call.rs
+++ b/crates/rpc/rpc-eth-api/src/helpers/call.rs
@@ -517,7 +517,7 @@ pub trait EthCall: EstimateCall + Call + LoadPendingBlock + LoadBlock + FullEthA

            let result = this.inspect(&mut db, evm_env.clone(), tx_env.clone(), &mut inspector)?;
            let access_list = inspector.into_access_list();
-            let gas_used = result.result.tx_gas_used();
+            let gas_used = result.result.gas_used();
            tx_env.set_access_list(access_list.clone());
            if let Err(err) = Self::Error::ensure_success(result.result) {
                return Ok(AccessListResult {
@@ -529,7 +529,7 @@ pub trait EthCall: EstimateCall + Call + LoadPendingBlock + LoadBlock + FullEthA

            // transact again to get the exact gas used
            let result = this.transact(&mut db, evm_env, tx_env)?;
-            let gas_used = result.result.tx_gas_used();
+            let gas_used = result.result.gas_used();
            let error = Self::Error::ensure_success(result.result).err().map(|e| e.to_string());

            Ok(AccessListResult { access_list, gas_used: U256::from(gas_used), error })
--- a/crates/rpc/rpc-eth-api/src/helpers/estimate.rs
+++ b/crates/rpc/rpc-eth-api/src/helpers/estimate.rs
@@ -204,7 +204,7 @@ pub trait EstimateCall: Call {

        // NOTE: this is the gas the transaction used, which is less than the
        // transaction requires to succeed.
-        let mut gas_used = res.result.tx_gas_used();
+        let mut gas_used = res.result.gas_used();
        // the lowest value is capped by the gas used by the unconstrained transaction
        let mut lowest_gas_limit = gas_used.saturating_sub(1);

@@ -225,7 +225,7 @@ pub trait EstimateCall: Call {
            res = evm.transact(optimistic_tx_env).map_err(Self::Error::from_evm_err)?;

            // Update the gas used based on the new result.
-            gas_used = res.result.tx_gas_used();
+            gas_used = res.result.gas_used();
            // Update the gas limit estimates (highest and lowest) based on the execution result.
            update_estimated_gas_range(
                res.result,
--- a/crates/rpc/rpc-eth-types/src/error/api.rs
+++ b/crates/rpc/rpc-eth-types/src/error/api.rs
@@ -126,7 +126,7 @@ pub trait FromEvmError<Evm: ConfigureEvm>:
            ExecutionResult::Success { output, .. } => Ok(output.into_data()),
            ExecutionResult::Revert { output, .. } => Err(Self::from_revert(output)),
            ExecutionResult::Halt { reason, gas, .. } => {
-                Err(Self::from_evm_halt(reason, gas.tx_gas_used()))
+                Err(Self::from_evm_halt(reason, gas.used()))
            }
        }
    }
--- a/crates/rpc/rpc-eth-types/src/error/mod.rs
+++ b/crates/rpc/rpc-eth-types/src/error/mod.rs
@@ -553,7 +553,6 @@ where
            EVMError::Header(err) => err.into(),
            EVMError::Database(err) => err.into(),
            EVMError::Custom(err) => Self::EvmCustom(err),
-            EVMError::CustomAny(err) => Self::EvmCustom(err.to_string()),
        }
    }
 }
--- a/crates/rpc/rpc-eth-types/src/simulate.rs
+++ b/crates/rpc/rpc-eth-types/src/simulate.rs
@@ -314,7 +314,7 @@ where
                        code: SIMULATE_VM_ERROR_CODE,
                        ..SimulateError::invalid_params()
                    }),
-                    gas_used: gas.tx_gas_used(),
+                    gas_used: gas.used(),
                    logs: Vec::new(),
                    status: false,
                    ..Default::default()
@@ -330,7 +330,7 @@ where
                        code: SIMULATE_REVERT_CODE,
                        ..SimulateError::invalid_params()
                    }),
-                    gas_used: gas.tx_gas_used(),
+                    gas_used: gas.used(),
                    status: false,
                    logs: Vec::new(),
                    ..Default::default()
@@ -342,7 +342,7 @@ where
                SimCallResult {
                    return_data: output.into_data(),
                    error: None,
-                    gas_used: gas.tx_gas_used(),
+                    gas_used: gas.used(),
                    logs: logs
                        .into_iter()
                        .map(|log| {
--- a/crates/rpc/rpc/src/eth/bundle.rs
+++ b/crates/rpc/rpc/src/eth/bundle.rs
@@ -188,7 +188,7 @@ where
                    let gas_price = tx
                        .effective_tip_per_gas(basefee)
                        .expect("fee is always valid; execution succeeded");
-                    let gas_used = result.tx_gas_used();
+                    let gas_used = result.gas_used();
                    total_gas_used += gas_used;

                    let gas_fees = U256::from(gas_used) * U256::from(gas_price);
--- a/crates/rpc/rpc/src/eth/filter.rs
+++ b/crates/rpc/rpc/src/eth/filter.rs
@@ -478,24 +478,22 @@ where
                    return Err(ProviderError::HeaderNotFound(block_hash.into()).into())
                };

-                // Read number and timestamp from cached block or provider header
-                let (block_number, block_timestamp) = if let Some(block) = &maybe_block {
-                    (block.header().number(), block.header().timestamp())
+                // Get header - from cached block if available, otherwise from provider
+                let header = if let Some(block) = &maybe_block {
+                    block.header().clone()
                } else {
-                    let header = self
-                        .provider()
+                    self.provider()
                        .header_by_hash_or_number(block_hash.into())?
-                        .ok_or_else(|| ProviderError::HeaderNotFound(block_hash.into()))?;
-                    (header.number(), header.timestamp())
+                        .ok_or_else(|| ProviderError::HeaderNotFound(block_hash.into()))?
                };

                // Check if the block has been pruned (EIP-4444)
                let earliest_block = self.provider().earliest_block_number()?;
-                if block_number < earliest_block {
+                if header.number() < earliest_block {
                    return Err(EthApiError::PrunedHistoryUnavailable.into());
                }

-                let block_num_hash = BlockNumHash::new(block_number, block_hash);
+                let block_num_hash = BlockNumHash::new(header.number(), block_hash);

                let mut all_logs = Vec::new();
                append_matching_block_logs(
@@ -507,7 +505,7 @@ where
                    block_num_hash,
                    &receipts,
                    false,
-                    block_timestamp,
+                    header.timestamp(),
                )?;

                Ok(all_logs)
--- a/crates/rpc/rpc/src/eth/sim_bundle.rs
+++ b/crates/rpc/rpc/src/eth/sim_bundle.rs
@@ -289,7 +289,7 @@ where
                        .into());
                    }

-                    let gas_used = result.tx_gas_used();
+                    let gas_used = result.gas_used();
                    total_gas_used += gas_used;

                    // coinbase is always present in the result state
--- a/crates/storage/db-api/Cargo.toml
+++ b/crates/storage/db-api/Cargo.toml
@@ -88,8 +88,8 @@ arbitrary = [
    "op-alloy-consensus?/arbitrary",
    "reth-ethereum-primitives/arbitrary",
 ]
-# op = [
-#     "dep:op-alloy-consensus",
-#     "reth-codecs/op",
-#     "reth-primitives-traits/op",
-# ]
+op = [
+    "dep:op-alloy-consensus",
+    "reth-codecs/op",
+    "reth-primitives-traits/op",
+]
--- a/crates/storage/db/Cargo.toml
+++ b/crates/storage/db/Cargo.toml
@@ -93,8 +93,8 @@ arbitrary = [
    "reth-primitives-traits/arbitrary",
    "reth-prune-types/arbitrary",
 ]
-# op = [
-#     "reth-db-api/op",
-#     "reth-primitives-traits/op",
-# ]
+op = [
+    "reth-db-api/op",
+    "reth-primitives-traits/op",
+]
 disable-lock = []
--- a/crates/storage/provider/src/providers/database/builder.rs
+++ b/crates/storage/provider/src/providers/database/builder.rs
@@ -106,10 +106,8 @@ impl<N> ProviderFactoryBuilder<N> {
        let db = open_db_read_only(db_dir, db_args)?;
        let static_file_provider =
            StaticFileProvider::read_only(static_files_dir, watch_static_files)?;
-        let rocksdb_provider = RocksDBProvider::builder(&rocksdb_dir)
-            .with_default_tables()
-            .with_read_only(true)
-            .build()?;
+        let rocksdb_provider =
+            RocksDBProvider::builder(&rocksdb_dir).with_default_tables().build()?;
        ProviderFactory::new(db, chainspec, static_file_provider, rocksdb_provider, runtime)
            .map_err(Into::into)
    }
--- a/crates/storage/provider/src/providers/state/historical.rs
+++ b/crates/storage/provider/src/providers/state/historical.rs
@@ -20,7 +20,7 @@ use reth_storage_errors::provider::ProviderResult;
 use reth_trie::{
    hashed_cursor::HashedPostStateCursorFactory,
    proof::{Proof, StorageProof},
-    trie_cursor::InMemoryTrieCursorFactory,
+    trie_cursor::{masked::MaskedTrieCursorFactory, InMemoryTrieCursorFactory},
    updates::TrieUpdates,
    witness::TrieWitness,
    AccountProof, HashedPostState, HashedPostStateSorted, HashedStorage, KeccakKeyHasher,
@@ -525,16 +525,18 @@ impl<
            let nodes_sorted = input.nodes.into_sorted();
            let state_sorted = input.state.into_sorted();
            TrieWitness::new(
-                InMemoryTrieCursorFactory::new(
-                    reth_trie_db::DatabaseTrieCursorFactory::<_, A>::new(self.tx()),
-                    &nodes_sorted,
+                MaskedTrieCursorFactory::new(
+                    InMemoryTrieCursorFactory::new(
+                        reth_trie_db::DatabaseTrieCursorFactory::<_, A>::new(self.tx()),
+                        &nodes_sorted,
+                    ),
+                    input.prefix_sets.freeze(),
                ),
                HashedPostStateCursorFactory::new(
                    reth_trie_db::DatabaseHashedCursorFactory::new(self.tx()),
                    &state_sorted,
                ),
            )
-            .with_prefix_sets_mut(input.prefix_sets)
            .always_include_root_node()
            .compute(target)
            .map_err(ProviderError::from)
--- a/crates/storage/provider/src/providers/state/latest.rs
+++ b/crates/storage/provider/src/providers/state/latest.rs
@@ -11,7 +11,7 @@ use reth_storage_errors::provider::{ProviderError, ProviderResult};
 use reth_trie::{
    hashed_cursor::HashedPostStateCursorFactory,
    proof::{Proof, StorageProof},
-    trie_cursor::InMemoryTrieCursorFactory,
+    trie_cursor::{masked::MaskedTrieCursorFactory, InMemoryTrieCursorFactory},
    updates::TrieUpdates,
    witness::TrieWitness,
    AccountProof, HashedPostState, HashedStorage, KeccakKeyHasher, MultiProof, MultiProofTargets,
@@ -223,16 +223,18 @@ impl<Provider: DBProvider + StorageSettingsCache> StateProofProvider
            let nodes_sorted = input.nodes.into_sorted();
            let state_sorted = input.state.into_sorted();
            Ok(TrieWitness::new(
-                InMemoryTrieCursorFactory::new(
-                    reth_trie_db::DatabaseTrieCursorFactory::<_, A>::new(self.tx()),
-                    &nodes_sorted,
+                MaskedTrieCursorFactory::new(
+                    InMemoryTrieCursorFactory::new(
+                        reth_trie_db::DatabaseTrieCursorFactory::<_, A>::new(self.tx()),
+                        &nodes_sorted,
+                    ),
+                    input.prefix_sets.freeze(),
                ),
                HashedPostStateCursorFactory::new(
                    reth_trie_db::DatabaseHashedCursorFactory::new(self.tx()),
                    &state_sorted,
                ),
            )
-            .with_prefix_sets_mut(input.prefix_sets)
            .always_include_root_node()
            .compute(target)?
            .into_values()
--- a/crates/transaction-pool/src/validate/eth.rs
+++ b/crates/transaction-pool/src/validate/eth.rs
@@ -1429,7 +1429,7 @@ pub fn ensure_intrinsic_gas<T: EthPoolTransaction>(
    );

    let gas_limit = transaction.gas_limit();
-    if gas_limit < gas.initial_total_gas || gas_limit < gas.floor_gas {
+    if gas_limit < gas.initial_gas || gas_limit < gas.floor_gas {
        Err(InvalidPoolTransactionError::IntrinsicGasTooLow)
    } else {
        Ok(())
--- a/crates/trie/sparse/src/arena/mod.rs
+++ b/crates/trie/sparse/src/arena/mod.rs
@@ -9,7 +9,7 @@ use nodes::{
 };

 use crate::{LeafLookup, LeafLookupError, LeafUpdate, SparseTrie, SparseTrieUpdates};
-use alloc::{borrow::Cow, boxed::Box, collections::VecDeque, vec::Vec};
+use alloc::{borrow::Cow, boxed::Box, vec::Vec};
 use alloy_primitives::{
    keccak256,
    map::{B256Map, HashMap, HashSet},
@@ -60,54 +60,6 @@ fn prefix_range(
    begin..end
 }

-/// Compacts an arena by BFS-copying all reachable nodes into a fresh `SlotMap`, dropping
-/// unreachable (pruned) slots. Parents are stored before children for cache-friendly top-down
-/// traversal.
-fn compact_arena(arena: &mut NodeArena, root: &mut Index) {
-    let mut new_arena = SlotMap::with_capacity(arena.len());
-    let mut queue = VecDeque::new();
-
-    let root_node = arena.remove(*root).expect("root exists");
-    let new_root = new_arena.insert(root_node);
-    queue.push_back(new_root);
-
-    while let Some(new_idx) = queue.pop_front() {
-        // Invariant: any node popped from `queue` has been moved into `new_arena` but
-        // its Branch.children have not been rewritten yet — every Revealed(idx) here is
-        // still an old-arena index, and the child is still present in `arena` because
-        // only this parent's iteration can remove it (each child has exactly one parent).
-        let old_children: SmallVec<[(usize, Index); 16]> = match &new_arena[new_idx] {
-            ArenaSparseNode::Branch(b) => b
-                .children
-                .iter()
-                .enumerate()
-                .filter_map(|(i, c)| match c {
-                    ArenaSparseNodeBranchChild::Revealed(old_idx) => Some((i, *old_idx)),
-                    _ => None,
-                })
-                .collect(),
-            _ => continue,
-        };
-
-        for (child_pos, old_child_idx) in old_children {
-            let child_node = arena.remove(old_child_idx).expect("child exists");
-            let new_child_idx = new_arena.insert(child_node);
-            let ArenaSparseNode::Branch(b) = &mut new_arena[new_idx] else { unreachable!() };
-            b.children[child_pos] = ArenaSparseNodeBranchChild::Revealed(new_child_idx);
-            queue.push_back(new_child_idx);
-        }
-    }
-
-    debug_assert!(
-        arena.is_empty(),
-        "compact_arena: {} orphaned nodes remaining after BFS drain",
-        arena.len(),
-    );
-
-    *arena = new_arena;
-    *root = new_root;
-}
-
 /// Reusable buffers shared by both [`ArenaSparseSubtrie`] and [`ArenaParallelSparseTrie`].
 #[derive(Debug, Default, Clone)]
 struct ArenaTrieBuffers {
@@ -174,7 +126,7 @@ impl ArenaSparseSubtrie {
    }

    fn clear(&mut self) {
-        self.arena = SlotMap::new();
+        self.arena.clear();
        self.buffers.clear();
        self.required_proofs.clear();
        self.num_leaves = 0;
@@ -255,11 +207,6 @@ impl ArenaSparseSubtrie {
        }

        self.num_leaves -= pruned_leaves;
-
-        if pruned > 0 {
-            compact_arena(&mut self.arena, &mut self.root);
-        }
-
        #[cfg(debug_assertions)]
        self.debug_assert_counters();
        pruned
@@ -2594,7 +2541,7 @@ impl SparseTrie for ArenaParallelSparseTrie {
                self.cleared_subtries.push(subtrie);
            }
        }
-        self.upper_arena = SlotMap::new();
+        self.upper_arena.clear();
        self.root = self.upper_arena.insert(ArenaSparseNode::EmptyRoot);
        if let Some(updates) = self.updates.as_mut() {
            updates.clear()
@@ -2603,15 +2550,15 @@ impl SparseTrie for ArenaParallelSparseTrie {
    }

    fn shrink_nodes_to(&mut self, size: usize) {
-        // We do not shrink the upper trie for now.
-        //
-        // As soon as a trie rotates from a live subtrie into the
-        // cleared_subtrie it will be properly shrunk.
-        for s in &mut self.cleared_subtries {
-            if s.arena.capacity() > size {
-                s.arena = SlotMap::with_capacity(size);
+        self.upper_arena.shrink_to(size);
+        for (_, node) in &mut self.upper_arena {
+            if let ArenaSparseNode::Subtrie(s) = node {
+                s.arena.shrink_to(size);
            }
        }
+        for s in &mut self.cleared_subtries {
+            s.arena.shrink_to(size);
+        }
    }

    fn shrink_values_to(&mut self, _size: usize) {
@@ -2788,10 +2735,6 @@ impl SparseTrie for ArenaParallelSparseTrie {
            }
        }

-        if pruned > 0 {
-            compact_arena(&mut self.upper_arena, &mut self.root);
-        }
-
        #[cfg(feature = "trie-debug")]
        self.record_initial_state();

--- a/crates/trie/trie/src/proof/mod.rs
+++ b/crates/trie/trie/src/proof/mod.rs
@@ -149,24 +149,15 @@ where
    ) -> Result<DecodedMultiProofV2, StateProofError> {
        let MultiProofTargetsV2 { mut account_targets, storage_targets } = targets;

-        let storage_prefix_sets: B256Map<_> = self
-            .prefix_sets
-            .storage_prefix_sets
-            .into_iter()
-            .map(|(addr, ps)| (addr, ps.freeze()))
-            .collect();
-
        // Compute account proofs using the V2 proof calculator with sync account encoding.
        let account_trie_cursor = self.trie_cursor_factory.account_trie_cursor()?;
        let hashed_account_cursor = self.hashed_cursor_factory.hashed_account_cursor()?;
        let mut account_value_encoder = SyncAccountValueEncoder::new(
            self.trie_cursor_factory.clone(),
            self.hashed_cursor_factory.clone(),
-        )
-        .with_storage_prefix_sets(storage_prefix_sets.clone());
+        );
        let mut account_calculator =
-            proof_v2::ProofCalculator::new(account_trie_cursor, hashed_account_cursor)
-                .with_prefix_set(self.prefix_sets.account_prefix_set.freeze());
+            proof_v2::ProofCalculator::new(account_trie_cursor, hashed_account_cursor);
        let account_proofs =
            account_calculator.proof(&mut account_value_encoder, &mut account_targets)?;

@@ -182,9 +173,6 @@ where
                storage_trie_cursor,
                hashed_storage_cursor,
            );
-            if let Some(prefix_set) = storage_prefix_sets.get(&hashed_address) {
-                storage_calculator = storage_calculator.with_prefix_set(prefix_set.clone());
-            }
            let proofs = storage_calculator.storage_proof(hashed_address, &mut targets)?;
            storage_proofs.insert(hashed_address, proofs);
        }
--- a/crates/trie/trie/src/trie_cursor/masked.rs
+++ b/crates/trie/trie/src/trie_cursor/masked.rs
@@ -0,0 +1,752 @@
+use super::{TrieCursor, TrieCursorFactory, TrieStorageCursor};
+use alloy_primitives::{map::B256Map, B256};
+use reth_storage_errors::db::DatabaseError;
+use reth_trie_common::{
+    prefix_set::{PrefixSet, TriePrefixSets},
+    BranchNodeCompact, Nibbles,
+};
+use std::sync::Arc;
+
+/// A [`TrieCursorFactory`] wrapper that creates cursors which invalidate cached trie hash data
+/// for children whose paths match the prefix sets in a [`TriePrefixSets`].
+///
+/// The `destroyed_accounts` field of the prefix sets is not used by the cursor — it is only
+/// relevant during trie update finalization, not during cursor traversal.
+#[derive(Debug, Clone)]
+pub struct MaskedTrieCursorFactory<CF> {
+    /// Underlying trie cursor factory.
+    cursor_factory: CF,
+    /// Frozen prefix sets used for masking.
+    prefix_sets: TriePrefixSets,
+}
+
+impl<CF> MaskedTrieCursorFactory<CF> {
+    /// Create a new factory from an inner cursor factory and frozen prefix sets.
+    pub const fn new(cursor_factory: CF, prefix_sets: TriePrefixSets) -> Self {
+        Self { cursor_factory, prefix_sets }
+    }
+}
+
+impl<CF: TrieCursorFactory> TrieCursorFactory for MaskedTrieCursorFactory<CF> {
+    type AccountTrieCursor<'a>
+        = MaskedTrieCursor<CF::AccountTrieCursor<'a>>
+    where
+        Self: 'a;
+
+    type StorageTrieCursor<'a>
+        = MaskedTrieCursor<CF::StorageTrieCursor<'a>>
+    where
+        Self: 'a;
+
+    fn account_trie_cursor(&self) -> Result<Self::AccountTrieCursor<'_>, DatabaseError> {
+        let cursor = self.cursor_factory.account_trie_cursor()?;
+        Ok(MaskedTrieCursor::new(cursor, self.prefix_sets.account_prefix_set.clone()))
+    }
+
+    fn storage_trie_cursor(
+        &self,
+        hashed_address: B256,
+    ) -> Result<Self::StorageTrieCursor<'_>, DatabaseError> {
+        let cursor = self.cursor_factory.storage_trie_cursor(hashed_address)?;
+        let prefix_set =
+            self.prefix_sets.storage_prefix_sets.get(&hashed_address).cloned().unwrap_or_default();
+        Ok(MaskedTrieCursor::new_storage(
+            cursor,
+            prefix_set,
+            self.prefix_sets.storage_prefix_sets.clone(),
+        ))
+    }
+}
+
+/// A [`TrieCursor`] wrapper that invalidates cached trie hash data for children whose paths match
+/// a [`PrefixSet`].
+///
+/// For each node returned by the inner cursor, hash bits are unset for children whose paths match
+/// the prefix set, and the corresponding hashes are removed from the node. If a node's `hash_mask`
+/// and `tree_mask` are both empty after masking, the node is skipped entirely.
+#[derive(Debug)]
+pub struct MaskedTrieCursor<C> {
+    /// The inner cursor.
+    cursor: C,
+    /// Prefix set used to determine which children's hashes to invalidate.
+    prefix_set: PrefixSet,
+    /// Storage prefix sets for swapping on `set_hashed_address`.
+    storage_prefix_sets: Option<B256Map<PrefixSet>>,
+}
+
+impl<C> MaskedTrieCursor<C> {
+    /// Create a new cursor wrapping `cursor`, masking hash bits for children whose paths match
+    /// `prefix_set`.
+    pub const fn new(cursor: C, prefix_set: PrefixSet) -> Self {
+        Self { cursor, prefix_set, storage_prefix_sets: None }
+    }
+
+    /// Create a new storage cursor that can swap its prefix set on `set_hashed_address`.
+    pub const fn new_storage(
+        cursor: C,
+        prefix_set: PrefixSet,
+        storage_prefix_sets: B256Map<PrefixSet>,
+    ) -> Self {
+        Self { cursor, prefix_set, storage_prefix_sets: Some(storage_prefix_sets) }
+    }
+}
+
+impl<C: TrieCursor> MaskedTrieCursor<C> {
+    /// Mask hash bits on a node for children whose paths match the prefix set.
+    ///
+    /// Returns `true` if the node should be kept, `false` if it should be skipped (both
+    /// `hash_mask` and `tree_mask` are empty after masking).
+    fn mask_node(&mut self, key: &Nibbles, node: &mut BranchNodeCompact) -> bool {
+        if !self.prefix_set.contains(key) {
+            return true;
+        }
+
+        // The subtree is modified — root hash is always invalid.
+        node.root_hash = None;
+
+        let original_hash_mask = node.hash_mask;
+        if original_hash_mask.is_empty() {
+            return true;
+        }
+
+        let mut new_hash_mask = original_hash_mask;
+        let mut child_path = *key;
+        let key_len = key.len();
+
+        for nibble in original_hash_mask.iter() {
+            child_path.truncate(key_len);
+            child_path.push(nibble);
+
+            if self.prefix_set.contains(&child_path) {
+                new_hash_mask.unset_bit(nibble);
+            }
+        }
+
+        if new_hash_mask != original_hash_mask {
+            // Remove hashes for unset bits in-place.
+            let hashes = Arc::make_mut(&mut node.hashes);
+            let mut write = 0;
+            for (read, nibble) in original_hash_mask.iter().enumerate() {
+                if new_hash_mask.is_bit_set(nibble) {
+                    hashes[write] = hashes[read];
+                    write += 1;
+                }
+            }
+            hashes.truncate(write);
+
+            node.hash_mask = new_hash_mask;
+
+            if node.hash_mask.is_empty() && node.tree_mask.is_empty() {
+                return false;
+            }
+        }
+
+        true
+    }
+
+    /// Apply masking to entries, advancing past fully-masked nodes.
+    fn mask_entries(
+        &mut self,
+        mut entry: Option<(Nibbles, BranchNodeCompact)>,
+    ) -> Result<Option<(Nibbles, BranchNodeCompact)>, DatabaseError> {
+        while let Some((key, mut node)) = entry {
+            if self.mask_node(&key, &mut node) {
+                return Ok(Some((key, node)));
+            }
+            entry = self.cursor.next()?;
+        }
+        Ok(None)
+    }
+}
+
+impl<C: TrieCursor> TrieCursor for MaskedTrieCursor<C> {
+    fn seek_exact(
+        &mut self,
+        key: Nibbles,
+    ) -> Result<Option<(Nibbles, BranchNodeCompact)>, DatabaseError> {
+        if let Some((key, mut node)) = self.cursor.seek_exact(key)? {
+            if self.mask_node(&key, &mut node) {
+                Ok(Some((key, node)))
+            } else {
+                Ok(None)
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn seek(
+        &mut self,
+        key: Nibbles,
+    ) -> Result<Option<(Nibbles, BranchNodeCompact)>, DatabaseError> {
+        let entry = self.cursor.seek(key)?;
+        self.mask_entries(entry)
+    }
+
+    fn next(&mut self) -> Result<Option<(Nibbles, BranchNodeCompact)>, DatabaseError> {
+        let entry = self.cursor.next()?;
+        self.mask_entries(entry)
+    }
+
+    fn current(&mut self) -> Result<Option<Nibbles>, DatabaseError> {
+        self.cursor.current()
+    }
+
+    fn reset(&mut self) {
+        self.cursor.reset();
+    }
+}
+
+impl<C: TrieStorageCursor> TrieStorageCursor for MaskedTrieCursor<C> {
+    fn set_hashed_address(&mut self, hashed_address: B256) {
+        self.cursor.set_hashed_address(hashed_address);
+        if let Some(storage_prefix_sets) = &self.storage_prefix_sets {
+            self.prefix_set = storage_prefix_sets.get(&hashed_address).cloned().unwrap_or_default();
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::trie_cursor::mock::MockTrieCursor;
+    use parking_lot::Mutex;
+    use reth_trie_common::prefix_set::PrefixSetMut;
+    use std::{collections::BTreeMap, sync::Arc};
+
+    fn make_cursor(nodes: Vec<(Nibbles, BranchNodeCompact)>) -> MockTrieCursor {
+        let map: BTreeMap<Nibbles, BranchNodeCompact> = nodes.into_iter().collect();
+        MockTrieCursor::new(Arc::new(map), Arc::new(Mutex::new(Vec::new())))
+    }
+
+    fn node(state_mask: u16) -> BranchNodeCompact {
+        BranchNodeCompact::new(state_mask, 0, 0, vec![], None)
+    }
+
+    fn node_with_hashes(state_mask: u16, hash_mask: u16, hashes: Vec<B256>) -> BranchNodeCompact {
+        BranchNodeCompact::new(state_mask, 0, hash_mask, hashes, None)
+    }
+
+    fn node_with_tree_mask(
+        state_mask: u16,
+        tree_mask: u16,
+        hash_mask: u16,
+        hashes: Vec<B256>,
+    ) -> BranchNodeCompact {
+        BranchNodeCompact::new(state_mask, tree_mask, hash_mask, hashes, None)
+    }
+
+    fn hash(byte: u8) -> B256 {
+        B256::repeat_byte(byte)
+    }
+
+    #[test]
+    fn test_seek_masks_matching_child_hashes() {
+        // Node at [0x1] with children 2 and 5 hashed.
+        // Prefix set marks child 2 as changed.
+        let nodes = vec![(
+            Nibbles::from_nibbles([0x1]),
+            node_with_hashes(0b0000_0000_0010_0100, 0b0000_0000_0010_0100, vec![hash(2), hash(5)]),
+        )];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x2]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let result = cursor.seek(Nibbles::default()).unwrap();
+        let (key, node) = result.unwrap();
+        assert_eq!(key, Nibbles::from_nibbles([0x1]));
+        // Hash bit 2 should be unset, only bit 5 remains.
+        assert!(!node.hash_mask.is_bit_set(2));
+        assert!(node.hash_mask.is_bit_set(5));
+        assert_eq!(&*node.hashes, &[hash(5)]);
+    }
+
+    #[test]
+    fn test_seek_skips_fully_masked_node() {
+        // Node at [0x1] with only child 3 hashed, tree_mask empty.
+        // Prefix set marks child 3 as changed → fully masked → skipped.
+        // Node at [0x2] is unaffected → returned.
+        let nodes = vec![
+            (
+                Nibbles::from_nibbles([0x1]),
+                node_with_hashes(0b0000_0000_0000_1000, 0b0000_0000_0000_1000, vec![hash(3)]),
+            ),
+            (Nibbles::from_nibbles([0x2]), node(0b0000_0000_0000_0001)),
+        ];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x3]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let result = cursor.seek(Nibbles::default()).unwrap();
+        assert_eq!(result, Some((Nibbles::from_nibbles([0x2]), node(0b0000_0000_0000_0001))));
+    }
+
+    #[test]
+    fn test_node_with_tree_mask_not_skipped() {
+        // Node at [0x1] with child 3 hashed, tree_mask has bit 3 set.
+        // Prefix set marks child 3 → hash cleared, but tree_mask keeps the node alive.
+        let nodes = vec![(
+            Nibbles::from_nibbles([0x1]),
+            node_with_tree_mask(
+                0b0000_0000_0000_1000,
+                0b0000_0000_0000_1000,
+                0b0000_0000_0000_1000,
+                vec![hash(3)],
+            ),
+        )];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x3]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let result = cursor.seek(Nibbles::default()).unwrap();
+        let (key, node) = result.unwrap();
+        assert_eq!(key, Nibbles::from_nibbles([0x1]));
+        assert!(node.hash_mask.is_empty());
+        assert!(node.tree_mask.is_bit_set(3));
+        assert!(node.hashes.is_empty());
+    }
+
+    #[test]
+    fn test_seek_exact_masks_hash_bits() {
+        let nodes = vec![(
+            Nibbles::from_nibbles([0x1]),
+            node_with_tree_mask(
+                0b0000_0000_0010_0100,
+                0b0000_0000_0010_0100,
+                0b0000_0000_0010_0100,
+                vec![hash(2), hash(5)],
+            ),
+        )];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x5]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let result = cursor.seek_exact(Nibbles::from_nibbles([0x1])).unwrap();
+        let (_, node) = result.unwrap();
+        assert!(node.hash_mask.is_bit_set(2));
+        assert!(!node.hash_mask.is_bit_set(5));
+        assert_eq!(&*node.hashes, &[hash(2)]);
+    }
+
+    #[test]
+    fn test_seek_exact_returns_none_for_fully_masked() {
+        let nodes = vec![(
+            Nibbles::from_nibbles([0x1]),
+            node_with_hashes(0b0000_0000_0000_0100, 0b0000_0000_0000_0100, vec![hash(2)]),
+        )];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x2]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let result = cursor.seek_exact(Nibbles::from_nibbles([0x1])).unwrap();
+        assert_eq!(result, None);
+    }
+
+    #[test]
+    fn test_next_masks_and_skips() {
+        // Three nodes: [0x1] unaffected, [0x2] fully masked, [0x3] unaffected.
+        let nodes = vec![
+            (
+                Nibbles::from_nibbles([0x1]),
+                node_with_hashes(0b0000_0000_0000_0010, 0b0000_0000_0000_0010, vec![hash(1)]),
+            ),
+            (
+                Nibbles::from_nibbles([0x2]),
+                node_with_hashes(0b0000_0000_0001_0000, 0b0000_0000_0001_0000, vec![hash(4)]),
+            ),
+            (
+                Nibbles::from_nibbles([0x3]),
+                node_with_hashes(0b0000_0000_0100_0000, 0b0000_0000_0100_0000, vec![hash(6)]),
+            ),
+        ];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x2, 0x4]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        // seek to [0x1], no match → returned unchanged.
+        let result = cursor.seek(Nibbles::from_nibbles([0x1])).unwrap();
+        let (key, node) = result.unwrap();
+        assert_eq!(key, Nibbles::from_nibbles([0x1]));
+        assert_eq!(&*node.hashes, &[hash(1)]);
+
+        // next() should skip [0x2] (fully masked), returning [0x3].
+        let result = cursor.next().unwrap();
+        let (key, node) = result.unwrap();
+        assert_eq!(key, Nibbles::from_nibbles([0x3]));
+        assert_eq!(&*node.hashes, &[hash(6)]);
+    }
+
+    #[test]
+    fn test_no_match_returns_unchanged() {
+        let nodes = vec![(
+            Nibbles::from_nibbles([0x2]),
+            node_with_hashes(0b0000_0000_0000_0010, 0b0000_0000_0000_0010, vec![hash(1)]),
+        )];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x3]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let result = cursor.seek(Nibbles::default()).unwrap();
+        let (key, node) = result.unwrap();
+        assert_eq!(key, Nibbles::from_nibbles([0x2]));
+        // Unchanged — prefix set doesn't match [0x2].
+        assert!(node.hash_mask.is_bit_set(1));
+        assert_eq!(&*node.hashes, &[hash(1)]);
+    }
+
+    #[test]
+    fn test_empty_prefix_set_returns_all_unchanged() {
+        let h1 = hash(1);
+        let h2 = hash(2);
+        let nodes = vec![
+            (
+                Nibbles::from_nibbles([0x1]),
+                node_with_hashes(0b0000_0000_0000_0010, 0b0000_0000_0000_0010, vec![h1]),
+            ),
+            (
+                Nibbles::from_nibbles([0x2]),
+                node_with_hashes(0b0000_0000_0000_0100, 0b0000_0000_0000_0100, vec![h2]),
+            ),
+        ];
+
+        let ps = PrefixSetMut::default();
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let r1 = cursor.seek(Nibbles::default()).unwrap().unwrap();
+        assert_eq!(r1.0, Nibbles::from_nibbles([0x1]));
+        assert_eq!(&*r1.1.hashes, &[h1]);
+
+        let r2 = cursor.next().unwrap().unwrap();
+        assert_eq!(r2.0, Nibbles::from_nibbles([0x2]));
+        assert_eq!(&*r2.1.hashes, &[h2]);
+
+        assert_eq!(cursor.next().unwrap(), None);
+    }
+
+    #[test]
+    fn test_root_hash_cleared_on_mask() {
+        let mut n =
+            node_with_hashes(0b0000_0000_0010_0100, 0b0000_0000_0010_0100, vec![hash(2), hash(5)]);
+        n.root_hash = Some(hash(0xFF));
+
+        let nodes = vec![(Nibbles::from_nibbles([0x1]), n)];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x2]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let (_, node) = cursor.seek(Nibbles::default()).unwrap().unwrap();
+        assert_eq!(node.root_hash, None);
+    }
+
+    #[test]
+    fn test_node_without_hashes_returned_unchanged() {
+        // Node with state_mask only (no hashes, no tree_mask) should pass through.
+        let nodes = vec![(Nibbles::from_nibbles([0x1]), node(0b0000_0000_0000_0011))];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x0]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let result = cursor.seek(Nibbles::default()).unwrap();
+        assert_eq!(result, Some((Nibbles::from_nibbles([0x1]), node(0b0000_0000_0000_0011))));
+    }
+
+    #[test]
+    fn test_empty_cursor_returns_none() {
+        let nodes = vec![];
+        let ps = PrefixSetMut::default();
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        assert_eq!(cursor.seek(Nibbles::default()).unwrap(), None);
+    }
+
+    #[test]
+    fn test_reset_delegates() {
+        let nodes =
+            vec![(Nibbles::from_nibbles([0x1]), node(1)), (Nibbles::from_nibbles([0x2]), node(2))];
+
+        let ps = PrefixSetMut::default();
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let _ = cursor.seek(Nibbles::from_nibbles([0x1])).unwrap();
+        assert_eq!(cursor.current().unwrap(), Some(Nibbles::from_nibbles([0x1])));
+
+        cursor.reset();
+        assert_eq!(cursor.current().unwrap(), None);
+    }
+
+    #[test]
+    fn test_partial_mask_preserves_remaining_hashes() {
+        // Node at [0x1] with children 0, 3, 7 hashed.
+        // Prefix set marks children 0 and 7 as changed.
+        // Only hash for child 3 should remain.
+        let nodes = vec![(
+            Nibbles::from_nibbles([0x1]),
+            node_with_tree_mask(
+                0b0000_0000_1000_1001,
+                0b0000_0000_1000_1001,
+                0b0000_0000_1000_1001,
+                vec![hash(0), hash(3), hash(7)],
+            ),
+        )];
+
+        let mut ps = PrefixSetMut::default();
+        ps.insert(Nibbles::from_nibbles([0x1, 0x0]));
+        ps.insert(Nibbles::from_nibbles([0x1, 0x7]));
+
+        let inner = make_cursor(nodes);
+        let mut cursor = MaskedTrieCursor::new(inner, ps.freeze());
+
+        let (key, node) = cursor.seek(Nibbles::default()).unwrap().unwrap();
+        assert_eq!(key, Nibbles::from_nibbles([0x1]));
+        assert!(!node.hash_mask.is_bit_set(0));
+        assert!(node.hash_mask.is_bit_set(3));
+        assert!(!node.hash_mask.is_bit_set(7));
+        assert_eq!(&*node.hashes, &[hash(3)]);
+        assert_eq!(node.root_hash, None);
+    }
+
+    mod proptest_tests {
+        use crate::{
+            hashed_cursor::{mock::MockHashedCursorFactory, HashedPostStateCursorFactory},
+            proof::Proof,
+            trie_cursor::{
+                masked::MaskedTrieCursorFactory, mock::MockTrieCursorFactory,
+                noop::NoopTrieCursorFactory,
+            },
+            StateRoot,
+        };
+        use alloy_primitives::{map::B256Set, B256, U256};
+        use proptest::prelude::*;
+        use reth_primitives_traits::Account;
+        use reth_trie_common::{HashedPostState, HashedStorage, MultiProofTargets};
+
+        fn account_strategy() -> impl Strategy<Value = Account> {
+            (any::<u64>(), any::<u64>(), any::<[u8; 32]>()).prop_map(
+                |(nonce, balance, code_hash)| Account {
+                    nonce,
+                    balance: U256::from(balance),
+                    bytecode_hash: Some(B256::from(code_hash)),
+                },
+            )
+        }
+
+        fn storage_value_strategy() -> impl Strategy<Value = U256> {
+            any::<u64>().prop_filter("non-zero", |v| *v != 0).prop_map(U256::from)
+        }
+
+        /// Generates a base dataset of 1000 storage slots for account `B256::ZERO`,
+        /// a 200-entry changeset partially overlapping with the base, and random
+        /// proof targets partially overlapping with both.
+        #[allow(clippy::type_complexity)]
+        fn test_input_strategy(
+        ) -> impl Strategy<Value = (Vec<(B256, U256)>, Account, Vec<(B256, Option<U256>)>, Vec<B256>)>
+        {
+            (
+                // 1000 base storage slots: unique keys with non-zero values
+                prop::collection::vec(
+                    (any::<[u8; 32]>().prop_map(B256::from), storage_value_strategy()),
+                    1000,
+                ),
+                account_strategy(),
+                // 200 changeset entries: (key, Option<value>) where None = removal
+                prop::collection::vec(
+                    (
+                        any::<[u8; 32]>().prop_map(B256::from),
+                        prop::option::of(storage_value_strategy()),
+                    ),
+                    200,
+                ),
+                // Extra random keys for proof targets
+                prop::collection::vec(any::<[u8; 32]>().prop_map(B256::from), 50),
+            )
+                .prop_flat_map(
+                    |(base_slots, account, changeset_raw, extra_targets)| {
+                        // Dedup base slots by key
+                        let mut base_map = alloy_primitives::map::B256Map::default();
+                        for (k, v) in &base_slots {
+                            base_map.insert(*k, *v);
+                        }
+                        let base_deduped: Vec<(B256, U256)> =
+                            base_map.iter().map(|(&k, &v)| (k, v)).collect();
+                        let base_keys: Vec<B256> = base_deduped.iter().map(|(k, _)| *k).collect();
+
+                        // Build changeset: 50% overlap with base keys, 50% new keys
+                        let changeset_len = changeset_raw.len();
+                        let half = changeset_len / 2;
+                        let base_keys_for_overlap = base_keys.clone();
+
+                        // Use indices to select from base keys for overlap portion
+                        let overlap_indices =
+                            prop::collection::vec(0..base_keys_for_overlap.len().max(1), half);
+
+                        overlap_indices.prop_map(move |indices| {
+                            let mut changeset: Vec<(B256, Option<U256>)> = Vec::new();
+
+                            // First half: overlapping with base keys
+                            for (i, (_, value)) in
+                                indices.iter().zip(changeset_raw.iter()).take(half)
+                            {
+                                let key = if base_keys_for_overlap.is_empty() {
+                                    changeset_raw[*i].0
+                                } else {
+                                    base_keys_for_overlap[*i % base_keys_for_overlap.len()]
+                                };
+                                changeset.push((key, *value));
+                            }
+
+                            // Second half: new keys from changeset_raw
+                            for (key, value) in changeset_raw.iter().skip(half) {
+                                changeset.push((*key, *value));
+                            }
+
+                            // Build proof targets: mix of base keys, changeset keys, and randoms
+                            let changeset_keys: Vec<B256> =
+                                changeset.iter().map(|(k, _)| *k).collect();
+                            let mut proof_slot_targets: Vec<B256> = Vec::new();
+
+                            // ~40% from base
+                            for k in base_keys.iter().take(40) {
+                                proof_slot_targets.push(*k);
+                            }
+                            // ~30% from changeset
+                            for k in changeset_keys.iter().take(30) {
+                                proof_slot_targets.push(*k);
+                            }
+                            // ~30% random
+                            for k in extra_targets.iter().take(30) {
+                                proof_slot_targets.push(*k);
+                            }
+
+                            (base_deduped.clone(), account, changeset, proof_slot_targets)
+                        })
+                    },
+                )
+        }
+
+        proptest! {
+            #![proptest_config(ProptestConfig::with_cases(50))]
+            #[test]
+            fn proptest_masked_cursor_multiproof_equivalence(
+                (base_slots, account, changeset, proof_slot_targets) in test_input_strategy()
+            ) {
+                reth_tracing::init_test_tracing();
+
+                let hashed_address = B256::ZERO;
+
+                // Step 1: Create the base hashed post state with a single account
+                // and 1000 storage slots.
+                let base_state = HashedPostState {
+                    accounts: std::iter::once((hashed_address, Some(account))).collect(),
+                    storages: std::iter::once((
+                        hashed_address,
+                        HashedStorage::from_iter(false, base_slots),
+                    ))
+                    .collect(),
+                };
+
+                // Step 2: Compute trie updates from state root over the full base state.
+                let base_hashed_cursor_factory =
+                    MockHashedCursorFactory::from_hashed_post_state(base_state);
+                let (_, trie_updates) = StateRoot::new(
+                    NoopTrieCursorFactory,
+                    base_hashed_cursor_factory.clone(),
+                )
+                .root_with_updates()
+                .expect("state root computation should succeed");
+
+                // Step 3: Create a MockTrieCursorFactory from those trie updates.
+                let mock_trie_cursor_factory =
+                    MockTrieCursorFactory::from_trie_updates(trie_updates);
+
+                // Step 4: Build the changeset post state. Removals use U256::ZERO.
+                let changeset_storage: Vec<(B256, U256)> = changeset
+                    .iter()
+                    .map(|(k, v)| (*k, v.unwrap_or(U256::ZERO)))
+                    .collect();
+                let changeset_state = HashedPostState {
+                    accounts: std::iter::once((hashed_address, Some(account))).collect(),
+                    storages: std::iter::once((
+                        hashed_address,
+                        HashedStorage::from_iter(false, changeset_storage),
+                    ))
+                    .collect(),
+                };
+
+                // Step 5: Generate prefix sets from the changeset.
+                let prefix_sets_mut = changeset_state.construct_prefix_sets();
+
+                // Step 6: Build proof targets.
+                let slot_targets: B256Set = proof_slot_targets.into_iter().collect();
+                let targets =
+                    MultiProofTargets::from_iter([(hashed_address, slot_targets)]);
+
+                // Step 7: Create the HashedPostStateCursorFactory overlaying changeset
+                // on the base.
+                let changeset_sorted = changeset_state.into_sorted();
+                let overlay_cursor_factory = HashedPostStateCursorFactory::new(
+                    base_hashed_cursor_factory,
+                    &changeset_sorted,
+                );
+
+                // Step 8a: Approach A — prefix sets passed to Proof directly.
+                let proof_a = Proof::new(
+                    mock_trie_cursor_factory.clone(),
+                    overlay_cursor_factory.clone(),
+                )
+                .with_prefix_sets_mut(prefix_sets_mut.clone());
+                let multiproof_a = proof_a
+                    .multiproof(targets.clone())
+                    .expect("multiproof A should succeed");
+
+                // Step 8b: Approach B — MaskedTrieCursorFactory, no prefix sets on Proof.
+                let masked_trie_cursor_factory = MaskedTrieCursorFactory::new(
+                    mock_trie_cursor_factory,
+                    prefix_sets_mut.freeze(),
+                );
+                let proof_b = Proof::new(
+                    masked_trie_cursor_factory,
+                    overlay_cursor_factory,
+                );
+                let multiproof_b = proof_b
+                    .multiproof(targets)
+                    .expect("multiproof B should succeed");
+
+                // Step 9: Compare results.
+                assert_eq!(
+                    multiproof_a, multiproof_b,
+                    "multiproof with prefix sets should equal multiproof with masked cursor"
+                );
+            }
+        }
+    }
+}
--- a/crates/trie/trie/src/trie_cursor/mod.rs
+++ b/crates/trie/trie/src/trie_cursor/mod.rs
@@ -11,6 +11,9 @@ pub mod subnode;
 /// Noop trie cursor implementations.
 pub mod noop;

+/// Masked trie cursor wrapper that skips nodes matching a prefix set.
+pub mod masked;
+
 /// Depth-first trie iterator.
 pub mod depth_first;

--- a/crates/trie/trie/src/witness.rs
+++ b/crates/trie/trie/src/witness.rs
@@ -1,23 +1,29 @@
 use crate::{
    hashed_cursor::{HashedCursor, HashedCursorFactory},
-    prefix_set::TriePrefixSetsMut,
-    proof::Proof,
-    proof_v2,
+    proof::{Proof, ProofTrieNodeProviderFactory},
    trie_cursor::TrieCursorFactory,
-    TRIE_ACCOUNT_RLP_MAX_SIZE,
 };
+use alloy_rlp::EMPTY_STRING_CODE;
+use alloy_trie::EMPTY_ROOT_HASH;
+use reth_trie_common::HashedPostState;
+use reth_trie_sparse::SparseTrie;
+
 use alloy_primitives::{
    keccak256,
-    map::{B256Map, HashMap},
-    Bytes, B256, U256,
+    map::{B256Map, B256Set, Entry, HashMap},
+    Bytes, B256,
 };
-use alloy_rlp::{Encodable, EMPTY_STRING_CODE};
-use alloy_trie::{nodes::BranchNodeRef, EMPTY_ROOT_HASH};
-use reth_execution_errors::{SparseStateTrieErrorKind, StateProofError, TrieWitnessError};
-use reth_trie_common::{
-    DecodedMultiProofV2, HashedPostState, MultiProofTargetsV2, ProofV2Target, TrieNodeV2,
+use itertools::Itertools;
+use reth_execution_errors::{
+    SparseStateTrieErrorKind, SparseTrieError, SparseTrieErrorKind, StateProofError,
+    TrieWitnessError,
 };
-use reth_trie_sparse::{LeafUpdate, SparseStateTrie, SparseTrie as _};
+use reth_trie_common::{MultiProofTargets, Nibbles};
+use reth_trie_sparse::{
+    provider::{RevealedNode, TrieNodeProvider, TrieNodeProviderFactory},
+    SparseStateTrie,
+};
+use std::sync::mpsc;

 /// State transition witness for the trie.
 #[derive(Debug)]
@@ -26,8 +32,6 @@ pub struct TrieWitness<T, H> {
    trie_cursor_factory: T,
    /// The factory for hashed cursors.
    hashed_cursor_factory: H,
-    /// A set of prefix sets that have changes.
-    prefix_sets: TriePrefixSetsMut,
    /// Flag indicating whether the root node should always be included (even if the target state
    /// is empty). This setting is useful if the caller wants to verify the witness against the
    /// parent state root.
@@ -43,7 +47,6 @@ impl<T, H> TrieWitness<T, H> {
        Self {
            trie_cursor_factory,
            hashed_cursor_factory,
-            prefix_sets: TriePrefixSetsMut::default(),
            always_include_root_node: false,
            witness: HashMap::default(),
        }
@@ -54,7 +57,6 @@ impl<T, H> TrieWitness<T, H> {
        TrieWitness {
            trie_cursor_factory,
            hashed_cursor_factory: self.hashed_cursor_factory,
-            prefix_sets: self.prefix_sets,
            always_include_root_node: self.always_include_root_node,
            witness: self.witness,
        }
@@ -65,18 +67,11 @@ impl<T, H> TrieWitness<T, H> {
        TrieWitness {
            trie_cursor_factory: self.trie_cursor_factory,
            hashed_cursor_factory,
-            prefix_sets: self.prefix_sets,
            always_include_root_node: self.always_include_root_node,
            witness: self.witness,
        }
    }

-    /// Set the prefix sets. They have to be mutable in order to allow extension with proof target.
-    pub fn with_prefix_sets_mut(mut self, prefix_sets: TriePrefixSetsMut) -> Self {
-        self.prefix_sets = prefix_sets;
-        self
-    }
-
    /// Set `always_include_root_node` to true. Root node will be included even in empty state.
    /// This setting is useful if the caller wants to verify the witness against the
    /// parent state root.
@@ -97,131 +92,84 @@ where
    /// # Arguments
    ///
    /// `state` - state transition containing both modified and touched accounts and storage slots.
-    pub fn compute(
-        mut self,
-        mut state: HashedPostState,
-    ) -> Result<B256Map<Bytes>, TrieWitnessError> {
+    pub fn compute(mut self, state: HashedPostState) -> Result<B256Map<Bytes>, TrieWitnessError> {
        let is_state_empty = state.is_empty();
        if is_state_empty && !self.always_include_root_node {
            return Ok(Default::default())
        }

-        // Expand wiped storages into explicit zero-value entries for every existing slot,
-        // so that downstream code can treat all storages uniformly.
-        self.expand_wiped_storages(&mut state)?;
-
        let proof_targets = if is_state_empty {
-            MultiProofTargetsV2 {
-                account_targets: vec![ProofV2Target::new(B256::ZERO)],
-                ..Default::default()
-            }
+            MultiProofTargets::account(B256::ZERO)
        } else {
-            Self::get_proof_targets(&state)
+            self.get_proof_targets(&state)?
        };
        let multiproof =
            Proof::new(self.trie_cursor_factory.clone(), self.hashed_cursor_factory.clone())
-                .with_prefix_sets_mut(self.prefix_sets.clone())
-                .multiproof_v2(proof_targets)?;
+                .multiproof(proof_targets.clone())?;

        // No need to reconstruct the rest of the trie, we just need to include
        // the root node and return.
        if is_state_empty {
            let (root_hash, root_node) = if let Some(root_node) =
-                multiproof.account_proofs.into_iter().find(|n| n.path.is_empty())
+                multiproof.account_subtree.into_inner().remove(&Nibbles::default())
            {
-                let mut encoded = Vec::new();
-                root_node.node.encode(&mut encoded);
-                let bytes = Bytes::from(encoded);
-                (keccak256(&bytes), bytes)
+                (keccak256(&root_node), root_node)
            } else {
                (EMPTY_ROOT_HASH, Bytes::from([EMPTY_STRING_CODE]))
            };
            return Ok(B256Map::from_iter([(root_hash, root_node)]))
        }

-        // Record all nodes from multiproof in the witness.
-        self.record_multiproof_nodes(&multiproof);
+        // Record all nodes from multiproof in the witness
+        for account_node in multiproof.account_subtree.values() {
+            if let Entry::Vacant(entry) = self.witness.entry(keccak256(account_node.as_ref())) {
+                entry.insert(account_node.clone());
+            }
+        }
+        for storage_node in multiproof.storages.values().flat_map(|s| s.subtree.values()) {
+            if let Entry::Vacant(entry) = self.witness.entry(keccak256(storage_node.as_ref())) {
+                entry.insert(storage_node.clone());
+            }
+        }

+        let (tx, rx) = mpsc::channel();
+        let blinded_provider_factory = WitnessTrieNodeProviderFactory::new(
+            ProofTrieNodeProviderFactory::new(self.trie_cursor_factory, self.hashed_cursor_factory),
+            tx,
+        );
        let mut sparse_trie = SparseStateTrie::new();
-        sparse_trie.reveal_decoded_multiproof_v2(multiproof)?;
+        sparse_trie.reveal_multiproof(multiproof)?;

-        // Build storage leaf updates for all accounts with storage changes, split into
-        // removals and upserts. Removals must be applied first so that branch collapse
-        // detection fires correctly: if a removal and an insertion target siblings under
-        // the same branch, processing the removal first may reduce the branch to a single
-        // blinded child, triggering a proof fetch for the sibling. Processing the insertion
-        // first would add a new child that keeps the count above one, masking the need.
-        let mut storage_removals: B256Map<B256Map<LeafUpdate>> = B256Map::default();
-        let mut storage_upserts: B256Map<B256Map<LeafUpdate>> = B256Map::default();
-        for (hashed_address, storage) in &state.storages {
-            for (&hashed_slot, value) in &storage.storage {
-                if value.is_zero() {
-                    storage_removals
-                        .entry(*hashed_address)
-                        .or_default()
-                        .insert(hashed_slot, LeafUpdate::Changed(vec![]));
+        // Attempt to update state trie to gather additional information for the witness.
+        for (hashed_address, hashed_slots) in
+            proof_targets.into_iter().sorted_unstable_by_key(|(ha, _)| *ha)
+        {
+            // Update storage trie first.
+            let provider = blinded_provider_factory.storage_node_provider(hashed_address);
+            let storage = state.storages.get(&hashed_address);
+            let storage_trie = sparse_trie.storage_trie_mut(&hashed_address).ok_or(
+                SparseStateTrieErrorKind::SparseStorageTrie(
+                    hashed_address,
+                    SparseTrieErrorKind::Blind,
+                ),
+            )?;
+            for hashed_slot in hashed_slots.into_iter().sorted_unstable() {
+                let storage_nibbles = Nibbles::unpack(hashed_slot);
+                let maybe_leaf_value = storage
+                    .and_then(|s| s.storage.get(&hashed_slot))
+                    .filter(|v| !v.is_zero())
+                    .map(|v| alloy_rlp::encode_fixed_size(v).to_vec());
+
+                if let Some(value) = maybe_leaf_value {
+                    storage_trie.update_leaf(storage_nibbles, value, &provider).map_err(|err| {
+                        SparseStateTrieErrorKind::SparseStorageTrie(hashed_address, err.into_kind())
+                    })?;
                } else {
-                    storage_upserts.entry(*hashed_address).or_default().insert(
-                        hashed_slot,
-                        LeafUpdate::Changed(alloy_rlp::encode_fixed_size(value).to_vec()),
-                    );
+                    storage_trie.remove_leaf(&storage_nibbles, &provider).map_err(|err| {
+                        SparseStateTrieErrorKind::SparseStorageTrie(hashed_address, err.into_kind())
+                    })?;
                }
            }
-        }
-
-        // Apply storage removals first, then upserts, fetching additional proofs as needed.
-        for storage_updates in [&mut storage_removals, &mut storage_upserts] {
-            loop {
-                let mut targets = MultiProofTargetsV2::default();
-
-                for (&hashed_address, slot_updates) in storage_updates.iter_mut() {
-                    if slot_updates.is_empty() {
-                        continue;
-                    }
-                    let storage_trie = sparse_trie
-                        .storage_trie_mut(&hashed_address)
-                        .expect("storage trie was revealed from multiproof");
-                    storage_trie
-                        .update_leaves(slot_updates, |key, min_len| {
-                            targets
-                                .storage_targets
-                                .entry(hashed_address)
-                                .or_default()
-                                .push(ProofV2Target::new(key).with_min_len(min_len));
-                        })
-                        .map_err(|err| {
-                            SparseStateTrieErrorKind::SparseStorageTrie(
-                                hashed_address,
-                                err.into_kind(),
-                            )
-                        })?;
-                }
-
-                if targets.is_empty() {
-                    break;
-                }
-
-                let multiproof = Proof::new(
-                    self.trie_cursor_factory.clone(),
-                    self.hashed_cursor_factory.clone(),
-                )
-                .with_prefix_sets_mut(self.prefix_sets.clone())
-                .multiproof_v2(targets)?;
-                self.record_multiproof_nodes(&multiproof);
-                sparse_trie.reveal_decoded_multiproof_v2(multiproof)?;
-            }
-        }
-
-        // Build account leaf updates, split into removals and upserts (same reasoning
-        // as for storage updates above).
-        let mut account_removals: B256Map<LeafUpdate> = B256Map::default();
-        let mut account_upserts: B256Map<LeafUpdate> = B256Map::default();
-        for &hashed_address in state.accounts.keys().chain(state.storages.keys()) {
-            if account_removals.contains_key(&hashed_address) ||
-                account_upserts.contains_key(&hashed_address)
-            {
-                continue;
-            }

            let account = state
                .accounts
@@ -229,149 +177,105 @@ where
                .ok_or(TrieWitnessError::MissingAccount(hashed_address))?
                .unwrap_or_default();

-            let storage_root =
-                if let Some(storage_trie) = sparse_trie.storage_trie_mut(&hashed_address) {
-                    storage_trie.root()
-                } else {
-                    self.account_storage_root(hashed_address)?
-                };
-
-            if account.is_empty() && storage_root == EMPTY_ROOT_HASH {
-                account_removals.insert(hashed_address, LeafUpdate::Changed(vec![]));
-            } else {
-                let mut rlp = Vec::with_capacity(TRIE_ACCOUNT_RLP_MAX_SIZE);
-                account.into_trie_account(storage_root).encode(&mut rlp);
-                account_upserts.insert(hashed_address, LeafUpdate::Changed(rlp));
+            if !sparse_trie.update_account(hashed_address, account, &blinded_provider_factory)? {
+                let nibbles = Nibbles::unpack(hashed_address);
+                sparse_trie.remove_account_leaf(&nibbles, &blinded_provider_factory)?;
            }
-        }

-        // Apply account removals first, then upserts, fetching additional proofs as needed.
-        for account_updates in [&mut account_removals, &mut account_upserts] {
-            loop {
-                let mut targets = MultiProofTargetsV2::default();
-
-                sparse_trie
-                    .trie_mut()
-                    .update_leaves(account_updates, |key, min_len| {
-                        targets.account_targets.push(ProofV2Target::new(key).with_min_len(min_len));
-                    })
-                    .map_err(SparseStateTrieErrorKind::from)?;
-
-                if targets.is_empty() {
-                    break;
-                }
-
-                let multiproof = Proof::new(
-                    self.trie_cursor_factory.clone(),
-                    self.hashed_cursor_factory.clone(),
-                )
-                .with_prefix_sets_mut(self.prefix_sets.clone())
-                .multiproof_v2(targets)?;
-                self.record_multiproof_nodes(&multiproof);
-                sparse_trie.reveal_decoded_multiproof_v2(multiproof)?;
+            while let Ok(node) = rx.try_recv() {
+                self.witness.insert(keccak256(&node), node);
            }
        }

        Ok(self.witness)
    }

-    /// Record all nodes from a V2 decoded multiproof in the witness.
-    fn record_multiproof_nodes(&mut self, multiproof: &DecodedMultiProofV2) {
-        let mut encoded = Vec::new();
-        for proof_node in &multiproof.account_proofs {
-            self.record_witness_node(&proof_node.node, &mut encoded);
-        }
-        for proof_nodes in multiproof.storage_proofs.values() {
-            for proof_node in proof_nodes {
-                self.record_witness_node(&proof_node.node, &mut encoded);
-            }
-        }
-    }
-
-    /// Record a single [`TrieNodeV2`] in the witness.
-    fn record_witness_node(&mut self, node: &TrieNodeV2, encoded: &mut Vec<u8>) {
-        encoded.clear();
-        node.encode(encoded);
-        let bytes = Bytes::from(encoded.clone());
-        self.witness.entry(keccak256(&bytes)).or_insert(bytes);
-
-        if let TrieNodeV2::Branch(branch) = node &&
-            !branch.key.is_empty()
-        {
-            encoded.clear();
-            BranchNodeRef::new(&branch.stack, branch.state_mask).encode(encoded);
-            let bytes = Bytes::from(encoded.clone());
-            self.witness.entry(keccak256(&bytes)).or_insert(bytes);
-        }
-    }
-
-    /// Compute the storage root for an account by walking the storage trie using the cursor
-    /// factories and trie input prefix sets. Records the root node in the witness.
-    fn account_storage_root(&mut self, hashed_address: B256) -> Result<B256, TrieWitnessError> {
-        let storage_trie_cursor = self
-            .trie_cursor_factory
-            .storage_trie_cursor(hashed_address)
-            .map_err(StateProofError::from)?;
-        let hashed_storage_cursor = self
-            .hashed_cursor_factory
-            .hashed_storage_cursor(hashed_address)
-            .map_err(StateProofError::from)?;
-        let mut calculator = proof_v2::StorageProofCalculator::new_storage(
-            storage_trie_cursor,
-            hashed_storage_cursor,
-        );
-        if let Some(prefix_set) = self.prefix_sets.storage_prefix_sets.get(&hashed_address) {
-            calculator = calculator.with_prefix_set(prefix_set.clone().freeze());
-        }
-        let root_node = calculator.storage_root_node(hashed_address)?;
-        let root_hash = calculator
-            .compute_root_hash(core::slice::from_ref(&root_node))?
-            .unwrap_or(EMPTY_ROOT_HASH);
-        drop(calculator);
-        let mut encoded = Vec::new();
-        self.record_witness_node(&root_node.node, &mut encoded);
-        Ok(root_hash)
-    }
-
-    /// Expand wiped storages into explicit zero-value entries for every existing slot in the
-    /// database. After this, all storages can be treated uniformly without special wiped handling.
-    fn expand_wiped_storages(&self, state: &mut HashedPostState) -> Result<(), StateProofError> {
-        for (hashed_address, storage) in &mut state.storages {
-            if !storage.wiped {
-                continue;
-            }
-            let mut storage_cursor =
-                self.hashed_cursor_factory.hashed_storage_cursor(*hashed_address)?;
-            let mut current_entry = storage_cursor.seek(B256::ZERO)?;
-            while let Some((hashed_slot, _)) = current_entry {
-                storage.storage.entry(hashed_slot).or_insert(U256::ZERO);
-                current_entry = storage_cursor.next()?;
-            }
-            storage.wiped = false;
-        }
-        Ok(())
-    }
-
    /// Retrieve proof targets for incoming hashed state.
-    /// Aggregates all accounts and slots present in the state. Wiped storages must have been
-    /// expanded via [`Self::expand_wiped_storages`] before calling this.
-    fn get_proof_targets(state: &HashedPostState) -> MultiProofTargetsV2 {
-        let mut targets = MultiProofTargetsV2::default();
-        for &hashed_address in state.accounts.keys() {
-            targets.account_targets.push(ProofV2Target::new(hashed_address));
+    /// This method will aggregate all accounts and slots present in the hash state as well as
+    /// select all existing slots from the database for the accounts that have been destroyed.
+    fn get_proof_targets(
+        &self,
+        state: &HashedPostState,
+    ) -> Result<MultiProofTargets, StateProofError> {
+        let mut proof_targets = MultiProofTargets::default();
+        for hashed_address in state.accounts.keys() {
+            proof_targets.insert(*hashed_address, B256Set::default());
        }
-        for (&hashed_address, storage) in &state.storages {
-            if !state.accounts.contains_key(&hashed_address) {
-                targets.account_targets.push(ProofV2Target::new(hashed_address));
+        for (hashed_address, storage) in &state.storages {
+            let mut storage_keys = storage.storage.keys().copied().collect::<B256Set>();
+            if storage.wiped {
+                // storage for this account was destroyed, gather all slots from the current state
+                let mut storage_cursor =
+                    self.hashed_cursor_factory.hashed_storage_cursor(*hashed_address)?;
+                // position cursor at the start
+                let mut current_entry = storage_cursor.seek(B256::ZERO)?;
+                while let Some((hashed_slot, _)) = current_entry {
+                    storage_keys.insert(hashed_slot);
+                    current_entry = storage_cursor.next()?;
+                }
            }
-            // Skip accounts with no storage slot changes — an empty target set would produce
-            // an empty proof vec which cannot be revealed (no root node).
-            if storage.storage.is_empty() {
-                continue;
-            }
-            let storage_keys = storage.storage.keys().map(|k| ProofV2Target::new(*k)).collect();
-            targets.storage_targets.insert(hashed_address, storage_keys);
+            proof_targets.insert(*hashed_address, storage_keys);
        }
-        targets
+        Ok(proof_targets)
+    }
+}
+
+#[derive(Debug, Clone)]
+struct WitnessTrieNodeProviderFactory<F> {
+    /// Trie node provider factory.
+    provider_factory: F,
+    /// Sender for forwarding fetched trie node.
+    tx: mpsc::Sender<Bytes>,
+}
+
+impl<F> WitnessTrieNodeProviderFactory<F> {
+    const fn new(provider_factory: F, tx: mpsc::Sender<Bytes>) -> Self {
+        Self { provider_factory, tx }
+    }
+}
+
+impl<F> TrieNodeProviderFactory for WitnessTrieNodeProviderFactory<F>
+where
+    F: TrieNodeProviderFactory,
+    F::AccountNodeProvider: TrieNodeProvider,
+    F::StorageNodeProvider: TrieNodeProvider,
+{
+    type AccountNodeProvider = WitnessTrieNodeProvider<F::AccountNodeProvider>;
+    type StorageNodeProvider = WitnessTrieNodeProvider<F::StorageNodeProvider>;
+
+    fn account_node_provider(&self) -> Self::AccountNodeProvider {
+        let provider = self.provider_factory.account_node_provider();
+        WitnessTrieNodeProvider::new(provider, self.tx.clone())
+    }
+
+    fn storage_node_provider(&self, account: B256) -> Self::StorageNodeProvider {
+        let provider = self.provider_factory.storage_node_provider(account);
+        WitnessTrieNodeProvider::new(provider, self.tx.clone())
+    }
+}
+
+#[derive(Debug)]
+struct WitnessTrieNodeProvider<P> {
+    /// Proof-based blinded.
+    provider: P,
+    /// Sender for forwarding fetched blinded node.
+    tx: mpsc::Sender<Bytes>,
+}
+
+impl<P> WitnessTrieNodeProvider<P> {
+    const fn new(provider: P, tx: mpsc::Sender<Bytes>) -> Self {
+        Self { provider, tx }
+    }
+}
+
+impl<P: TrieNodeProvider> TrieNodeProvider for WitnessTrieNodeProvider<P> {
+    fn trie_node(&self, path: &Nibbles) -> Result<Option<RevealedNode>, SparseTrieError> {
+        let maybe_node = self.provider.trie_node(path)?;
+        if let Some(node) = &maybe_node {
+            self.tx
+                .send(node.node.clone())
+                .map_err(|error| SparseTrieErrorKind::Other(Box::new(error)))?;
+        }
+        Ok(maybe_node)
    }
 }
--- a/examples/custom-beacon-withdrawals/src/main.rs
+++ b/examples/custom-beacon-withdrawals/src/main.rs
@@ -5,7 +5,7 @@

 use alloy_eips::eip4895::Withdrawal;
 use alloy_evm::{
-    block::{BlockExecutorFactory, BlockExecutorFor, ExecutableTx, GasOutput},
+    block::{BlockExecutorFactory, BlockExecutorFor, ExecutableTx},
    eth::{EthBlockExecutionCtx, EthBlockExecutor, EthTxResult},
    precompiles::PrecompilesMap,
    revm::context::Block as _,
@@ -211,10 +211,7 @@ where
        self.inner.execute_transaction_without_commit(tx)
    }

-    fn commit_transaction(
-        &mut self,
-        output: Self::Result,
-    ) -> Result<GasOutput, BlockExecutionError> {
+    fn commit_transaction(&mut self, output: Self::Result) -> Result<u64, BlockExecutionError> {
        self.inner.commit_transaction(output)
    }

--- a/examples/custom-evm/src/main.rs
+++ b/examples/custom-evm/src/main.rs
@@ -110,7 +110,7 @@ pub fn prague_custom() -> &'static Precompiles {
        let precompile = Precompile::new(
            PrecompileId::custom("custom"),
            address!("0x0000000000000000000000000000000000000999"),
-            |_, _| PrecompileResult::Ok(PrecompileOutput::new(0, 0, Bytes::new())),
+            |_, _| PrecompileResult::Ok(PrecompileOutput::new(0, Bytes::new())),
        );
        precompiles.extend([precompile]);
        precompiles
--- a/scripts/compare-witness-recent.sh
+++ b/scripts/compare-witness-recent.sh
@@ -1,149 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-usage() {
-    echo "Usage: $0 [-n NUM_BLOCKS] [-b BLOCK] <remote_rpc_url>"
-    echo ""
-    echo "Fetches debug_executionWitness from both localhost:8545 and the"
-    echo "given remote RPC, then compares them."
-    echo ""
-    echo "Options:"
-    echo "  -n NUM   Compare the last NUM blocks (default: 20)"
-    echo "  -b BLOCK Compare only the given block number"
-    exit 1
-}
-
-NUM_BLOCKS=20
-SINGLE_BLOCK=""
-
-while getopts "n:b:h" opt; do
-    case "$opt" in
-        n) NUM_BLOCKS="$OPTARG" ;;
-        b) SINGLE_BLOCK="$OPTARG" ;;
-        *) usage ;;
-    esac
-done
-shift $((OPTIND - 1))
-
-if [[ $# -lt 1 ]]; then
-    echo "Error: remote_rpc_url is required" >&2
-    usage
-fi
-REMOTE_RPC="$1"
-LOCAL_RPC="http://127.0.0.1:8545"
-
-log()  { echo "[$(date '+%H:%M:%S')] $*" >&2; }
-
-if [[ -n "$SINGLE_BLOCK" ]]; then
-    start=$SINGLE_BLOCK
-    latest=$SINGLE_BLOCK
-    log "Comparing block $SINGLE_BLOCK"
-else
-    # Get latest block number from local node
-    if ! latest=$(cast bn --rpc-url "$LOCAL_RPC" --rpc-timeout 600 2>&1); then
-        log "FATAL: failed to get block number from local RPC: $latest"
-        exit 1
-    fi
-
-    start=$((latest - NUM_BLOCKS + 1))
-    if (( start < 0 )); then
-        start=0
-    fi
-
-    log "Comparing blocks $start..$latest ($NUM_BLOCKS blocks)"
-fi
-
-errors=0
-
-for (( block = start; block <= latest; block++ )); do
-    block_hex=$(printf '"0x%x"' "$block")
-    log "Checking block $block ($block_hex)"
-
-    # Fetch witness from both RPCs
-    local_witness=$(cast rpc debug_executionWitness "$block_hex" --rpc-url "$LOCAL_RPC" --rpc-timeout 600 2>&1) || {
-        log "WARN: failed to get witness from local RPC for block $block: $local_witness"
-        ((errors++)) || true
-        continue
-    }
-
-    remote_witness=$(cast rpc debug_executionWitness "$block_hex" --rpc-url "$REMOTE_RPC" --rpc-timeout 600 2>&1) || {
-        log "WARN: failed to get witness from remote RPC for block $block: $remote_witness"
-        ((errors++)) || true
-        continue
-    }
-
-    # Normalize: sort all arrays of objects by a stable key so ordering doesn't cause false diffs
-    normalize='walk(if type == "array" then sort_by(if type == "object" then (keys | join(",")) + ":" + (to_entries | map(.value | tostring) | join(",")) else tostring end) else . end) | . as $root | $root'
-    local_file=$(mktemp)
-    remote_file=$(mktemp)
-    trap "rm -f '$local_file' '$remote_file'" EXIT
-    echo "$local_witness"  | jq -S "$normalize" > "$local_file"
-    echo "$remote_witness" | jq -S "$normalize" > "$remote_file"
-
-    # Compare: for "state", local may contain extra nodes (superset OK).
-    # For "codes", "keys", "headers", require exact set equality.
-    has_error=false
-
-    # Check exact-match fields (as sorted sets)
-    for field in codes keys headers; do
-        local_set=$(jq -r --arg f "$field" '.[$f] // [] | sort | .[]' "$local_file")
-        remote_set=$(jq -r --arg f "$field" '.[$f] // [] | sort | .[]' "$remote_file")
-        if [[ "$local_set" != "$remote_set" ]]; then
-            log "ERROR: block $block field '$field' differs"
-            diff <(echo "$remote_set") <(echo "$local_set") | head -30 || true
-            has_error=true
-        fi
-    done
-
-    # Check state: every remote node must be present in local (extras OK)
-    missing=$(jq -r -n \
-        --slurpfile l "$local_file" \
-        --slurpfile r "$remote_file" \
-        '($l[0].state // [] | map({(.):true}) | add // {}) as $local_set |
-         [$r[0].state // [] | .[] | select($local_set[.] | not)] |
-         if length == 0 then empty else .[] end')
-    if [[ -n "$missing" ]]; then
-        n_missing=$(echo "$missing" | wc -l)
-        log "ERROR: block $block state has $n_missing missing node(s) (present in remote, absent in local):"
-        echo "$missing" | head -20
-        has_error=true
-    fi
-
-    extra=$(jq -r -n \
-        --slurpfile l "$local_file" \
-        --slurpfile r "$remote_file" \
-        '($r[0].state // [] | map({(.):true}) | add // {}) as $remote_set |
-         [$l[0].state // [] | .[] | select($remote_set[.] | not)] |
-         if length == 0 then empty else .[] end')
-    n_extra=0
-    if [[ -n "$extra" ]]; then
-        n_extra=$(echo "$extra" | wc -l)
-    fi
-
-    if ! $has_error; then
-        if [[ $n_extra -gt 0 ]]; then
-            log "OK: block $block witnesses match ($n_extra extra state node(s) in local)"
-        else
-            log "OK: block $block witnesses match exactly"
-        fi
-    else
-        cp "$local_file" "witness-local-${block}.json"
-        cp "$remote_file" "witness-remote-${block}.json"
-        log "Wrote witness-local-${block}.json and witness-remote-${block}.json"
-        if [[ $n_extra -gt 0 ]]; then
-            log "  (local also has $n_extra extra state node(s))"
-        fi
-        ((errors++)) || true
-        log "---"
-    fi
-
-    rm -f "$local_file" "$remote_file"
-done
-
-total=$((latest - start + 1))
-if (( errors > 0 )); then
-    log "DONE: $errors block(s) had errors out of $total"
-    exit 1
-else
-    log "DONE: all $total block(s) matched"
-fi