diff --git a/core/blockchain.go b/core/blockchain.go index 963e7bcec5..663dbc31f6 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -196,6 +196,9 @@ type BlockChainConfig struct { // If the value is zero, all transactions of the entire chain will be indexed. // If the value is -1, indexing is disabled. TxLookupLimit int64 + + // EnableStateSizeTracking indicates whether the state size tracking is enabled. + EnableStateSizeTracking bool } // DefaultConfig returns the default config. @@ -333,8 +336,7 @@ type BlockChain struct { prefetcher Prefetcher processor Processor // Block transaction processor interface logger *tracing.Hooks - - stateSizeGen *state.StateSizeGenerator // State size tracking + stateSizer *state.SizeTracker // State size tracking lastForkReadyAlert time.Time // Last time there was a fork readiness print out } @@ -530,9 +532,10 @@ func NewBlockChain(db ethdb.Database, genesis *Genesis, engine consensus.Engine, } // Start state size tracker - bc.stateSizeGen = state.NewStateSizeGenerator(bc.statedb.DiskDB(), bc.triedb, head.Root) - log.Info("Started state size generator", "root", head.Root) - + if bc.cfg.EnableStateSizeTracking { + bc.stateSizer = state.NewSizeTracker(bc.db, bc.triedb) + bc.stateSizer.Start() + } return bc, nil } @@ -1259,12 +1262,10 @@ func (bc *BlockChain) stopWithoutSaving() { // Signal shutdown to all goroutines. bc.InterruptInsert(true) - // Stop state size generator if running - if bc.stateSizeGen != nil { - bc.stateSizeGen.Stop() - log.Info("Stopped state size generator") + // Stop state size tracker + if bc.stateSizer != nil { + bc.stateSizer.Stop() } - // Now wait for all chain modifications to end and persistent goroutines to exit. // // Note: Close waits for the mutex to become available, i.e. any running chain @@ -1604,10 +1605,9 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types. if err != nil { return err } - - // Track state size changes if generator is running - if bc.stateSizeGen != nil && stateUpdate != nil { - bc.stateSizeGen.Track(stateUpdate) + // Emit the state update to the state sizestats if it's active + if bc.stateSizer != nil { + bc.stateSizer.Notify(stateUpdate) } // If node is running in path mode, skip explicit gc operation // which is unnecessary in this mode. diff --git a/core/state/metrics.go b/core/state/metrics.go index 57f6d4a4ed..dd4b2e9838 100644 --- a/core/state/metrics.go +++ b/core/state/metrics.go @@ -29,14 +29,4 @@ var ( storageTriesUpdatedMeter = metrics.NewRegisteredMeter("state/update/storagenodes", nil) accountTrieDeletedMeter = metrics.NewRegisteredMeter("state/delete/accountnodes", nil) storageTriesDeletedMeter = metrics.NewRegisteredMeter("state/delete/storagenodes", nil) - - // State size metrics - accountCountGauge = metrics.NewRegisteredGauge("state/account/count", nil) - accountBytesGauge = metrics.NewRegisteredGauge("state/account/bytes", nil) - storageCountGauge = metrics.NewRegisteredGauge("state/storage/count", nil) - storageBytesGauge = metrics.NewRegisteredGauge("state/storage/bytes", nil) - trienodeCountGauge = metrics.NewRegisteredGauge("state/trienode/count", nil) - trienodeBytesGauge = metrics.NewRegisteredGauge("state/trienode/bytes", nil) - contractCountGauge = metrics.NewRegisteredGauge("state/contract/count", nil) - contractBytesGauge = metrics.NewRegisteredGauge("state/contract/bytes", nil) ) diff --git a/core/state/state_sizer.go b/core/state/state_sizer.go new file mode 100644 index 0000000000..687b2d043e --- /dev/null +++ b/core/state/state_sizer.go @@ -0,0 +1,547 @@ +// Copyright 2025 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 . + +package state + +import ( + "container/heap" + "errors" + "fmt" + "maps" + "slices" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" + "github.com/ethereum/go-ethereum/triedb" + "golang.org/x/sync/errgroup" +) + +const ( + statEvictThreshold = 128 // the depth of statistic to be preserved +) + +// Metrics for uploading the state statistics. +var ( + accountsGauge = metrics.NewRegisteredGauge("state/size/account/count", nil) + accountBytesGauge = metrics.NewRegisteredGauge("state/size/account/bytes", nil) + storagesGauge = metrics.NewRegisteredGauge("state/size/storage/count", nil) + storageBytesGauge = metrics.NewRegisteredGauge("state/size/storage/bytes", nil) + accountTrienodesGauge = metrics.NewRegisteredGauge("state/size/trienode/account/count", nil) + accountTrienodeBytesGauge = metrics.NewRegisteredGauge("state/size/trienode/account/bytes", nil) + storageTrienodesGauge = metrics.NewRegisteredGauge("state/size/trienode/storage/count", nil) + storageTrienodeBytesGauge = metrics.NewRegisteredGauge("state/size/trienode/storage/bytes", nil) + contractCodesGauge = metrics.NewRegisteredGauge("state/size/contractcode/count", nil) + contractCodeBytesGauge = metrics.NewRegisteredGauge("state/size/contractcode/bytes", nil) +) + +// Database key scheme for states. +var ( + accountKeySize = int64(len(rawdb.SnapshotAccountPrefix) + common.HashLength) + storageKeySize = int64(len(rawdb.SnapshotStoragePrefix) + common.HashLength + common.HashLength) + accountTrienodePrefixSize = int64(len(rawdb.TrieNodeAccountPrefix)) + storageTrienodePrefixSize = int64(len(rawdb.TrieNodeStoragePrefix) + common.HashLength) + codeKeySize = int64(len(rawdb.CodePrefix) + common.HashLength) +) + +// SizeStats represents either the current state size statistics or the size +// differences resulting from a state transition. +type SizeStats struct { + StateRoot common.Hash // State root hash at the time of measurement + BlockNumber uint64 // Associated block number at the time of measurement + + Accounts int64 // Total number of accounts in the state + AccountBytes int64 // Total storage size used by all account data (in bytes) + Storages int64 // Total number of storage slots across all accounts + StorageBytes int64 // Total storage size used by all storage slot data (in bytes) + AccountTrienodes int64 // Total number of account trie nodes in the state + AccountTrienodeBytes int64 // Total storage size occupied by account trie nodes (in bytes) + StorageTrienodes int64 // Total number of storage trie nodes in the state + StorageTrienodeBytes int64 // Total storage size occupied by storage trie nodes (in bytes) + ContractCodes int64 // Total number of contract codes in the state + ContractCodeBytes int64 // Total size of all contract code (in bytes) +} + +// add applies the given state diffs and produces a new version of the statistics. +func (s SizeStats) add(diff SizeStats) SizeStats { + var combo SizeStats + combo.StateRoot = diff.StateRoot + combo.BlockNumber = diff.BlockNumber + + combo.Accounts += diff.Accounts + combo.AccountBytes += diff.AccountBytes + combo.Storages += diff.Storages + combo.StorageBytes += diff.StorageBytes + combo.AccountTrienodes += diff.AccountTrienodes + combo.AccountTrienodeBytes += diff.AccountTrienodeBytes + combo.StorageTrienodes += diff.StorageTrienodes + combo.StorageTrienodeBytes += diff.StorageTrienodeBytes + combo.ContractCodes += diff.ContractCodes + combo.ContractCodeBytes += diff.ContractCodeBytes + return combo +} + +// calSizeStats measures the state size changes of the provided state update. +func calSizeStats(update *stateUpdate) (SizeStats, error) { + stats := SizeStats{ + BlockNumber: update.blockNumber, + StateRoot: update.root, + } + + // Measure the account changes + for addr, oldValue := range update.accountsOrigin { + addrHash := crypto.Keccak256Hash(addr.Bytes()) + newValue, exists := update.accounts[addrHash] + if !exists { + return SizeStats{}, fmt.Errorf("account %x not found", addr) + } + oldLen, newLen := len(oldValue), len(newValue) + + switch { + case oldLen > 0 && newLen == 0: + // Account deletion + stats.Accounts -= 1 + stats.AccountBytes -= accountKeySize + int64(oldLen) + case oldLen == 0 && newLen > 0: + // Account creation + stats.Accounts += 1 + stats.AccountBytes += accountKeySize + int64(newLen) + default: + // Account update + stats.AccountBytes += int64(newLen - oldLen) + } + } + + // Measure storage changes + for addr, slots := range update.storagesOrigin { + addrHash := crypto.Keccak256Hash(addr.Bytes()) + subset, exists := update.storages[addrHash] + if !exists { + return SizeStats{}, fmt.Errorf("storage %x not found", addr) + } + for key, oldValue := range slots { + var ( + exists bool + newValue []byte + ) + if update.rawStorageKey { + newValue, exists = subset[crypto.Keccak256Hash(key.Bytes())] + } else { + newValue, exists = subset[key] + } + if !exists { + return SizeStats{}, fmt.Errorf("storage slot %x-%x not found", addr, key) + } + oldLen, newLen := len(oldValue), len(newValue) + + switch { + case oldLen > 0 && newLen == 0: + // Storage deletion + stats.Storages -= 1 + stats.StorageBytes -= storageKeySize + int64(oldLen) + case oldLen == 0 && newLen > 0: + // Storage creation + stats.Storages += 1 + stats.StorageBytes += storageKeySize + int64(newLen) + default: + // Storage update + stats.StorageBytes += int64(newLen - oldLen) + } + } + } + + // Measure trienode changes + for owner, subset := range update.nodes.Sets { + var ( + keyPrefix int64 + isAccount = owner == (common.Hash{}) + ) + if isAccount { + keyPrefix = accountTrienodePrefixSize + } else { + keyPrefix = storageTrienodePrefixSize + } + + // Iterate over Origins since every modified node has an origin entry + for path, oldNode := range subset.Origins { + newNode, exists := subset.Nodes[path] + if !exists { + return SizeStats{}, fmt.Errorf("node %x-%v not found", owner, path) + } + keySize := keyPrefix + int64(len(path)) + + switch { + case len(oldNode) > 0 && len(newNode.Blob) == 0: + // Node deletion + if isAccount { + stats.AccountTrienodes -= 1 + stats.AccountTrienodeBytes -= keySize + int64(len(oldNode)) + } else { + stats.StorageTrienodes -= 1 + stats.StorageTrienodeBytes -= keySize + int64(len(oldNode)) + } + case len(oldNode) == 0 && len(newNode.Blob) > 0: + // Node creation + if isAccount { + stats.AccountTrienodes += 1 + stats.AccountTrienodeBytes += keySize + int64(len(newNode.Blob)) + } else { + stats.StorageTrienodes += 1 + stats.StorageTrienodeBytes += keySize + int64(len(newNode.Blob)) + } + default: + // Node update + if isAccount { + stats.AccountTrienodeBytes += int64(len(newNode.Blob) - len(oldNode)) + } else { + stats.StorageTrienodeBytes += int64(len(newNode.Blob) - len(oldNode)) + } + } + } + } + + // Measure code changes. Note that the reported contract code size may be slightly + // inaccurate due to database deduplication (code is stored by its hash). However, + // this deviation is negligible and acceptable for measurement purposes. + for _, code := range update.codes { + stats.ContractCodes += 1 + stats.ContractCodeBytes += codeKeySize + int64(len(code.blob)) + } + return stats, nil +} + +// SizeTracker handles the state size initialization and tracks of state size metrics. +type SizeTracker struct { + db ethdb.KeyValueStore + triedb *triedb.Database + abort chan struct{} + aborted chan struct{} + updateCh chan *stateUpdate +} + +// NewSizeTracker creates a new state size tracker and starts it automatically +func NewSizeTracker(db ethdb.KeyValueStore, triedb *triedb.Database) *SizeTracker { + return &SizeTracker{ + db: db, + triedb: triedb, + abort: make(chan struct{}), + aborted: make(chan struct{}), + updateCh: make(chan *stateUpdate), + } +} + +func (t *SizeTracker) Start() { + go t.run() +} + +func (t *SizeTracker) Stop() { + close(t.abort) + <-t.aborted +} + +// sizeStatsHeap is a heap.Interface implementation over statesize statistics for +// retrieving the oldest statistics for eviction. +type sizeStatsHeap []SizeStats + +func (h sizeStatsHeap) Len() int { return len(h) } +func (h sizeStatsHeap) Less(i, j int) bool { return h[i].BlockNumber < h[j].BlockNumber } +func (h sizeStatsHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] } + +func (h *sizeStatsHeap) Push(x any) { + *h = append(*h, x.(SizeStats)) +} + +func (h *sizeStatsHeap) Pop() any { + old := *h + n := len(old) + x := old[n-1] + *h = old[0 : n-1] + return x +} + +// run performs the state size initialization and handles updates +func (t *SizeTracker) run() { + defer close(t.aborted) + + stats, err := t.init() // launch background thread for state size init + if err != nil { + return + } + h := sizeStatsHeap(slices.Collect(maps.Values(stats))) + heap.Init(&h) + + for { + select { + case u := <-t.updateCh: + base, found := stats[u.originRoot] + if !found { + continue + } + diff, err := calSizeStats(u) + if err != nil { + continue + } + stat := base.add(diff) + t.upload(stat) + + stats[u.root] = stat + heap.Push(&h, stats[u.root]) + for u.blockNumber-h[0].BlockNumber > statEvictThreshold { + heap.Pop(&h) + } + + case <-t.abort: + return + } + } +} + +type buildResult struct { + stat SizeStats + root common.Hash + blockNumber uint64 + err error +} + +func (t *SizeTracker) init() (map[common.Hash]SizeStats, error) { + // Wait for snapshot completion and then init + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + +wait: + for { + select { + case <-ticker.C: + if t.triedb.SnapshotCompleted() { + break wait + } + case <-t.abort: + return nil, errors.New("size tracker closed") + } + } + + var ( + updates = make(map[common.Hash]*stateUpdate) + children = make(map[common.Hash][]common.Hash) + done = make(chan buildResult) + ) + for { + select { + case u := <-t.updateCh: + updates[u.root] = u + children[u.originRoot] = append(children[u.originRoot], u.root) + + case <-ticker.C: + root := rawdb.ReadSnapshotRoot(t.db) + if root == (common.Hash{}) { + continue + } + entry, exists := updates[root] + if !exists { + continue + } + if done == nil { + done = make(chan buildResult) + go t.build(entry.root, entry.blockNumber, done) + } + + case result := <-done: + if result.err != nil { + return nil, result.err + } + var ( + stats = make(map[common.Hash]SizeStats) + apply func(root common.Hash, stats SizeStats) error + ) + apply = func(root common.Hash, stat SizeStats) error { + for _, child := range children[root] { + entry, ok := updates[child] + if !ok { + return fmt.Errorf("the state update is not found, %x", child) + } + diff, err := calSizeStats(entry) + if err != nil { + return err + } + stats[child] = stat.add(diff) + if err := apply(child, stats[child]); err != nil { + return err + } + } + return nil + } + if err := apply(result.root, result.stat); err != nil { + return nil, err + } + return stats, nil + + case <-t.abort: + return nil, errors.New("size tracker closed") + } + } +} + +func (t *SizeTracker) build(root common.Hash, blockNumber uint64, done chan buildResult) { + // Metrics will be directly updated by each goroutine + var ( + accounts, accountBytes int64 + storages, storageBytes int64 + codes, codeBytes int64 + + accountTrienodes, accountTrienodeBytes int64 + storageTrienodes, storageTrienodeBytes int64 + + group errgroup.Group + ) + + // Start all table iterations concurrently with direct metric updates + group.Go(func() error { + count, bytes, err := t.iterateTable(t.abort, rawdb.SnapshotAccountPrefix, "account") + if err != nil { + return err + } + accounts, accountBytes = count, bytes + return nil + }) + + group.Go(func() error { + count, bytes, err := t.iterateTable(t.abort, rawdb.SnapshotStoragePrefix, "storage") + if err != nil { + return err + } + storages, storageBytes = count, bytes + return nil + }) + + group.Go(func() error { + count, bytes, err := t.iterateTable(t.abort, rawdb.TrieNodeAccountPrefix, "accountnode") + if err != nil { + return err + } + accountTrienodes, accountTrienodeBytes = count, bytes + return nil + }) + + group.Go(func() error { + count, bytes, err := t.iterateTable(t.abort, rawdb.TrieNodeStoragePrefix, "storagenode") + if err != nil { + return err + } + storageTrienodes, storageTrienodeBytes = count, bytes + return nil + }) + + group.Go(func() error { + count, bytes, err := t.iterateTable(t.abort, rawdb.CodePrefix, "contractcode") + if err != nil { + return err + } + codes, codeBytes = count, bytes + return nil + }) + + // Wait for all goroutines to complete + if err := group.Wait(); err != nil { + done <- buildResult{err: err} + } else { + stat := SizeStats{ + StateRoot: root, + BlockNumber: blockNumber, + Accounts: accounts, + AccountBytes: accountBytes, + Storages: storages, + StorageBytes: storageBytes, + AccountTrienodes: accountTrienodes, + AccountTrienodeBytes: accountTrienodeBytes, + StorageTrienodes: storageTrienodes, + StorageTrienodeBytes: storageTrienodeBytes, + ContractCodes: codes, + ContractCodeBytes: codeBytes, + } + done <- buildResult{ + root: root, + blockNumber: blockNumber, + stat: stat, + } + } +} + +// Notify is an async method used to send the state update to the size tracker. +// It ignores empty updates (where no state changes occurred). +// If the channel is full, it drops the update to avoid blocking. +func (t *SizeTracker) Notify(update *stateUpdate) { + if update == nil || update.empty() { + return + } + select { + case t.updateCh <- update: + case <-t.abort: + return + } +} + +// iterateTable performs iteration over a specific table and returns the results. +func (t *SizeTracker) iterateTable(closed chan struct{}, prefix []byte, name string) (int64, int64, error) { + var ( + start = time.Now() + logged = time.Now() + count, bytes int64 + ) + iter := t.db.NewIterator(prefix, nil) + defer iter.Release() + + log.Info("Iterating state", "category", name) + for iter.Next() { + count++ + bytes += int64(len(iter.Key()) + len(iter.Value())) + + if time.Since(logged) > time.Second*8 { + logged = time.Now() + + select { + case <-closed: + log.Info("State iteration cancelled", "category", name) + return 0, 0, errors.New("size tracker closed") + default: + log.Info("Iterating state", "category", name, "count", count, "size", common.StorageSize(bytes)) + } + } + } + // Check for iterator errors + if err := iter.Error(); err != nil { + log.Error("Iterator error", "category", name, "err", err) + return 0, 0, err + } + log.Info("Finished state iteration", "category", name, "count", count, "size", common.StorageSize(bytes), "elapsed", common.PrettyDuration(time.Since(start))) + return count, bytes, nil +} + +func (t *SizeTracker) upload(stats SizeStats) { + accountsGauge.Update(stats.Accounts) + accountBytesGauge.Update(stats.AccountBytes) + storagesGauge.Update(stats.Storages) + storageBytesGauge.Update(stats.StorageBytes) + accountTrienodesGauge.Update(stats.AccountTrienodes) + accountTrienodeBytesGauge.Update(stats.AccountTrienodeBytes) + storageTrienodesGauge.Update(stats.StorageTrienodes) + storageTrienodeBytesGauge.Update(stats.StorageTrienodeBytes) + contractCodesGauge.Update(stats.ContractCodes) + contractCodeBytesGauge.Update(stats.ContractCodeBytes) +} diff --git a/core/state/statedb.go b/core/state/statedb.go index 265561df2d..f74065666b 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -668,7 +668,7 @@ func (s *StateDB) CreateContract(addr common.Address) { // Copy creates a deep, independent copy of the state. // Snapshots of the copied state cannot be applied to the copy. func (s *StateDB) Copy() *StateDB { - // Copy all the basic fields, initialize the memory ones + // Copy all the basic fields, init the memory ones state := &StateDB{ db: s.db, reader: s.reader, @@ -1156,7 +1156,7 @@ func (s *StateDB) GetTrie() Trie { // commit gathers the state mutations accumulated along with the associated // trie changes, resetting all internal flags with the new state as the base. -func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool) (*stateUpdate, error) { +func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNumber uint64) (*stateUpdate, error) { // Short circuit in case any database failure occurred earlier. if s.dbErr != nil { return nil, fmt.Errorf("commit aborted due to earlier error: %v", s.dbErr) @@ -1308,13 +1308,13 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool) (*stateU origin := s.originalRoot s.originalRoot = root - return newStateUpdate(noStorageWiping, origin, root, deletes, updates, nodes), nil + return newStateUpdate(noStorageWiping, origin, root, blockNumber, deletes, updates, nodes), nil } // commitAndFlush is a wrapper of commit which also commits the state mutations // to the configured data stores. func (s *StateDB) commitAndFlush(block uint64, deleteEmptyObjects bool, noStorageWiping bool) (*stateUpdate, error) { - ret, err := s.commit(deleteEmptyObjects, noStorageWiping) + ret, err := s.commit(deleteEmptyObjects, noStorageWiping, block) if err != nil { return nil, err } diff --git a/core/state/statedb_test.go b/core/state/statedb_test.go index 147546a3c7..20bcbabdf8 100644 --- a/core/state/statedb_test.go +++ b/core/state/statedb_test.go @@ -281,7 +281,7 @@ func TestCopyObjectState(t *testing.T) { cpy := orig.Copy() for _, op := range cpy.mutations { if have, want := op.applied, false; have != want { - t.Fatalf("Error in test itself, the 'done' flag should not be set before Commit, have %v want %v", have, want) + t.Fatalf("Error in test itself, the 'aborted' flag should not be set before Commit, have %v want %v", have, want) } } orig.Commit(0, true, false) diff --git a/core/state/statesize.go b/core/state/statesize.go deleted file mode 100644 index a696958064..0000000000 --- a/core/state/statesize.go +++ /dev/null @@ -1,519 +0,0 @@ -// Copyright 2025 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 . - -package state - -import ( - "context" - "encoding/json" - "time" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/core/rawdb" - "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/triedb" - "golang.org/x/sync/errgroup" -) - -var ( - accountSnapKeySize = int64(len(rawdb.SnapshotAccountPrefix) + common.HashLength) - storageSnapKeySize = int64(len(rawdb.SnapshotStoragePrefix) + common.HashLength + common.HashLength) - accountTrieKeyPrefixSize = int64(len(rawdb.TrieNodeAccountPrefix)) - storageTrieKeyPrefixSize = int64(len(rawdb.TrieNodeStoragePrefix) + common.HashLength) - codeKeySize = int64(len(rawdb.CodePrefix) + common.HashLength) -) - -// stateSizeMetrics represents the current state size statistics -type stateSizeMetrics struct { - Root common.Hash // Root hash of the state trie - AccountCount int64 - AccountBytes int64 - StorageCount int64 - StorageBytes int64 - TrieNodeCount int64 - TrieNodeBytes int64 - ContractCount int64 - ContractBytes int64 -} - -// StateSizeGenerator handles the initialization and tracking of state size metrics -type StateSizeGenerator struct { - db ethdb.KeyValueStore - triedb *triedb.Database - abort chan struct{} - done chan struct{} - updateChan chan *stateUpdate // Async message channel for updates - metrics *stateSizeMetrics - buffered *stateSizeMetrics -} - -// NewStateSizeGenerator creates a new state size generator and starts it automatically -func NewStateSizeGenerator(db ethdb.KeyValueStore, triedb *triedb.Database, root common.Hash) *StateSizeGenerator { - g := &StateSizeGenerator{ - db: db, - triedb: triedb, - abort: make(chan struct{}), - done: make(chan struct{}), - updateChan: make(chan *stateUpdate, 1000), // Buffered channel for updates - metrics: &stateSizeMetrics{Root: root}, - buffered: &stateSizeMetrics{Root: root}, - } - - // Start the generator automatically - go g.generate() - - return g -} - -// Stop terminates the background generation and persists the metrics. -func (g *StateSizeGenerator) Stop() { - close(g.abort) - - <-g.done - - g.persistMetrics() -} - -// generate performs the state size initialization and handles updates -func (g *StateSizeGenerator) generate() { - defer close(g.done) - - var inited bool - - // Check if we already have existing metrics - if g.hasExistingMetrics() { - log.Info("State size metrics already initialized") - inited = true - } - - initDone := g.initialize() - - for { - select { - case update := <-g.updateChan: - g.handleUpdate(update, inited) - - case <-g.abort: - log.Info("State size generation aborted") - - // Wait for initialization to complete with timeout - if initDone != nil { - select { - case <-initDone: - log.Debug("Initialization completed before abort") - case <-time.After(5 * time.Second): - log.Warn("Initialization did not finish in time during abort") - } - } - return - - case <-initDone: - // Initialization completed, merge buffered metrics - if g.buffered != nil { - log.Info("Merging buffered metrics into main metrics") - g.metrics.Root = g.buffered.Root - g.metrics.AccountCount += g.buffered.AccountCount - g.metrics.AccountBytes += g.buffered.AccountBytes - g.metrics.StorageCount += g.buffered.StorageCount - g.metrics.StorageBytes += g.buffered.StorageBytes - g.metrics.TrieNodeCount += g.buffered.TrieNodeCount - g.metrics.TrieNodeBytes += g.buffered.TrieNodeBytes - g.metrics.ContractCount += g.buffered.ContractCount - g.metrics.ContractBytes += g.buffered.ContractBytes - - g.buffered = nil - } - - inited = true - initDone = nil // Clear the channel to prevent future selects - } - } -} - -// initialize starts the initialization process if not already initialized -func (g *StateSizeGenerator) initialize() chan struct{} { - done := make(chan struct{}) - - // Wait for snapshot completion and then initialize - go func() { - defer close(done) - - LOOP: - // Wait for snapshot generator to complete first - for { - root, done := g.triedb.SnapshotCompleted() - if done { - g.metrics.Root = root - g.buffered.Root = root - break LOOP - } - - select { - case <-g.abort: - log.Info("State size initialization aborted during snapshot wait") - return - case <-time.After(10 * time.Second): - // Continue checking for snapshot completion - } - } - - // Start actual initialization - start := time.Now() - log.Info("Starting state size initialization") - if err := g.initializeMetrics(); err != nil { - log.Error("Failed to initialize state size metrics", "err", err) - return - } - - log.Info("Completed state size initialization", "elapsed", time.Since(start)) - }() - - return done -} - -// handleUpdate processes a single update with proper root continuity checking -func (g *StateSizeGenerator) handleUpdate(update *stateUpdate, initialized bool) { - diff := g.calculateUpdateDiff(update) - - var targetMetrics *stateSizeMetrics - if initialized { - targetMetrics = g.metrics - } else { - targetMetrics = g.buffered - } - - // Check root continuity - the update should build on our current state - if targetMetrics.Root != (common.Hash{}) && targetMetrics.Root != update.originRoot { - log.Warn("State update root discontinuity detected", "current", targetMetrics.Root, "updateOrigin", update.originRoot, "updateNew", update.root) - } - - // Update to the new state root - targetMetrics.Root = update.root - targetMetrics.AccountCount += diff.AccountCount - targetMetrics.AccountBytes += diff.AccountBytes - targetMetrics.StorageCount += diff.StorageCount - targetMetrics.StorageBytes += diff.StorageBytes - targetMetrics.TrieNodeCount += diff.TrieNodeCount - targetMetrics.TrieNodeBytes += diff.TrieNodeBytes - targetMetrics.ContractCount += diff.ContractCount - targetMetrics.ContractBytes += diff.ContractBytes - - // Fire the metrics and persist only if initialization is done - if initialized { - g.updateMetrics() - g.persistMetrics() - } -} - -// calculateUpdateDiff calculates the diff for a state update -func (g *StateSizeGenerator) calculateUpdateDiff(update *stateUpdate) stateSizeMetrics { - var diff stateSizeMetrics - - // Calculate account changes - for addr, oldValue := range update.accountsOrigin { - addrHash := crypto.Keccak256Hash(addr.Bytes()) - newValue, exists := update.accounts[addrHash] - if !exists { - log.Warn("State update missing account", "address", addr) - continue - } - - oldLen, newLen := len(oldValue), len(newValue) - if oldLen > 0 && newLen == 0 { - // Account deletion - diff.AccountCount -= 1 - diff.AccountBytes -= accountSnapKeySize + int64(oldLen) - } else if oldLen == 0 && newLen > 0 { - // Account creation - diff.AccountCount += 1 - diff.AccountBytes += accountSnapKeySize + int64(newLen) - } else { - // Account update - diff.AccountBytes += int64(newLen - oldLen) - } - } - - // Calculate storage changes - for addr, slots := range update.storagesOrigin { - addrHash := crypto.Keccak256Hash(addr.Bytes()) - subset, exists := update.storages[addrHash] - if !exists { - log.Warn("State update missing storage", "address", addr) - continue - } - for key, oldValue := range slots { - var ( - exists bool - newValue []byte - ) - if update.rawStorageKey { - newValue, exists = subset[crypto.Keccak256Hash(key.Bytes())] - } else { - newValue, exists = subset[key] - } - if !exists { - log.Warn("State update missing storage slot", "address", addr, "key", key) - continue - } - - oldLen, newLen := len(oldValue), len(newValue) - if oldLen > 0 && newLen == 0 { - // Storage deletion - diff.StorageCount -= 1 - diff.StorageBytes -= storageSnapKeySize + int64(oldLen) - } else if oldLen == 0 && newLen > 0 { - // Storage creation - diff.StorageCount += 1 - diff.StorageBytes += storageSnapKeySize + int64(newLen) - } else { - // Storage update - diff.StorageBytes += int64(newLen - oldLen) - } - } - } - - // Calculate trie node changes - for owner, subset := range update.nodes.Sets { - isAccountTrie := owner == (common.Hash{}) - var keyPrefixSize int64 - if isAccountTrie { - keyPrefixSize = accountTrieKeyPrefixSize - } else { - keyPrefixSize = storageTrieKeyPrefixSize - } - - // Iterate over Origins since every modified node has an origin entry - for path, oldNode := range subset.Origins { - newNode, hasNew := subset.Nodes[path] - - keySize := keyPrefixSize + int64(len(path)) - - if len(oldNode) > 0 && (!hasNew || len(newNode.Blob) == 0) { - // Node deletion - diff.TrieNodeCount -= 1 - diff.TrieNodeBytes -= keySize + int64(len(oldNode)) - } else if len(oldNode) == 0 && hasNew && len(newNode.Blob) > 0 { - // Node creation - diff.TrieNodeCount += 1 - diff.TrieNodeBytes += keySize + int64(len(newNode.Blob)) - } else if len(oldNode) > 0 && hasNew && len(newNode.Blob) > 0 { - // Node update - diff.TrieNodeBytes += int64(len(newNode.Blob) - len(oldNode)) - } - } - } - - // Calculate code changes - for _, code := range update.codes { - diff.ContractCount += 1 - diff.ContractBytes += codeKeySize + int64(len(code.blob)) - } - - return diff -} - -// Track is an async method used to send the state update to the generator. -// It ignores empty updates (where no state changes occurred). -// If the channel is full, it drops the update to avoid blocking. -func (g *StateSizeGenerator) Track(update *stateUpdate) { - if update == nil || update.empty() { - return - } - - g.updateChan <- update -} - -// hasExistingMetrics checks if state size metrics already exist in the database -// and if they are continuous with the current root -func (g *StateSizeGenerator) hasExistingMetrics() bool { - data := rawdb.ReadStateSizeMetrics(g.db) - if data == nil { - return false - } - - var existed stateSizeMetrics - if err := json.Unmarshal(data, &existed); err != nil { - log.Warn("Failed to decode existed state size metrics", "err", err) - return false - } - - // Check if the existing metrics root matches our current root - if (g.metrics.Root != common.Hash{}) && existed.Root != g.metrics.Root { - log.Info("Existing state size metrics found but root mismatch", "existed", existed.Root, "current", g.metrics.Root) - return false - } - - // Root matches - load the existing metrics - log.Info("Loading existing state size metrics", "root", existed.Root) - g.metrics = &existed - return true -} - -// initializeMetrics performs the actual metrics initialization using errgroup -func (g *StateSizeGenerator) initializeMetrics() error { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - go func() { - select { - case <-g.abort: - cancel() // Cancel context when abort is signaled - case <-ctx.Done(): - // Context already cancelled - } - }() - - // Create errgroup with context - group, ctx := errgroup.WithContext(ctx) - - // Metrics will be directly updated by each goroutine - var ( - accountSnapCount, accountSnapBytes int64 - storageSnapCount, storageSnapBytes int64 - accountTrieCount, accountTrieBytes int64 - storageTrieCount, storageTrieBytes int64 - contractCount, contractBytes int64 - ) - - // Start all table iterations concurrently with direct metric updates - group.Go(func() error { - count, bytes, err := g.iterateTable(ctx, rawdb.SnapshotAccountPrefix, "accountSnap") - if err != nil { - return err - } - accountSnapCount, accountSnapBytes = count, bytes - return nil - }) - - group.Go(func() error { - count, bytes, err := g.iterateTable(ctx, rawdb.SnapshotStoragePrefix, "storageSnap") - if err != nil { - return err - } - storageSnapCount, storageSnapBytes = count, bytes - return nil - }) - - group.Go(func() error { - count, bytes, err := g.iterateTable(ctx, rawdb.TrieNodeAccountPrefix, "accountTrie") - if err != nil { - return err - } - accountTrieCount, accountTrieBytes = count, bytes - return nil - }) - - group.Go(func() error { - count, bytes, err := g.iterateTable(ctx, rawdb.TrieNodeStoragePrefix, "storageTrie") - if err != nil { - return err - } - storageTrieCount, storageTrieBytes = count, bytes - return nil - }) - - group.Go(func() error { - count, bytes, err := g.iterateTable(ctx, rawdb.CodePrefix, "contract") - if err != nil { - return err - } - contractCount, contractBytes = count, bytes - return nil - }) - - // Wait for all goroutines to complete - if err := group.Wait(); err != nil { - return err - } - - g.metrics.AccountCount = accountSnapCount - g.metrics.AccountBytes = accountSnapBytes - g.metrics.StorageCount = storageSnapCount - g.metrics.StorageBytes = storageSnapBytes - g.metrics.TrieNodeCount = accountTrieCount + storageTrieCount - g.metrics.TrieNodeBytes = accountTrieBytes + storageTrieBytes - g.metrics.ContractCount = contractCount - g.metrics.ContractBytes = contractBytes - - g.updateMetrics() - g.persistMetrics() - - return nil -} - -// iterateTable performs iteration over a specific table and returns the results -func (g *StateSizeGenerator) iterateTable(ctx context.Context, prefix []byte, name string) (int64, int64, error) { - log.Info("Iterating over state size", "table", name) - start := time.Now() - - var count, bytes int64 - iter := g.db.NewIterator(prefix, nil) - defer iter.Release() - - for iter.Next() { - count++ - bytes += int64(len(iter.Key()) + len(iter.Value())) - - // Check for cancellation periodically for performance - if count%10000 == 0 { - select { - case <-ctx.Done(): - log.Info("State size iteration cancelled", "table", name, "count", count) - return 0, 0, ctx.Err() - default: - } - } - } - - // Check for iterator errors - if err := iter.Error(); err != nil { - log.Error("Iterator error during state size calculation", "table", name, "err", err) - return 0, 0, err - } - - log.Info("Finished iterating over state size", "table", name, "count", count, "bytes", bytes, "elapsed", common.PrettyDuration(time.Since(start))) - - return count, bytes, nil -} - -func (g *StateSizeGenerator) updateMetrics() { - accountCountGauge.Update(g.metrics.AccountCount) - accountBytesGauge.Update(g.metrics.AccountBytes) - storageCountGauge.Update(g.metrics.StorageCount) - storageBytesGauge.Update(g.metrics.StorageBytes) - trienodeCountGauge.Update(g.metrics.TrieNodeCount) - trienodeBytesGauge.Update(g.metrics.TrieNodeBytes) - contractCountGauge.Update(g.metrics.ContractCount) - contractBytesGauge.Update(g.metrics.ContractBytes) -} - -// persistMetrics saves the current metrics to the database -func (g *StateSizeGenerator) persistMetrics() { - // RLP doesn't support int64, so we use JSON for simplicity - data, err := json.Marshal(*g.metrics) - if err != nil { - log.Error("Failed to encode state size metrics", "err", err) - return - } - - batch := g.db.NewBatch() - rawdb.WriteStateSizeMetrics(batch, data) - if err := batch.Write(); err != nil { - log.Error("Failed to persist state size metrics", "err", err) - } -} diff --git a/core/state/stateupdate.go b/core/state/stateupdate.go index 75c4ca028c..a62e2b2d2d 100644 --- a/core/state/stateupdate.go +++ b/core/state/stateupdate.go @@ -64,8 +64,10 @@ type accountUpdate struct { // execution. It contains information about mutated contract codes, accounts, // and storage slots, along with their original values. type stateUpdate struct { - originRoot common.Hash // hash of the state before applying mutation - root common.Hash // hash of the state after applying mutation + originRoot common.Hash // hash of the state before applying mutation + root common.Hash // hash of the state after applying mutation + blockNumber uint64 // Associated block number + accounts map[common.Hash][]byte // accounts stores mutated accounts in 'slim RLP' encoding accountsOrigin map[common.Address][]byte // accountsOrigin stores the original values of mutated accounts in 'slim RLP' encoding @@ -95,7 +97,7 @@ func (sc *stateUpdate) empty() bool { // // rawStorageKey is a flag indicating whether to use the raw storage slot key or // the hash of the slot key for constructing state update object. -func newStateUpdate(rawStorageKey bool, originRoot common.Hash, root common.Hash, deletes map[common.Hash]*accountDelete, updates map[common.Hash]*accountUpdate, nodes *trienode.MergedNodeSet) *stateUpdate { +func newStateUpdate(rawStorageKey bool, originRoot common.Hash, root common.Hash, blockNumber uint64, deletes map[common.Hash]*accountDelete, updates map[common.Hash]*accountUpdate, nodes *trienode.MergedNodeSet) *stateUpdate { var ( accounts = make(map[common.Hash][]byte) accountsOrigin = make(map[common.Address][]byte) @@ -164,6 +166,7 @@ func newStateUpdate(rawStorageKey bool, originRoot common.Hash, root common.Hash return &stateUpdate{ originRoot: originRoot, root: root, + blockNumber: blockNumber, accounts: accounts, accountsOrigin: accountsOrigin, storages: storages, diff --git a/core/state/trie_prefetcher_test.go b/core/state/trie_prefetcher_test.go index 41349c0c0e..37cd2b6826 100644 --- a/core/state/trie_prefetcher_test.go +++ b/core/state/trie_prefetcher_test.go @@ -73,7 +73,7 @@ func TestVerklePrefetcher(t *testing.T) { state, err := New(types.EmptyRootHash, sdb) if err != nil { - t.Fatalf("failed to initialize state: %v", err) + t.Fatalf("failed to init state: %v", err) } // Create an account and check if the retrieved balance is correct addr := testrand.Address() diff --git a/triedb/database.go b/triedb/database.go index cba35a67e9..d2637bd909 100644 --- a/triedb/database.go +++ b/triedb/database.go @@ -377,10 +377,10 @@ func (db *Database) Disk() ethdb.Database { } // SnapshotCompleted returns the indicator if the snapshot is completed. -func (db *Database) SnapshotCompleted() (common.Hash, bool) { +func (db *Database) SnapshotCompleted() bool { pdb, ok := db.backend.(*pathdb.Database) if !ok { - return common.Hash{}, false + return false } return pdb.SnapshotCompleted() } diff --git a/triedb/pathdb/database.go b/triedb/pathdb/database.go index 0156ddce1b..1592f97d01 100644 --- a/triedb/pathdb/database.go +++ b/triedb/pathdb/database.go @@ -683,13 +683,12 @@ func (db *Database) StorageIterator(root common.Hash, account common.Hash, seek } // SnapshotCompleted returns the snapshot root if the snapshot generation is completed. -func (db *Database) SnapshotCompleted() (common.Hash, bool) { - if db.waitSync { - return common.Hash{}, false +func (db *Database) SnapshotCompleted() bool { + db.lock.RLock() + wait := db.waitSync + db.lock.RUnlock() + if wait { + return false } - dl := db.tree.bottom() - if dl.genComplete() { - return dl.rootHash(), true - } - return common.Hash{}, false + return db.tree.bottom().genComplete() }