diff --git a/core/blockchain.go b/core/blockchain.go
index b88b1bcb32..2b11084cb8 100644
--- a/core/blockchain.go
+++ b/core/blockchain.go
@@ -72,11 +72,10 @@ var (
 	accountReadTimer   = metrics.NewRegisteredResettingTimer("chain/account/reads", nil)
 	accountHashTimer   = metrics.NewRegisteredResettingTimer("chain/account/hashes", nil)
 	accountUpdateTimer = metrics.NewRegisteredResettingTimer("chain/account/updates", nil)
-	accountCommitTimer = metrics.NewRegisteredResettingTimer("chain/account/commits", nil)
+	hasherCommitTimer  = metrics.NewRegisteredResettingTimer("chain/trie/commits", nil)
 
 	storageReadTimer   = metrics.NewRegisteredResettingTimer("chain/storage/reads", nil)
 	storageUpdateTimer = metrics.NewRegisteredResettingTimer("chain/storage/updates", nil)
-	storageCommitTimer = metrics.NewRegisteredResettingTimer("chain/storage/commits", nil)
 	codeReadTimer      = metrics.NewRegisteredResettingTimer("chain/code/reads", nil)
 	codeReadBytesTimer = metrics.NewRegisteredResettingTimer("chain/code/readbytes", nil)
 
@@ -2112,12 +2111,16 @@ type ExecuteConfig struct {
 // it writes the block and associated state to database.
 func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash, block *types.Block, config ExecuteConfig) (result *blockProcessingResult, blockEndErr error) {
 	var (
-		err       error
-		startTime = time.Now()
-		statedb   *state.StateDB
-		interrupt atomic.Bool
-		sdb       = state.NewDatabase(bc.triedb, bc.codedb).WithSnapshot(bc.snaps)
+		err         error
+		startTime   = time.Now()
+		statedb     *state.StateDB
+		interrupt   atomic.Bool
+		sdb         = state.NewDatabase(bc.triedb, bc.codedb).WithSnapshot(bc.snaps)
+		makeWitness bool
 	)
+	if bc.chainConfig.IsByzantium(block.Number()) && (config.StatelessSelfValidation || config.MakeWitness) {
+		makeWitness = true
+	}
 	defer interrupt.Store(true) // terminate the prefetch at the end
 
 	if bc.cfg.NoPrefetch {
@@ -2126,6 +2129,10 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash,
 			return nil, err
 		}
 	} else {
+		// Enable trie node prewarming. The read-only state should also
+		// be prewarmed for constructing a comprehensive execution witness.
+		sdb = sdb.EnablePrefetch(makeWitness)
+
 		// If prefetching is enabled, run that against the current state to pre-cache
 		// transactions and probabilistically some of the account/storage trie nodes.
 		//
@@ -2171,20 +2178,16 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash,
 	// while processing transactions. Before Byzantium the prefetcher is mostly
 	// useless due to the intermediate root hashing after each transaction.
 	var witness *stateless.Witness
-	if bc.chainConfig.IsByzantium(block.Number()) {
+	if makeWitness {
 		// Generate witnesses either if we're self-testing, or if it's the
 		// only block being inserted. A bit crude, but witnesses are huge,
 		// so we refuse to make an entire chain of them.
-		if config.StatelessSelfValidation || config.MakeWitness {
-			witness, err = stateless.NewWitness(block.Header(), bc, config.EnableWitnessStats)
-			if err != nil {
-				return nil, err
-			}
+		witness, err = stateless.NewWitness(block.Header(), bc, config.EnableWitnessStats)
+		if err != nil {
+			return nil, err
 		}
-		statedb.StartPrefetcher("chain", witness)
-		defer statedb.StopPrefetcher()
+		statedb.TraceWitness(witness)
 	}
-
 	// Instrument the blockchain tracing
 	if config.EnableTracer {
 		if bc.logger != nil && bc.logger.OnBlockStart != nil {
@@ -2252,34 +2255,9 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash,
 	}
 
 	var (
-		xvtime   = time.Since(xvstart)
 		proctime = time.Since(startTime) // processing + validation + cross validation
-		stats    = &ExecuteStats{}
+		stats    = NewExecuteStats(statedb, ptime, vtime, time.Since(xvstart))
 	)
-	// Update the metrics touched during block processing and validation
-	stats.AccountReads = statedb.AccountReads     // Account reads are complete(in processing)
-	stats.StorageReads = statedb.StorageReads     // Storage reads are complete(in processing)
-	stats.AccountUpdates = statedb.AccountUpdates // Account updates are complete(in validation)
-	stats.StorageUpdates = statedb.StorageUpdates // Storage updates are complete(in validation)
-	stats.AccountHashes = statedb.AccountHashes   // Account hashes are complete(in validation)
-	stats.CodeReads = statedb.CodeReads
-
-	stats.AccountLoaded = statedb.AccountLoaded
-	//stats.AccountUpdated = statedb.AccountUpdated
-	stats.AccountDeleted = statedb.AccountDeleted
-	stats.StorageLoaded = statedb.StorageLoaded
-	stats.StorageUpdated = int(statedb.StorageUpdated.Load())
-	stats.StorageDeleted = int(statedb.StorageDeleted.Load())
-
-	stats.CodeLoaded = statedb.CodeLoaded
-	stats.CodeLoadBytes = statedb.CodeLoadBytes
-	//stats.CodeUpdated = statedb.CodeUpdated
-	//stats.CodeUpdateBytes = statedb.CodeUpdateBytes
-
-	stats.Execution = ptime - (statedb.AccountReads + statedb.StorageReads + statedb.CodeReads)          // The time spent on EVM processing
-	stats.Validation = vtime - (statedb.AccountHashes + statedb.AccountUpdates + statedb.StorageUpdates) // The time spent on block validation
-	stats.CrossValidation = xvtime                                                                       // The time spent on stateless cross validation
-
 	// Write the block to the chain and get the status.
 	var status WriteStatus
 	if config.WriteState {
@@ -2294,10 +2272,9 @@ func (bc *BlockChain) ProcessBlock(ctx context.Context, parentRoot common.Hash,
 			return nil, err
 		}
 		// Update the metrics touched during block commit
-		stats.AccountCommits = statedb.AccountCommits  // Account commits are complete, we can mark them
-		stats.StorageCommits = statedb.StorageCommits  // Storage commits are complete, we can mark them
+		stats.HasherCommit = statedb.HasherCommits     // Storage commits are complete, we can mark them
 		stats.DatabaseCommit = statedb.DatabaseCommits // Database commits are complete, we can mark them
-		stats.BlockWrite = time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.DatabaseCommits
+		stats.BlockWrite = time.Since(wstart) - statedb.HasherCommits - statedb.DatabaseCommits
 	}
 	// Report the collected witness statistics
 	if witness != nil {
diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go
index 3614702d1a..82e9305bdc 100644
--- a/core/blockchain_reader.go
+++ b/core/blockchain_reader.go
@@ -424,6 +424,11 @@ func (bc *BlockChain) StateAt(root common.Hash) (*state.StateDB, error) {
 	return state.New(root, state.NewDatabase(bc.triedb, bc.codedb).WithSnapshot(bc.snaps))
 }
 
+// StateWithPrefetching returns a new mutable state based on a particular point in time.
+func (bc *BlockChain) StateWithPrefetching(root common.Hash) (*state.StateDB, error) {
+	return state.New(root, state.NewDatabase(bc.triedb, bc.codedb).WithSnapshot(bc.snaps).EnablePrefetch(false))
+}
+
 // HistoricState returns a historic state specified by the given root.
 // Live states are not available and won't be served, please use `State`
 // or `StateAt` instead.
diff --git a/core/blockchain_stats.go b/core/blockchain_stats.go
index d753b0b700..4608151654 100644
--- a/core/blockchain_stats.go
+++ b/core/blockchain_stats.go
@@ -29,14 +29,30 @@ import (
 // ExecuteStats includes all the statistics of a block execution in details.
 type ExecuteStats struct {
 	// State read times
-	AccountReads   time.Duration // Time spent on the account reads
-	StorageReads   time.Duration // Time spent on the storage reads
+	AccountReads time.Duration // Time spent on the account reads
+	StorageReads time.Duration // Time spent on the storage reads
+	CodeReads    time.Duration // Time spent on the contract code read
+
+	// State hash times
 	AccountHashes  time.Duration // Time spent on the account trie hash
 	AccountUpdates time.Duration // Time spent on the account trie update
-	AccountCommits time.Duration // Time spent on the account trie commit
 	StorageUpdates time.Duration // Time spent on the storage trie update
-	StorageCommits time.Duration // Time spent on the storage trie commit
-	CodeReads      time.Duration // Time spent on the contract code read
+
+	// EVM execution time
+	Execution time.Duration // Time spent on the EVM execution
+
+	// Validation times
+	Validation      time.Duration // Time spent on the block validation
+	CrossValidation time.Duration // Optional, time spent on the block cross validation
+
+	// Commit times
+	HasherCommit   time.Duration // Time spent on trie commit
+	DatabaseCommit time.Duration // Time spent on database commit
+	BlockWrite     time.Duration // Time spent on block write
+
+	// Others
+	TotalTime     time.Duration // The total time spent on block execution
+	MgasPerSecond float64       // The million gas processed per second
 
 	AccountLoaded   int // Number of accounts loaded
 	AccountUpdated  int // Number of accounts updated
@@ -49,19 +65,40 @@ type ExecuteStats struct {
 	CodeUpdated     int // Number of contract code written (CREATE/CREATE2 + EIP-7702)
 	CodeUpdateBytes int // Total bytes of code written
 
-	Execution       time.Duration // Time spent on the EVM execution
-	Validation      time.Duration // Time spent on the block validation
-	CrossValidation time.Duration // Optional, time spent on the block cross validation
-	DatabaseCommit  time.Duration // Time spent on database commit
-	BlockWrite      time.Duration // Time spent on block write
-	TotalTime       time.Duration // The total time spent on block execution
-	MgasPerSecond   float64       // The million gas processed per second
-
 	// Cache hit rates
 	StateReadCacheStats     state.ReaderStats
 	StatePrefetchCacheStats state.ReaderStats
 }
 
+func NewExecuteStats(stateDB *state.StateDB, process time.Duration, validation time.Duration, crossValidation time.Duration) *ExecuteStats {
+	return &ExecuteStats{
+		// State read times
+		AccountReads: stateDB.AccountReads,
+		StorageReads: stateDB.StorageReads,
+		CodeReads:    stateDB.CodeReads,
+
+		// State hash times
+		AccountHashes:  stateDB.AccountHashes,
+		AccountUpdates: stateDB.AccountUpdates,
+		StorageUpdates: stateDB.StorageUpdates,
+
+		Execution:       process - stateDB.StateReadTime(),
+		Validation:      validation - stateDB.StateHashTime(),
+		CrossValidation: crossValidation,
+
+		AccountLoaded:   stateDB.AccountLoaded,
+		AccountUpdated:  stateDB.AccountUpdated,
+		AccountDeleted:  stateDB.AccountDeleted,
+		StorageLoaded:   stateDB.StorageLoaded,
+		StorageUpdated:  int(stateDB.StorageUpdated.Load()),
+		StorageDeleted:  int(stateDB.StorageDeleted.Load()),
+		CodeLoaded:      stateDB.CodeLoaded,
+		CodeLoadBytes:   stateDB.CodeLoadBytes,
+		CodeUpdated:     stateDB.CodeUpdated,
+		CodeUpdateBytes: stateDB.CodeUpdateBytes,
+	}
+}
+
 // reportMetrics uploads execution statistics to the metrics system.
 func (s *ExecuteStats) reportMetrics() {
 	if s.AccountLoaded != 0 {
@@ -80,8 +117,7 @@ func (s *ExecuteStats) reportMetrics() {
 	accountUpdateTimer.Update(s.AccountUpdates) // Account updates are complete(in validation)
 	storageUpdateTimer.Update(s.StorageUpdates) // Storage updates are complete(in validation)
 	accountHashTimer.Update(s.AccountHashes)    // Account hashes are complete(in validation)
-	accountCommitTimer.Update(s.AccountCommits) // Account commits are complete, we can mark them
-	storageCommitTimer.Update(s.StorageCommits) // Storage commits are complete, we can mark them
+	hasherCommitTimer.Update(s.HasherCommit)    // Trie commits are complete, we can mark them
 
 	blockExecutionTimer.Update(s.Execution)                 // The time spent on EVM processing
 	blockValidationTimer.Update(s.Validation)               // The time spent on block validation
@@ -206,7 +242,7 @@ func (s *ExecuteStats) logSlow(block *types.Block, slowBlockThreshold time.Durat
 			ExecutionMs: durationToMs(s.Execution),
 			StateReadMs: durationToMs(s.AccountReads + s.StorageReads + s.CodeReads),
 			StateHashMs: durationToMs(s.AccountHashes + s.AccountUpdates + s.StorageUpdates),
-			CommitMs:    durationToMs(max(s.AccountCommits, s.StorageCommits) + s.DatabaseCommit + s.BlockWrite),
+			CommitMs:    durationToMs(s.HasherCommit + s.DatabaseCommit + s.BlockWrite),
 			TotalMs:     durationToMs(s.TotalTime),
 		},
 		Throughput: slowBlockThru{
diff --git a/core/state/database.go b/core/state/database.go
index dfdae2fa06..a468dbf9ed 100644
--- a/core/state/database.go
+++ b/core/state/database.go
@@ -150,6 +150,9 @@ type CachingDB struct {
 	triedb *triedb.Database
 	codedb *CodeDB
 	snap   *snapshot.Tree
+
+	prefetch     bool
+	prefetchRead bool
 }
 
 // NewDatabase creates a state database with the provided data sources.
@@ -177,6 +180,13 @@ func (db *CachingDB) WithSnapshot(snapshot *snapshot.Tree) *CachingDB {
 	return db
 }
 
+// EnablePrefetch enables the hasher prefetching feature.
+func (db *CachingDB) EnablePrefetch(prefetchRead bool) *CachingDB {
+	db.prefetch = true
+	db.prefetchRead = prefetchRead
+	return db
+}
+
 // StateReader returns a state reader associated with the specified state root.
 func (db *CachingDB) StateReader(stateRoot common.Hash) (StateReader, error) {
 	var readers []StateReader
@@ -224,7 +234,7 @@ func (db *CachingDB) Reader(stateRoot common.Hash) (Reader, error) {
 // Hasher implements Database, returning a hasher associated with the specified
 // state root.
 func (db *CachingDB) Hasher(stateRoot common.Hash) (Hasher, error) {
-	return newMerkleHasher(stateRoot, db.triedb, true, true)
+	return newMerkleHasher(stateRoot, db.triedb, db.prefetch, db.prefetchRead)
 }
 
 // ReadersWithCacheStats creates a pair of state readers that share the same
diff --git a/core/state/database_hasher.go b/core/state/database_hasher.go
index 9610191b91..8bc5277a08 100644
--- a/core/state/database_hasher.go
+++ b/core/state/database_hasher.go
@@ -93,9 +93,9 @@ type Prefetcher interface {
 // WitnessCollector is an optional extension implemented by hashers that can
 // construct a state witness for the most recent committed state transition.
 type WitnessCollector interface {
-	// Witness returns the state witness corresponding to the most recent
+	// CollectWitness returns the state witness corresponding to the most recent
 	// committed state transition.
-	Witness() (*stateless.Witness, error)
+	CollectWitness(*stateless.Witness)
 }
 
 // Prover is an optional extension implemented by hashers that can construct
diff --git a/core/state/database_hasher_merkle.go b/core/state/database_hasher_merkle.go
index 28aaa513f3..237f7f1f07 100644
--- a/core/state/database_hasher_merkle.go
+++ b/core/state/database_hasher_merkle.go
@@ -21,12 +21,14 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/stateless"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/trie"
 	"github.com/ethereum/go-ethereum/trie/trienode"
 	"github.com/ethereum/go-ethereum/triedb"
+	"golang.org/x/sync/errgroup"
 )
 
 // wrapTrie pairs a StateTrie with an optional background prefetcher that
@@ -36,6 +38,7 @@ type wrapTrie struct {
 	prefetcher *prefetcher
 }
 
+// newWrapTrie creates a merkle trie with the optional prefetcher enabled.
 func newWrapTrie(id *trie.ID, db *triedb.Database, prefetch bool, prefetchRead bool) (*wrapTrie, error) {
 	t, err := trie.NewStateTrie(id, db)
 	if err != nil {
@@ -59,7 +62,7 @@ func (tr *wrapTrie) term() {
 	tr.prefetcher = nil
 }
 
-// The methods below shadow the embedded StateTrie so that any direct trie
+// The methods below shadow the embedded trie.StateTrie so that any direct trie
 // access auto-terminates the prefetcher first. This makes data-race freedom
 // structural: callers never need to remember to call term() manually.
 
@@ -98,9 +101,9 @@ func (tr *wrapTrie) Prove(key []byte, proofDb ethdb.KeyValueWriter) error {
 	return tr.StateTrie.Prove(key, proofDb)
 }
 
-func (tr *wrapTrie) copy() *wrapTrie {
+func (tr *wrapTrie) Witness() map[string][]byte {
 	tr.term()
-	return &wrapTrie{StateTrie: tr.StateTrie.Copy()}
+	return tr.StateTrie.Witness()
 }
 
 func (tr *wrapTrie) prefetchAccounts(addresses []common.Address, read bool) {
@@ -117,22 +120,31 @@ func (tr *wrapTrie) prefetchStorage(addr common.Address, keys []common.Hash, rea
 	tr.prefetcher.scheduleSlots(addr, keys, read)
 }
 
-// rootReader wraps the account trie for loading the storage root. It is
-// essential to use an independent trie to prevent potential data races
-// with the optional prefetcher.
-type rootReader struct {
+// copy returns a deep-copied state trie. Notably the prefetcher is deliberately
+// not copied, as it only belongs to the original one.
+func (tr *wrapTrie) copy() *wrapTrie {
+	tr.term()
+	return &wrapTrie{StateTrie: tr.StateTrie.Copy()}
+}
+
+// storageRootReader wraps the account trie for loading the storage root. It is
+// essential to use an independent trie to prevent potential data races with
+// the optional prefetcher.
+//
+// TODO(rjl493456442) use the flat state for better read efficiency.
+type storageRootReader struct {
 	tr *trie.StateTrie
 }
 
-func newRootReader(root common.Hash, db *triedb.Database) (*rootReader, error) {
+func newStorageRootReader(root common.Hash, db *triedb.Database) (*storageRootReader, error) {
 	t, err := trie.NewStateTrie(trie.StateTrieID(root), db)
 	if err != nil {
 		return nil, err
 	}
-	return &rootReader{tr: t}, nil
+	return &storageRootReader{tr: t}, nil
 }
 
-func (r *rootReader) readStorageRoot(address common.Address) (common.Hash, error) {
+func (r *storageRootReader) read(address common.Address) (common.Hash, error) {
 	acct, err := r.tr.GetAccount(address)
 	if err != nil {
 		return common.Hash{}, err
@@ -143,8 +155,8 @@ func (r *rootReader) readStorageRoot(address common.Address) (common.Hash, error
 	return acct.Root, nil
 }
 
-func (r *rootReader) copy() *rootReader {
-	return &rootReader{tr: r.tr.Copy()}
+func (r *storageRootReader) copy() *storageRootReader {
+	return &storageRootReader{tr: r.tr.Copy()}
 }
 
 // merkleHasher is a Hasher implementation backed by the traditional two-layer
@@ -152,7 +164,7 @@ func (r *rootReader) copy() *rootReader {
 type merkleHasher struct {
 	db           *triedb.Database
 	root         common.Hash
-	reader       *rootReader
+	reader       *storageRootReader
 	prefetch     bool
 	prefetchRead bool
 
@@ -160,7 +172,7 @@ type merkleHasher struct {
 	storageTries map[common.Address]*wrapTrie
 
 	// deletedTries preserves storage tries of accounts that were deleted
-	// during the block. Keyed by address; only the first deletion per
+	// during the block keyed by address. Only the first deletion per
 	// address is recorded (the pre-block incarnation).
 	deletedTries map[common.Address]*wrapTrie
 
@@ -169,7 +181,8 @@ type merkleHasher struct {
 	// UpdateStorage or set to EmptyRootHash on deletion.
 	storageRoots map[common.Address]Hashes
 
-	storageLock sync.Mutex // guards storage trie fields
+	// Lock guards storage trie fields
+	storageLock sync.Mutex
 }
 
 func newMerkleHasher(root common.Hash, db *triedb.Database, prefetch bool, prefetchRead bool) (*merkleHasher, error) {
@@ -177,7 +190,7 @@ func newMerkleHasher(root common.Hash, db *triedb.Database, prefetch bool, prefe
 	if err != nil {
 		return nil, err
 	}
-	r, err := newRootReader(root, db)
+	r, err := newStorageRootReader(root, db)
 	if err != nil {
 		return nil, err
 	}
@@ -201,11 +214,14 @@ func (h *merkleHasher) storageRoot(addr common.Address) (common.Hash, error) {
 	if hashes, ok := h.storageRoots[addr]; ok {
 		return hashes.Hash, nil
 	}
-	root, err := h.reader.readStorageRoot(addr)
+	root, err := h.reader.read(addr)
 	if err != nil {
 		return common.Hash{}, err
 	}
-	h.storageRoots[addr] = Hashes{Prev: root, Hash: root}
+	h.storageRoots[addr] = Hashes{
+		Prev: root,
+		Hash: root,
+	}
 	return root, nil
 }
 
@@ -223,6 +239,7 @@ func (h *merkleHasher) openStorageTrie(address common.Address, prefetch bool) (*
 		return nil, err
 	}
 	id := trie.StorageTrieID(h.root, crypto.Keccak256Hash(address.Bytes()), root)
+
 	tr, err := newWrapTrie(id, h.db, h.prefetch && prefetch, h.prefetchRead)
 	if err != nil {
 		return nil, err
@@ -231,8 +248,9 @@ func (h *merkleHasher) openStorageTrie(address common.Address, prefetch bool) (*
 	return tr, nil
 }
 
+// deleteAccount removes the account specified by the address from the state.
 func (h *merkleHasher) deleteAccount(addr common.Address) error {
-	// Deletion: capture the original storage root before modifying the trie.
+	// Capture the original storage root before modifying the trie.
 	_, err := h.storageRoot(addr)
 	if err != nil {
 		return err
@@ -251,6 +269,7 @@ func (h *merkleHasher) deleteAccount(addr common.Address) error {
 	return h.acctTrie.DeleteAccount(addr)
 }
 
+// update writes the account specified by the address into the state.
 func (h *merkleHasher) updateAccount(addr common.Address, account AccountMut) error {
 	root, err := h.storageRoot(addr)
 	if err != nil {
@@ -265,10 +284,12 @@ func (h *merkleHasher) updateAccount(addr common.Address, account AccountMut) er
 	return h.acctTrie.UpdateAccount(addr, data, 0)
 }
 
-// UpdateAccount implements Hasher.
+// UpdateAccount implements Hasher, writing a list of account mutations
+// into the state. The assumption is held all the storage changes have
+// already been written beforehand.
 func (h *merkleHasher) UpdateAccount(addresses []common.Address, accounts []AccountMut) error {
+	var err error
 	for i, addr := range addresses {
-		var err error
 		if accounts[i].Account == nil {
 			err = h.deleteAccount(addr)
 		} else {
@@ -281,7 +302,9 @@ func (h *merkleHasher) UpdateAccount(addresses []common.Address, accounts []Acco
 	return nil
 }
 
-// UpdateStorage implements Hasher.
+// UpdateStorage implements Hasher, writing a list of storage slot mutations
+// into the state. This function must be invoked first before writing the
+// associated account metadata into the state.
 func (h *merkleHasher) UpdateStorage(address common.Address, keys []common.Hash, values []common.Hash) error {
 	tr, err := h.openStorageTrie(address, false)
 	if err != nil {
@@ -289,18 +312,19 @@ func (h *merkleHasher) UpdateStorage(address common.Address, keys []common.Hash,
 	}
 	for i, key := range keys {
 		if values[i] == (common.Hash{}) {
-			if err := tr.DeleteStorage(address, key[:]); err != nil {
-				return err
-			}
+			err = tr.DeleteStorage(address, key[:])
 		} else {
-			if err := tr.UpdateStorage(address, key[:], common.TrimLeftZeroes(values[i][:])); err != nil {
-				return err
-			}
+			err = tr.UpdateStorage(address, key[:], common.TrimLeftZeroes(values[i][:]))
+		}
+		if err != nil {
+			return err
 		}
 	}
 	// Hash outside the lock to allow full parallelism across accounts.
 	hash := tr.Hash()
 
+	// Write back the storage root back for reflecting the most recent
+	// changes.
 	h.storageLock.Lock()
 	h.storageRoots[address] = Hashes{
 		Prev: h.storageRoots[address].Prev,
@@ -310,50 +334,64 @@ func (h *merkleHasher) UpdateStorage(address common.Address, keys []common.Hash,
 	return nil
 }
 
+// Hash implements Hasher, computing the state root hash without committing.
 func (h *merkleHasher) Hash() common.Hash {
 	return h.acctTrie.Hash()
 }
 
-// Close terminates all prefetcher goroutines. Safe to call multiple times.
-func (h *merkleHasher) Close() {
-	h.acctTrie.term()
-	for _, tr := range h.storageTries {
-		tr.term()
-	}
-	for _, tr := range h.deletedTries {
-		tr.term()
-	}
-}
-
+// Commit implements Hasher, finalizing all pending changes and returning
+// the resulting state root hash, along with the set of dirty trie nodes
+// generated by the updates.
 func (h *merkleHasher) Commit() (common.Hash, *trienode.MergedNodeSet, map[common.Address]Hashes, error) {
 	// Explicitly terminate all resolved tries. Some of them may not be
 	// terminated due to read-only prefetching. This is essential to
 	// prevent goroutine leaks.
 	h.Close()
 
-	nodes := trienode.NewMergedNodeSet()
+	var (
+		eg   errgroup.Group
+		root common.Hash
+
+		lock  sync.Mutex
+		nodes = trienode.NewMergedNodeSet()
+		merge = func(set *trienode.NodeSet) error {
+			lock.Lock()
+			defer lock.Unlock()
+
+			return nodes.Merge(set)
+		}
+	)
+	eg.Go(func() error {
+		r, set := h.acctTrie.Commit(true)
+		root = r
+		if set == nil {
+			return nil
+		}
+		return merge(set)
+	})
 	for _, tr := range h.storageTries {
-		if _, set := tr.Commit(false); set != nil {
-			if err := nodes.Merge(set); err != nil {
-				return common.Hash{}, nil, nil, err
+		eg.Go(func() error {
+			_, set := tr.Commit(false)
+			if set == nil {
+				return nil
 			}
-		}
+			return merge(set)
+		})
 	}
-	root, set := h.acctTrie.Commit(true)
-	if set != nil {
-		if err := nodes.Merge(set); err != nil {
-			return common.Hash{}, nil, nil, err
-		}
+	if err := eg.Wait(); err != nil {
+		return common.Hash{}, nil, nil, err
 	}
 	return root, nodes, h.storageRoots, nil
 }
 
+// Copy implements Hasher, returning a deep-copied hasher instance.
 func (h *merkleHasher) Copy() Hasher {
 	cpy := &merkleHasher{
 		db:           h.db,
 		root:         h.root,
 		reader:       h.reader.copy(),
 		prefetch:     false,
+		prefetchRead: false,
 		acctTrie:     h.acctTrie.copy(),
 		storageTries: make(map[common.Address]*wrapTrie, len(h.storageTries)),
 		deletedTries: make(map[common.Address]*wrapTrie, len(h.deletedTries)),
@@ -368,12 +406,24 @@ func (h *merkleHasher) Copy() Hasher {
 	return cpy
 }
 
-// ProveAccount implements Prover.
+// Close terminates all prefetcher goroutines. Safe to call multiple times.
+func (h *merkleHasher) Close() {
+	h.acctTrie.term()
+	for _, tr := range h.storageTries {
+		tr.term()
+	}
+	for _, tr := range h.deletedTries {
+		tr.term()
+	}
+}
+
+// ProveAccount implements Prover, constructing a proof for the given account.
 func (h *merkleHasher) ProveAccount(addr common.Address, proofDb ethdb.KeyValueWriter) error {
 	return h.acctTrie.Prove(crypto.Keccak256(addr.Bytes()), proofDb)
 }
 
-// ProveStorage implements Prover.
+// ProveStorage implements Prover, constructing a proof for the given storage
+// slot of the specified account.
 func (h *merkleHasher) ProveStorage(addr common.Address, key common.Hash, proofDb ethdb.KeyValueWriter) error {
 	tr, err := h.openStorageTrie(addr, false)
 	if err != nil {
@@ -382,6 +432,19 @@ func (h *merkleHasher) ProveStorage(addr common.Address, key common.Hash, proofD
 	return tr.Prove(crypto.Keccak256(key.Bytes()), proofDb)
 }
 
+// CollectWitness implements WitnessCollector. It aggregates all trie nodes
+// accessed (both read and write) across the account trie, all active storage
+// tries and deleted storage tries into a single state witness.
+func (h *merkleHasher) CollectWitness(witness *stateless.Witness) {
+	witness.AddState(h.acctTrie.Witness(), common.Hash{})
+	for addr, tr := range h.storageTries {
+		witness.AddState(tr.Witness(), crypto.Keccak256Hash(addr.Bytes()))
+	}
+	for addr, tr := range h.deletedTries {
+		witness.AddState(tr.Witness(), crypto.Keccak256Hash(addr.Bytes()))
+	}
+}
+
 // PrefetchAccount implements Prefetcher, preloading the nodes of specific accounts.
 func (h *merkleHasher) PrefetchAccount(addresses []common.Address, read bool) {
 	if !h.prefetch {
diff --git a/core/state/database_hasher_merkle_test.go b/core/state/database_hasher_merkle_test.go
index 559420e42c..e60fd18bbb 100644
--- a/core/state/database_hasher_merkle_test.go
+++ b/core/state/database_hasher_merkle_test.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/core/stateless"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/triedb"
 	"github.com/holiman/uint256"
@@ -539,3 +540,90 @@ func TestMerkleHasherCopy(t *testing.T) {
 		t.Fatal("original root changed after mutating copy")
 	}
 }
+
+// proofNodes collects the raw RLP-encoded trie nodes written by Prove calls.
+type proofNodes struct{ nodes [][]byte }
+
+func (p *proofNodes) Put(key []byte, value []byte) error {
+	p.nodes = append(p.nodes, common.CopyBytes(value))
+	return nil
+}
+func (p *proofNodes) Delete([]byte) error { return nil }
+
+// TestMerkleHasherWitness verifies that the witness returned by Witness()
+// contains every trie node on the Merkle proof path for each accessed account
+// and storage slot, including nodes from deleted storage tries.
+func TestMerkleHasherWitness(t *testing.T) {
+	h := makeBaseState(t, hasherTestConfig{"prefetchAll", true, true})
+
+	// Mutate addr1 storage, then delete and recreate with different
+	// storage so that both deletedTries and storageTries are populated.
+	h.PrefetchStorage(hasherAddr1, []common.Hash{hasherSlot1}, false)
+	if err := h.UpdateStorage(hasherAddr1, []common.Hash{hasherSlot1}, []common.Hash{hasherVal2}); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.UpdateAccount([]common.Address{hasherAddr1}, []AccountMut{hasherDeleteAccount()}); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.UpdateStorage(hasherAddr1, []common.Hash{hasherSlot3}, []common.Hash{hasherVal3}); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.UpdateAccount(
+		[]common.Address{hasherAddr1, hasherAddr2},
+		[]AccountMut{hasherAccount(10, 500), hasherAccount(2, 300)},
+	); err != nil {
+		t.Fatal(err)
+	}
+	witness := &stateless.Witness{
+		Codes: make(map[string]struct{}),
+		State: make(map[string]struct{}),
+	}
+	h.CollectWitness(witness)
+
+	if len(witness.State) == 0 {
+		t.Fatal("witness should contain trie nodes")
+	}
+	// Open a separate prover from the same pre-state root. Proofs
+	// generated here traverse the same trie paths that the mutating
+	// hasher loaded, so every proof node must be in the witness.
+	prover, err := newMerkleHasher(h.root, h.db, false, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer prover.Close()
+
+	// Collect all expected proof nodes into a single set. The union of
+	// account proofs (addr1, addr2) and storage proofs (addr1/slot1)
+	// should exactly equal witness.State — no missing, no extra.
+	expected := make(map[string]struct{})
+
+	for _, addr := range []common.Address{hasherAddr1, hasherAddr2} {
+		pn := &proofNodes{}
+		if err := prover.ProveAccount(addr, pn); err != nil {
+			t.Fatal(err)
+		}
+		for _, node := range pn.nodes {
+			expected[string(node)] = struct{}{}
+		}
+	}
+	// Storage proof for addr1/slot1 (accessed before deletion).
+	// Slot2 was in the base state but never read or written during the
+	// block, so its leaf node is correctly absent from the witness.
+	pn := &proofNodes{}
+	if err := prover.ProveStorage(hasherAddr1, hasherSlot1, pn); err != nil {
+		t.Fatal(err)
+	}
+	for _, node := range pn.nodes {
+		expected[string(node)] = struct{}{}
+	}
+	// Every expected proof node must be in the witness.
+	for node := range expected {
+		if _, ok := witness.State[node]; !ok {
+			t.Fatal("proof node missing from witness")
+		}
+	}
+	// The witness must not contain any extra nodes beyond the proofs.
+	if len(witness.State) != len(expected) {
+		t.Fatalf("witness has %d nodes, expected %d (extra junk present)", len(witness.State), len(expected))
+	}
+}
diff --git a/core/state/metrics.go b/core/state/metrics.go
index dd4b2e9838..be5f9303af 100644
--- a/core/state/metrics.go
+++ b/core/state/metrics.go
@@ -19,14 +19,14 @@ package state
 import "github.com/ethereum/go-ethereum/metrics"
 
 var (
-	accountReadMeters        = metrics.NewRegisteredMeter("state/read/account", nil)
-	storageReadMeters        = metrics.NewRegisteredMeter("state/read/storage", nil)
-	accountUpdatedMeter      = metrics.NewRegisteredMeter("state/update/account", nil)
-	storageUpdatedMeter      = metrics.NewRegisteredMeter("state/update/storage", nil)
-	accountDeletedMeter      = metrics.NewRegisteredMeter("state/delete/account", nil)
-	storageDeletedMeter      = metrics.NewRegisteredMeter("state/delete/storage", nil)
-	accountTrieUpdatedMeter  = metrics.NewRegisteredMeter("state/update/accountnodes", nil)
-	storageTriesUpdatedMeter = metrics.NewRegisteredMeter("state/update/storagenodes", nil)
-	accountTrieDeletedMeter  = metrics.NewRegisteredMeter("state/delete/accountnodes", nil)
-	storageTriesDeletedMeter = metrics.NewRegisteredMeter("state/delete/storagenodes", nil)
+	accountReadMeters   = metrics.NewRegisteredMeter("state/read/account", nil)
+	storageReadMeters   = metrics.NewRegisteredMeter("state/read/storage", nil)
+	accountUpdatedMeter = metrics.NewRegisteredMeter("state/update/account", nil)
+	storageUpdatedMeter = metrics.NewRegisteredMeter("state/update/storage", nil)
+	accountDeletedMeter = metrics.NewRegisteredMeter("state/delete/account", nil)
+	storageDeletedMeter = metrics.NewRegisteredMeter("state/delete/storage", nil)
+	//accountTrieUpdatedMeter  = metrics.NewRegisteredMeter("state/update/accountnodes", nil)
+	//storageTriesUpdatedMeter = metrics.NewRegisteredMeter("state/update/storagenodes", nil)
+	//accountTrieDeletedMeter  = metrics.NewRegisteredMeter("state/delete/accountnodes", nil)
+	//storageTriesDeletedMeter = metrics.NewRegisteredMeter("state/delete/storagenodes", nil)
 )
diff --git a/core/state/reader.go b/core/state/reader.go
index 9c144e71e2..120253ff0c 100644
--- a/core/state/reader.go
+++ b/core/state/reader.go
@@ -63,7 +63,6 @@ type Account struct {
 }
 
 // newEmptyAccount returns an empty account.
-// nolint:unused
 func newEmptyAccount() *Account {
 	return &Account{
 		Balance:  uint256.NewInt(0),
diff --git a/core/state/state_mut.go b/core/state/state_mut.go
index c62b152633..5e7f3c359e 100644
--- a/core/state/state_mut.go
+++ b/core/state/state_mut.go
@@ -38,7 +38,11 @@ type mutation struct {
 }
 
 func (m *mutation) copy() *mutation {
-	return &mutation{typ: m.typ, applied: m.applied, precedingDelete: m.precedingDelete}
+	return &mutation{
+		typ:             m.typ,
+		applied:         m.applied,
+		precedingDelete: m.precedingDelete,
+	}
 }
 
 func (m *mutation) isDelete() bool {
diff --git a/core/state/state_object.go b/core/state/state_object.go
index 491a2752ec..86f43f1b2c 100644
--- a/core/state/state_object.go
+++ b/core/state/state_object.go
@@ -171,14 +171,13 @@ func (s *stateObject) GetCommittedState(key common.Hash) common.Hash {
 		s.originStorage[key] = common.Hash{} // track the empty slot as origin value
 		return common.Hash{}
 	}
-	s.db.StorageLoaded++
-
 	start := time.Now()
 	value, err := s.db.reader.Storage(s.address, key)
 	if err != nil {
 		s.db.setError(err)
 		return common.Hash{}
 	}
+	s.db.StorageLoaded++
 	s.db.StorageReads += time.Since(start)
 
 	s.originStorage[key] = value
@@ -267,8 +266,10 @@ func (s *stateObject) updateTrie() error {
 		return nil
 	}
 	var (
-		keys = make([]common.Hash, 0, len(s.uncommittedStorage))
-		vals = make([]common.Hash, 0, len(s.uncommittedStorage))
+		updates int64
+		deletes int64
+		keys    = make([]common.Hash, 0, len(s.uncommittedStorage))
+		vals    = make([]common.Hash, 0, len(s.uncommittedStorage))
 	)
 	for key, origin := range s.uncommittedStorage {
 		// Skip noop changes, persist actual changes
@@ -281,10 +282,17 @@ func (s *stateObject) updateTrie() error {
 			log.Error("Storage slot is not found in pending area", "address", s.address, "slot", key)
 			continue
 		}
+		if value == (common.Hash{}) {
+			deletes += 1
+		} else {
+			updates += 1
+		}
 		keys = append(keys, key)
 		vals = append(vals, value)
 	}
 	s.uncommittedStorage = make(Storage) // empties the commit markers
+	s.db.StorageUpdated.Add(updates)
+	s.db.StorageDeleted.Add(deletes)
 
 	return s.db.hasher.UpdateStorage(s.address, keys, vals)
 }
@@ -337,7 +345,7 @@ func (s *stateObject) commit() (*accountUpdate, error) {
 	// commit the contract code if it's modified
 	if s.dirtyCode {
 		s.dirtyCode = false // reset the dirty flag
-	
+
 		op.code = &contractCode{
 			hash: common.BytesToHash(s.CodeHash()),
 			blob: s.code,
diff --git a/core/state/statedb.go b/core/state/statedb.go
index fdf86528fd..6f428d0715 100644
--- a/core/state/statedb.go
+++ b/core/state/statedb.go
@@ -23,7 +23,6 @@ import (
 	"maps"
 	"slices"
 	"sort"
-	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -111,30 +110,11 @@ type StateDB struct {
 	// Snapshot and RevertToSnapshot.
 	journal *journal
 
+	// State witness if cross validation is needed
+	witness *stateless.Witness
+
 	// Measurements gathered during execution for debugging purposes
-	AccountReads   time.Duration
-	AccountHashes  time.Duration
-	AccountUpdates time.Duration
-	AccountCommits time.Duration
-
-	StorageReads    time.Duration
-	StorageUpdates  time.Duration
-	StorageCommits  time.Duration
-	DatabaseCommits time.Duration
-	CodeReads       time.Duration
-
-	AccountLoaded  int          // Number of accounts retrieved from the database during the state transition
-	AccountDeleted int          // Number of accounts deleted during the state transition
-	StorageLoaded  int          // Number of storage slots retrieved from the database during the state transition
-	StorageUpdated atomic.Int64 // Number of storage slots updated during the state transition
-	StorageDeleted atomic.Int64 // Number of storage slots deleted during the state transition
-
-	// CodeLoadBytes is the total number of bytes read from contract code.
-	// This value may be smaller than the actual number of bytes read, since
-	// some APIs (e.g. CodeSize) may load the entire code from either the
-	// cache or the database when the size is not available in the cache.
-	CodeLoaded    int // Number of contract code loaded during the state transition
-	CodeLoadBytes int // Total bytes of resolved code
+	Stats
 }
 
 // New creates a new state from a given trie.
@@ -173,16 +153,11 @@ func NewWithReader(root common.Hash, db Database, reader Reader) (*StateDB, erro
 	return sdb, nil
 }
 
-// StartPrefetcher initializes a new trie prefetcher to pull in nodes from the
-// state trie concurrently while the state is mutated so that when we reach the
-// commit phase, most of the needed data is already hot.
-func (s *StateDB) StartPrefetcher(namespace string, witness *stateless.Witness) {
+// TraceWitness enables execution witness gathering.
+func (s *StateDB) TraceWitness(witness *stateless.Witness) {
+	s.witness = witness
 }
 
-// StopPrefetcher terminates a running prefetcher and reports any leftover stats
-// from the gathered metrics.
-func (s *StateDB) StopPrefetcher() {}
-
 // setError remembers the first non-nil error it is called with.
 func (s *StateDB) setError(err error) {
 	if s.dbErr == nil {
@@ -206,7 +181,7 @@ func (s *StateDB) AddLog(log *types.Log) {
 }
 
 // GetLogs returns the logs matching the specified transaction hash, and annotates
-// them with the given blockNumber and blockHash.
+// them with the given block attributes.
 func (s *StateDB) GetLogs(hash common.Hash, blockNumber uint64, blockHash common.Hash, blockTime uint64) []*types.Log {
 	logs := s.logs[hash]
 	for _, l := range logs {
@@ -217,6 +192,7 @@ func (s *StateDB) GetLogs(hash common.Hash, blockNumber uint64, blockHash common
 	return logs
 }
 
+// Logs returns the un-annotated logs in order.
 func (s *StateDB) Logs() []*types.Log {
 	logs := make([]*types.Log, 0, s.logSize)
 	for _, lgs := range s.logs {
@@ -296,6 +272,9 @@ func (s *StateDB) TxIndex() int {
 func (s *StateDB) GetCode(addr common.Address) []byte {
 	stateObject := s.getStateObject(addr)
 	if stateObject != nil {
+		if s.witness != nil {
+			s.witness.AddCode(stateObject.Code())
+		}
 		return stateObject.Code()
 	}
 	return nil
@@ -304,6 +283,9 @@ func (s *StateDB) GetCode(addr common.Address) []byte {
 func (s *StateDB) GetCodeSize(addr common.Address) int {
 	stateObject := s.getStateObject(addr)
 	if stateObject != nil {
+		if s.witness != nil {
+			s.witness.AddCode(stateObject.Code())
+		}
 		return stateObject.CodeSize()
 	}
 	return 0
@@ -502,14 +484,13 @@ func (s *StateDB) getStateObject(addr common.Address) *stateObject {
 	if _, ok := s.stateObjectsDestruct[addr]; ok {
 		return nil
 	}
-	s.AccountLoaded++
-
 	start := time.Now()
 	acct, err := s.reader.Account(addr)
 	if err != nil {
 		s.setError(fmt.Errorf("getStateObject (%x) error: %w", addr.Bytes(), err))
 		return nil
 	}
+	s.AccountLoaded++
 	s.AccountReads += time.Since(start)
 
 	// Short circuit if the account is not found
@@ -611,6 +592,9 @@ func (s *StateDB) Copy() *StateDB {
 		transientStorage: s.transientStorage.Copy(),
 		journal:          s.journal.copy(),
 	}
+	if s.witness != nil {
+		state.witness = s.witness.Copy()
+	}
 	if s.accessEvents != nil {
 		state.accessEvents = s.accessEvents.Copy()
 	}
@@ -756,6 +740,7 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
 			continue
 		}
 		op.precedingDelete = false
+
 		delAddrs = append(delAddrs, addr)
 		delAccts = append(delAccts, AccountMut{Account: nil})
 	}
@@ -764,6 +749,7 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
 			s.setError(err)
 			return common.Hash{}
 		}
+		s.AccountDeleted += len(delAddrs)
 	}
 	s.AccountUpdates += time.Since(start)
 
@@ -797,14 +783,20 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
 
 		if op.isDelete() {
 			accounts = append(accounts, AccountMut{Account: nil})
+			s.AccountDeleted += 1
 			continue
 		}
 		obj := s.stateObjects[addr]
 		mut := AccountMut{Account: &obj.data}
 		if obj.dirtyCode {
 			mut.Code = &CodeMut{Code: obj.code}
+
+			// Count code writes post-Finalise so reverted CREATEs are excluded.
+			s.CodeUpdated += 1
+			s.CodeUpdateBytes += len(obj.code)
 		}
 		accounts = append(accounts, mut)
+		s.AccountUpdated += 1
 	}
 	if err := s.hasher.UpdateAccount(addresses, accounts); err != nil {
 		s.setError(err)
@@ -932,8 +924,7 @@ func (s *StateDB) handleDestruction(noStorageWiping bool) (map[common.Hash]*acco
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to delete storage, err: %w", err)
 		}
-		op.storages = storages
-		op.storagesOrigin = storagesOrigin
+		op.storages, op.storagesOrigin = storages, storagesOrigin
 
 		// Aggregate the associated trie node changes.
 		if err := nodes.Merge(set); err != nil {
@@ -957,15 +948,6 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
 	if s.dbErr != nil {
 		return nil, fmt.Errorf("commit aborted due to database error: %v", s.dbErr)
 	}
-	// Commit objects to the trie, measuring the elapsed time
-	var (
-		accountTrieNodesUpdated int
-		accountTrieNodesDeleted int
-		storageTrieNodesUpdated int
-		storageTrieNodesDeleted int
-
-		updates = make(map[common.Hash]*accountUpdate, len(s.mutations)) // aggregated account updates
-	)
 	// Given that some accounts could be destroyed and then recreated within
 	// the same block, account deletions must be processed first. This ensures
 	// that the storage trie nodes deleted during destruction and recreated
@@ -974,6 +956,8 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
 	if err != nil {
 		return nil, err
 	}
+	// Aggregated account updates
+	updates := make(map[common.Hash]*accountUpdate, len(s.mutations))
 	for addr, op := range s.mutations {
 		if op.isDelete() {
 			continue
@@ -992,29 +976,16 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
 	// Handle all state updates afterwards, concurrently to one another to shave
 	// off some milliseconds from the commit operation. Also accumulate the code
 	// writes to run in parallel with the computations.
+	start := time.Now()
 	root, set, secondaryHashes, err := s.hasher.Commit()
 	if err != nil {
 		return nil, err
 	}
+	s.HasherCommits = time.Since(start)
+
 	if err := nodes.MergeSet(set); err != nil {
 		return nil, err
 	}
-	accountReadMeters.Mark(int64(s.AccountLoaded))
-	storageReadMeters.Mark(int64(s.StorageLoaded))
-	storageUpdatedMeter.Mark(s.StorageUpdated.Load())
-	accountDeletedMeter.Mark(int64(s.AccountDeleted))
-	storageDeletedMeter.Mark(s.StorageDeleted.Load())
-	accountTrieUpdatedMeter.Mark(int64(accountTrieNodesUpdated))
-	accountTrieDeletedMeter.Mark(int64(accountTrieNodesDeleted))
-	storageTriesUpdatedMeter.Mark(int64(storageTrieNodesUpdated))
-	storageTriesDeletedMeter.Mark(int64(storageTrieNodesDeleted))
-
-	// Clear the metric markers
-	s.AccountLoaded, s.AccountDeleted = 0, 0
-	s.StorageLoaded = 0
-	s.StorageUpdated.Store(0)
-	s.StorageDeleted.Store(0)
-
 	// Clear all internal flags and update state root at the end.
 	s.mutations = make(map[common.Address]*mutation)
 	s.stateObjectsDestruct = make(map[common.Address]*stateObject)
@@ -1022,6 +993,12 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
 	origin := s.originalRoot
 	s.originalRoot = root
 
+	if s.witness != nil {
+		builder, ok := s.hasher.(WitnessCollector)
+		if ok {
+			builder.CollectWitness(s.witness)
+		}
+	}
 	return newStateUpdate(noStorageWiping, origin, root, blockNumber, deletes, updates, nodes, secondaryHashes), nil
 }
 
@@ -1160,7 +1137,7 @@ func (s *StateDB) SlotInAccessList(addr common.Address, slot common.Hash) (addre
 
 // Witness retrieves the current state witness being collected.
 func (s *StateDB) Witness() *stateless.Witness {
-	return nil
+	return s.witness
 }
 
 func (s *StateDB) AccessEvents() *AccessEvents {
diff --git a/core/state/statedb_stats.go b/core/state/statedb_stats.go
new file mode 100644
index 0000000000..c69a157206
--- /dev/null
+++ b/core/state/statedb_stats.go
@@ -0,0 +1,61 @@
+// Copyright 2026 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package state
+
+import (
+	"sync/atomic"
+	"time"
+)
+
+// Stats contains all measurements gathered during state execution for
+// debugging and metrics purposes.
+type Stats struct {
+	AccountReads    time.Duration // Account read time
+	StorageReads    time.Duration // Storage read time
+	CodeReads       time.Duration // Code read time
+	AccountHashes   time.Duration // Account trie hash time
+	AccountUpdates  time.Duration // Account trie update time
+	StorageUpdates  time.Duration // Storage trie update and hash time
+	HasherCommits   time.Duration // Trie commit time
+	DatabaseCommits time.Duration // Database commit time
+
+	AccountLoaded  int          // Number of accounts retrieved from the database during the state transition
+	AccountUpdated int          // Number of accounts updated during the state transition
+	AccountDeleted int          // Number of accounts deleted during the state transition
+	StorageLoaded  int          // Number of storage slots retrieved from the database during the state transition
+	StorageUpdated atomic.Int64 // Number of storage slots updated during the state transition
+	StorageDeleted atomic.Int64 // Number of storage slots deleted during the state transition
+
+	// CodeLoadBytes is the total number of bytes read from contract code.
+	// This value may be smaller than the actual number of bytes read, since
+	// some APIs (e.g. CodeSize) may load the entire code from either the
+	// cache or the database when the size is not available in the cache.
+	CodeLoaded      int // Number of contract code loaded during the state transition
+	CodeLoadBytes   int // Total bytes of resolved code
+	CodeUpdated     int // Number of contracts with code changes that persisted
+	CodeUpdateBytes int // Total bytes of persisted code written
+}
+
+// StateReadTime returns the total time spent on the state read.
+func (s *Stats) StateReadTime() time.Duration {
+	return s.AccountReads + s.StorageReads + s.CodeReads
+}
+
+// StateHashTime returns the total time spent on the state hash.
+func (s *Stats) StateHashTime() time.Duration {
+	return s.AccountHashes + s.AccountUpdates + s.StorageUpdates
+}
diff --git a/miner/worker.go b/miner/worker.go
index 39a61de318..b75241c20a 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -80,11 +80,6 @@ func (env *environment) txFitsSize(tx *types.Transaction) bool {
 	return env.size+tx.Size() < params.MaxBlockSize-maxBlockSizeBufferZone
 }
 
-// discard terminates the background threads before discarding it.
-func (env *environment) discard() {
-	env.state.StopPrefetcher()
-}
-
 const (
 	commitInterruptNone int32 = iota
 	commitInterruptNewHead
@@ -147,8 +142,6 @@ func (miner *Miner) generateWork(ctx context.Context, genParam *generateParams,
 	if err != nil {
 		return &newPayloadResult{err: err}
 	}
-	defer work.discard()
-
 	// Check withdrawals fit max block size.
 	// Due to the cap on withdrawal count, this can actually never happen, but we still need to
 	// check to ensure the CL notices there's a problem if the withdrawal cap is ever lifted.
@@ -334,8 +327,8 @@ func (miner *Miner) makeEnv(parent *types.Header, header *types.Header, coinbase
 		if err != nil {
 			return nil, err
 		}
+		state.TraceWitness(bundle)
 	}
-	state.StartPrefetcher("miner", bundle)
 	// Note the passed coinbase may be different with header.Coinbase.
 	return &environment{
 		signer:   types.MakeSigner(miner.chainConfig, header.Number, header.Time),