From 32970211b6b7e1c4ac4f359b06a31ae94e0f97f7 Mon Sep 17 00:00:00 2001 From: Gary Rong Date: Thu, 1 Aug 2024 10:59:57 +0800 Subject: [PATCH] triedb/pathdb, core: utilize history reader --- core/blockchain_reader.go | 5 ++ core/state/database_archive.go | 124 ++++++++++++++++++++++++++ core/state/statedb.go | 19 ++-- eth/api_backend.go | 10 ++- eth/state_accessor.go | 4 + triedb/database.go | 9 ++ triedb/pathdb/database.go | 4 + triedb/pathdb/history_index_reader.go | 19 ++-- triedb/pathdb/history_indexer.go | 14 +-- triedb/pathdb/reader.go | 75 ++++++++++++++++ 10 files changed, 260 insertions(+), 23 deletions(-) create mode 100644 core/state/database_archive.go diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 5de9de062b..0d7c86d8ae 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -356,6 +356,11 @@ func (bc *BlockChain) StateAt(root common.Hash) (*state.StateDB, error) { return state.New(root, bc.stateDb) } +// ArchiveState returns a new mutable archive state. +func (bc *BlockChain) ArchiveState(root common.Hash) (*state.StateDB, error) { + return state.New(root, state.NewArchiveDatabase(bc.stateDb)) +} + // Config retrieves the chain's fork configuration. func (bc *BlockChain) Config() *params.ChainConfig { return bc.chainConfig } diff --git a/core/state/database_archive.go b/core/state/database_archive.go new file mode 100644 index 0000000000..443f4a17c4 --- /dev/null +++ b/core/state/database_archive.go @@ -0,0 +1,124 @@ +// Copyright 2024 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 ( + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/triedb" + "github.com/ethereum/go-ethereum/triedb/pathdb" +) + +// stateReader wraps a pathdb archive reader. +type archiveReader struct { + reader *pathdb.ArchiveReader +} + +// Account implements Reader, retrieving the account specified by the address. +// +// An error will be returned if the associated snapshot is already stale or +// the requested account is not yet covered by the snapshot. +// +// The returned account might be nil if it's not existent. +func (r *archiveReader) Account(addr common.Address) (*types.StateAccount, error) { + account, err := r.reader.Account(addr) + if err != nil { + return nil, err + } + if account == nil { + return nil, nil + } + acct := &types.StateAccount{ + Nonce: account.Nonce, + Balance: account.Balance, + CodeHash: account.CodeHash, + Root: common.BytesToHash(account.Root), + } + if len(acct.CodeHash) == 0 { + acct.CodeHash = types.EmptyCodeHash.Bytes() + } + if acct.Root == (common.Hash{}) { + acct.Root = types.EmptyRootHash + } + return acct, nil +} + +// Storage implements Reader, retrieving the storage slot specified by the +// address and slot key. +// +// An error will be returned if the associated snapshot is already stale or +// the requested storage slot is not yet covered by the snapshot. +// +// The returned storage slot might be empty if it's not existent. +func (r *archiveReader) Storage(addr common.Address, key common.Hash) (common.Hash, error) { + blob, err := r.reader.Storage(addr, key) + if err != nil { + return common.Hash{}, err + } + if len(blob) == 0 { + return common.Hash{}, nil + } + _, content, _, err := rlp.Split(blob) + if err != nil { + return common.Hash{}, err + } + var slot common.Hash + slot.SetBytes(content) + return slot, nil +} + +// Stats returns the statistics of the reader, specifically detailing the time +// spent on account reading and storage reading. +func (r *archiveReader) Stats() (time.Duration, time.Duration) { return 0, 0 } + +// Copy implements Reader, returning a deep-copied archive readerr. +func (r *archiveReader) Copy() Reader { + return &archiveReader{reader: r.reader} +} + +// ArchiveDB is the implementation of Database interface, with the ability to +// access historical state. +type ArchiveDB struct { + Database + triedb *triedb.Database +} + +// NewArchiveDatabase creates an archive database. +func NewArchiveDatabase(db Database) *ArchiveDB { + return &ArchiveDB{ + Database: db, + triedb: db.TrieDB(), + } +} + +// Reader implements Database interface, returning a reader of the specific state. +func (db *ArchiveDB) Reader(stateRoot common.Hash) (Reader, error) { + // Short circuit if the requested state is available in live database + r, err := db.Database.Reader(stateRoot) + if err == nil { + return r, nil + } + // Construct the archive reader then + hr, err := db.triedb.HistoricReader(stateRoot) + if err != nil { + return nil, err + } + return &archiveReader{reader: hr}, nil +} diff --git a/core/state/statedb.go b/core/state/statedb.go index 09bb1d071a..350467ce7c 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -86,9 +86,9 @@ func (m *mutation) isDelete() bool { type StateDB struct { db Database prefetcher *triePrefetcher - trie Trie logger *tracing.Hooks reader Reader + trie Trie // trie is only resolved when it's accessed // originalRoot is the pre-state root, before any changes were made. // It will be updated when the Commit is called. @@ -167,17 +167,12 @@ type StateDB struct { // New creates a new state from a given trie. func New(root common.Hash, db Database) (*StateDB, error) { - tr, err := db.OpenTrie(root) - if err != nil { - return nil, err - } reader, err := db.Reader(root) if err != nil { return nil, err } return &StateDB{ db: db, - trie: tr, originalRoot: root, reader: reader, stateObjects: make(map[common.Address]*stateObject), @@ -650,7 +645,6 @@ func (s *StateDB) Copy() *StateDB { // Copy all the basic fields, initialize the memory ones state := &StateDB{ db: s.db, - trie: mustCopyTrie(s.trie), reader: s.reader.Copy(), originalRoot: s.originalRoot, stateObjects: make(map[common.Address]*stateObject, len(s.stateObjects)), @@ -676,6 +670,9 @@ func (s *StateDB) Copy() *StateDB { validRevisions: slices.Clone(s.validRevisions), nextRevisionId: s.nextRevisionId, } + if s.trie != nil { + state.trie = mustCopyTrie(s.trie) + } if s.witness != nil { state.witness = s.witness.Copy() } @@ -880,6 +877,14 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash { s.trie = trie } } + if s.trie == nil { + tr, err := s.db.OpenTrie(s.originalRoot) + if err != nil { + s.setError(err) + return common.Hash{} + } + s.trie = tr + } // Perform updates before deletions. This prevents resolution of unnecessary trie nodes // in circumstances similar to the following: // diff --git a/eth/api_backend.go b/eth/api_backend.go index 8a9898b956..faedc3e639 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -205,7 +205,10 @@ func (b *EthAPIBackend) StateAndHeaderByNumber(ctx context.Context, number rpc.B } stateDb, err := b.eth.BlockChain().StateAt(header.Root) if err != nil { - return nil, nil, err + stateDb, err = b.eth.BlockChain().ArchiveState(header.Root) + if err != nil { + return nil, nil, err + } } return stateDb, header, nil } @@ -227,7 +230,10 @@ func (b *EthAPIBackend) StateAndHeaderByNumberOrHash(ctx context.Context, blockN } stateDb, err := b.eth.BlockChain().StateAt(header.Root) if err != nil { - return nil, nil, err + stateDb, err = b.eth.BlockChain().ArchiveState(header.Root) + if err != nil { + return nil, nil, err + } } return stateDb, header, nil } diff --git a/eth/state_accessor.go b/eth/state_accessor.go index 6dfb2a13e6..404e182596 100644 --- a/eth/state_accessor.go +++ b/eth/state_accessor.go @@ -182,6 +182,10 @@ func (eth *Ethereum) pathState(block *types.Block) (*state.StateDB, func(), erro if err == nil { return statedb, noopReleaser, nil } + statedb, err = eth.blockchain.ArchiveState(block.Root()) + if err == nil { + return statedb, noopReleaser, nil + } // TODO historic state is not supported in path-based scheme. // Fully archive node in pbss will be implemented by relying // on state history, but needs more work on top. diff --git a/triedb/database.go b/triedb/database.go index 5facca4667..1c2fbbf2be 100644 --- a/triedb/database.go +++ b/triedb/database.go @@ -350,6 +350,15 @@ func (db *Database) StorageIterator(root common.Hash, account common.Hash, seek return pdb.StorageIterator(root, account, seek) } +// HistoricReader constructs a reader for accessing the requested historic state. +func (db *Database) HistoricReader(root common.Hash) (*pathdb.ArchiveReader, error) { + pdb, ok := db.backend.(*pathdb.Database) + if !ok { + return nil, errors.New("not supported") + } + return pdb.HistoricReader(root) +} + // IsVerkle returns the indicator if the database is holding a verkle tree. func (db *Database) IsVerkle() bool { return db.config.IsVerkle diff --git a/triedb/pathdb/database.go b/triedb/pathdb/database.go index 2c374d97e9..41a486b720 100644 --- a/triedb/pathdb/database.go +++ b/triedb/pathdb/database.go @@ -519,6 +519,10 @@ func (db *Database) Close() error { // Release the memory held by clean cache. db.tree.bottom().resetCache() + // Shutdown background history indexer + if db.indexer != nil { + db.indexer.close() + } // Close the attached state history freezer. if db.freezer == nil { return nil diff --git a/triedb/pathdb/history_index_reader.go b/triedb/pathdb/history_index_reader.go index df6dae8842..fa71a1b388 100644 --- a/triedb/pathdb/history_index_reader.go +++ b/triedb/pathdb/history_index_reader.go @@ -300,13 +300,14 @@ func (r *historyReader) resolve(owner common.Address, state common.Hash, id uint return r.resolveStorage(owner, state, id) } -func (r *historyReader) read(owner common.Address, state common.Hash, id uint64, latest uint64) ([]byte, error) { +func (r *historyReader) read(owner common.Address, state common.Hash, targetID uint64, latestID uint64, latestValue []byte) ([]byte, error) { tail, err := r.freezer.Tail() if err != nil { return nil, err } - // id == tail is allowed, as the first history object preserved is tail+1 - if id < tail { + // targetID == tail is allowed, as the first history object + // available is tail+1 + if targetID < tail { return nil, errors.New("historic state is pruned") } head := rawdb.ReadStateHistoryIndexHead(r.disk) @@ -314,7 +315,7 @@ func (r *historyReader) read(owner common.Address, state common.Hash, id uint64, /* the available range of histories is [tail+1, head] */ - if head == nil || *head <= id { + if head == nil || *head <= targetID { return nil, errors.New("state history is not fully indexed") } ir, ok := r.readers[owner.Hex()+state.Hex()] @@ -325,15 +326,15 @@ func (r *historyReader) read(owner common.Address, state common.Hash, id uint64, } r.readers[owner.Hex()+state.Hex()] = ir } - id, err = ir.readGreaterThan(id) + targetID, err = ir.readGreaterThan(targetID) if err != nil { return nil, err } - if id == math.MaxUint64 { - if *head < latest { + if targetID == math.MaxUint64 { + if *head < latestID { return nil, errors.New("state history is not fully indexed") } - return nil, errors.New("not found") + return latestValue, nil } - return r.resolve(owner, state, id) + return r.resolve(owner, state, targetID) } diff --git a/triedb/pathdb/history_indexer.go b/triedb/pathdb/history_indexer.go index 3a1bdde49e..2175297dc5 100644 --- a/triedb/pathdb/history_indexer.go +++ b/triedb/pathdb/history_indexer.go @@ -112,11 +112,15 @@ func (i *historyIndexer) run(done chan struct{}, head uint64, interrupt *atomic. log.Error("Failed to find next state history for indexing", "err", err) return } - // TODO what if head is lower than the index head. It can - // happen if the entire state history freezer is reset. - //if begin > head { - // - //} + // Short circuit if no indexing tasks left + if begin == head+1 { + return + } + // Deep reorg occurs, TODO put the reorg logic here + if begin > head+1 { + log.Error("Deep reorg detected", "head", head, "next", begin) + return + } log.Info("Start history indexing", "begin", begin, "head", head) var ( diff --git a/triedb/pathdb/reader.go b/triedb/pathdb/reader.go index 120a1538a6..a599b50e0f 100644 --- a/triedb/pathdb/reader.go +++ b/triedb/pathdb/reader.go @@ -21,7 +21,9 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/triedb/database" @@ -147,3 +149,76 @@ func (db *Database) StateReader(root common.Hash) (database.StateReader, error) } return &reader{layer: layer}, nil } + +// ArchiveReader is a wrapper over the history reader, providing access to a +// specific historical state. +type ArchiveReader struct { + db *Database + reader *historyReader + id uint64 +} + +// HistoricReader constructs a reader for accessing the requested historic state. +func (db *Database) HistoricReader(root common.Hash) (*ArchiveReader, error) { + id := rawdb.ReadStateID(db.diskdb, root) + if id == nil { + return nil, fmt.Errorf("state %#x is not available", root) + } + return &ArchiveReader{ + id: *id, + db: db, + reader: newHistoryReader(db.diskdb, db.freezer), + }, nil +} + +// AccountRLP directly retrieves the account RLP associated with a particular +// address in the slim data format. An error will be returned if the read +// operation exits abnormally. Specifically, if the layer is already stale. +// +// Note: +// - the returned account is not a copy, please don't modify it. +// - no error will be returned if the requested account is not found in database. +func (r *ArchiveReader) AccountRLP(address common.Address) ([]byte, error) { + bottom := r.db.tree.bottom() + latest, err := bottom.account(crypto.Keccak256Hash(address.Bytes()), 0) + if err != nil { + return nil, err + } + return r.reader.read(address, common.Hash{}, r.id, bottom.stateID(), latest) +} + +// Account directly retrieves the account associated with a particular address in +// the slim data format. An error will be returned if the read operation exits +// abnormally. Specifically, if the layer is already stale. +// +// No error will be returned if the requested account is not found in database +func (r *ArchiveReader) Account(address common.Address) (*types.SlimAccount, error) { + blob, err := r.AccountRLP(address) + if err != nil { + return nil, err + } + if len(blob) == 0 { + return nil, nil + } + account := new(types.SlimAccount) + if err := rlp.DecodeBytes(blob, account); err != nil { + panic(err) + } + return account, nil +} + +// Storage directly retrieves the storage data associated with a particular key, +// within a particular account. An error will be returned if the read operation +// exits abnormally. Specifically, if the layer is already stale. +// +// Note: +// - the returned storage data is not a copy, please don't modify it. +// - no error will be returned if the requested slot is not found in database. +func (r *ArchiveReader) Storage(address common.Address, key common.Hash) ([]byte, error) { + bottom := r.db.tree.bottom() + latest, err := bottom.storage(crypto.Keccak256Hash(address.Bytes()), crypto.Keccak256Hash(key.Bytes()), 0) + if err != nil { + return nil, err + } + return r.reader.read(address, crypto.Keccak256Hash(key.Bytes()), r.id, bottom.stateID(), latest) +}