core/state: bypass per-account updateTrie in IntermediateRoot for binary trie

In binary trie mode, IntermediateRoot calls updateTrie() once per dirty
account. But with the binary trie there is only one unified trie
(OpenStorageTrie returns self), so each call redundantly does per-account
trie setup: getPrefetchedTrie, getTrie, slice allocations for
deletions/used, and prefetcher.used — all for the same trie pointer.

Replace the per-account updateTrie() calls with a single flat loop that
applies all storage updates directly to s.trie. The MPT path is
unchanged. The prefetcher trie replacement is guarded to avoid
overwriting the binary trie that received updates.

core/state/statedb.go

@@ -824,22 +824,55 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
 		workers errgroup.Group
 	)
 	if s.db.TrieDB().IsVerkle() {
-		// Whilst MPT storage tries are independent, Verkle has one single trie
+		// Bypass per-account updateTrie() for binary trie. In binary trie mode
-		// for all the accounts and all the storage slots merged together. The
+		// there is only one unified trie (OpenStorageTrie returns self), so the
-		// former can thus be simply parallelized, but updating the latter will
+		// per-account trie setup in updateTrie() (getPrefetchedTrie, getTrie,
-		// need concurrency support within the trie itself. That's a TODO for a
+		// prefetcher.used) is redundant overhead. Apply all storage updates
-		// later time.
+		// directly in a single pass.
-		workers.SetLimit(1)
+		for addr, op := range s.mutations {
-	}
+			if op.applied || op.isDelete() {
-	for addr, op := range s.mutations {
+				continue
-		if op.applied || op.isDelete() {
+			}
-			continue
+			obj := s.stateObjects[addr]
+			if len(obj.uncommittedStorage) == 0 {
+				continue
+			}
+			for key, origin := range obj.uncommittedStorage {
+				value, exist := obj.pendingStorage[key]
+				if value == origin || !exist {
+					continue
+				}
+				if (value != common.Hash{}) {
+					if err := s.trie.UpdateStorage(addr, key[:], common.TrimLeftZeroes(value[:])); err != nil {
+						s.setError(err)
+					}
+				} else {
+					if err := s.trie.DeleteStorage(addr, key[:]); err != nil {
+						s.setError(err)
+					}
+				}
+			}
 		}
-		obj := s.stateObjects[addr] // closure for the task runner below
+		// Clear uncommittedStorage and assign trie on each touched object.
-		workers.Go(func() error {
+		// obj.trie must be set because this path bypasses updateTrie(), which
-			if s.db.TrieDB().IsVerkle() {
+		// is where obj.trie normally gets lazily loaded via getTrie().
-				obj.updateTrie()
+		for addr, op := range s.mutations {
-			} else {
+			if op.applied || op.isDelete() {
+				continue
+			}
+			obj := s.stateObjects[addr]
+			if len(obj.uncommittedStorage) > 0 {
+				obj.uncommittedStorage = make(Storage)
+			}
+			obj.trie = s.trie
+		}
+	} else {
+		for addr, op := range s.mutations {
+			if op.applied || op.isDelete() {
+				continue
+			}
+			obj := s.stateObjects[addr] // closure for the task runner below
+			workers.Go(func() error {
 				obj.updateRoot()
 
 				// If witness building is enabled and the state object has a trie,
@@ -847,9 +880,9 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
 				if s.witness != nil && obj.trie != nil {
 					s.witness.AddState(obj.trie.Witness())
 				}
-			}
+				return nil
-			return nil
+			})
-		})
+		}
 	}
 	// If witness building is enabled, gather all the read-only accesses.
 	// Skip witness collection in Verkle mode, they will be gathered
@@ -911,7 +944,7 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
 	// only a single trie is used for state hashing. Replacing a non-nil verkle tree
 	// here could result in losing uncommitted changes from storage.
 	start = time.Now()
-	if s.prefetcher != nil {
+	if s.prefetcher != nil && !s.db.TrieDB().IsVerkle() {
 		if trie := s.prefetcher.trie(common.Hash{}, s.originalRoot); trie == nil {
 			log.Error("Failed to retrieve account pre-fetcher trie")
 		} else {