mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-23 13:16:42 +00:00
triedb/pathdb: generalize the history indexer
It's a pre-requisite for shipping full archive node, reusing the existing history indexing mechanism for trienode history.
This commit is contained in:
parent
6a7f64e760
commit
1533096f4b
11 changed files with 465 additions and 342 deletions
|
|
@ -189,7 +189,7 @@ func New(diskdb ethdb.Database, config *Config, isVerkle bool) *Database {
|
||||||
}
|
}
|
||||||
// TODO (rjl493456442) disable the background indexing in read-only mode
|
// TODO (rjl493456442) disable the background indexing in read-only mode
|
||||||
if db.stateFreezer != nil && db.config.EnableStateIndexing {
|
if db.stateFreezer != nil && db.config.EnableStateIndexing {
|
||||||
db.stateIndexer = newHistoryIndexer(db.diskdb, db.stateFreezer, db.tree.bottom().stateID())
|
db.stateIndexer = newHistoryIndexer(db.diskdb, db.stateFreezer, db.tree.bottom().stateID(), typeStateHistory)
|
||||||
log.Info("Enabled state history indexing")
|
log.Info("Enabled state history indexing")
|
||||||
}
|
}
|
||||||
fields := config.fields()
|
fields := config.fields()
|
||||||
|
|
@ -245,7 +245,7 @@ func (db *Database) repairHistory() error {
|
||||||
}
|
}
|
||||||
// Truncate the extra state histories above in freezer in case it's not
|
// Truncate the extra state histories above in freezer in case it's not
|
||||||
// aligned with the disk layer. It might happen after a unclean shutdown.
|
// aligned with the disk layer. It might happen after a unclean shutdown.
|
||||||
pruned, err := truncateFromHead(db.stateFreezer, id)
|
pruned, err := truncateFromHead(db.stateFreezer, typeStateHistory, id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Crit("Failed to truncate extra state histories", "err", err)
|
log.Crit("Failed to truncate extra state histories", "err", err)
|
||||||
}
|
}
|
||||||
|
|
@ -448,7 +448,7 @@ func (db *Database) Enable(root common.Hash) error {
|
||||||
// 2. Re-initialize the indexer so it starts indexing from the new state root.
|
// 2. Re-initialize the indexer so it starts indexing from the new state root.
|
||||||
if db.stateIndexer != nil && db.stateFreezer != nil && db.config.EnableStateIndexing {
|
if db.stateIndexer != nil && db.stateFreezer != nil && db.config.EnableStateIndexing {
|
||||||
db.stateIndexer.close()
|
db.stateIndexer.close()
|
||||||
db.stateIndexer = newHistoryIndexer(db.diskdb, db.stateFreezer, db.tree.bottom().stateID())
|
db.stateIndexer = newHistoryIndexer(db.diskdb, db.stateFreezer, db.tree.bottom().stateID(), typeStateHistory)
|
||||||
log.Info("Re-enabled state history indexing")
|
log.Info("Re-enabled state history indexing")
|
||||||
}
|
}
|
||||||
log.Info("Rebuilt trie database", "root", root)
|
log.Info("Rebuilt trie database", "root", root)
|
||||||
|
|
@ -502,7 +502,7 @@ func (db *Database) Recover(root common.Hash) error {
|
||||||
if err := db.diskdb.SyncKeyValue(); err != nil {
|
if err := db.diskdb.SyncKeyValue(); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
_, err := truncateFromHead(db.stateFreezer, dl.stateID())
|
_, err := truncateFromHead(db.stateFreezer, typeStateHistory, dl.stateID())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -378,7 +378,7 @@ func (dl *diskLayer) writeStateHistory(diff *diffLayer) (bool, error) {
|
||||||
log.Debug("Skip tail truncation", "persistentID", persistentID, "tailID", tail+1, "headID", diff.stateID(), "limit", limit)
|
log.Debug("Skip tail truncation", "persistentID", persistentID, "tailID", tail+1, "headID", diff.stateID(), "limit", limit)
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
pruned, err := truncateFromTail(dl.db.stateFreezer, newFirst-1)
|
pruned, err := truncateFromTail(dl.db.stateFreezer, typeStateHistory, newFirst-1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -19,11 +19,147 @@ package pathdb
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"iter"
|
||||||
|
|
||||||
|
"github.com/ethereum/go-ethereum/common"
|
||||||
"github.com/ethereum/go-ethereum/ethdb"
|
"github.com/ethereum/go-ethereum/ethdb"
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// historyType represents the category of historical data.
|
||||||
|
type historyType uint8
|
||||||
|
|
||||||
|
const (
|
||||||
|
// typeStateHistory indicates history data related to account or storage changes.
|
||||||
|
typeStateHistory historyType = 0
|
||||||
|
)
|
||||||
|
|
||||||
|
// String returns the string format representation.
|
||||||
|
func (h historyType) String() string {
|
||||||
|
switch h {
|
||||||
|
case typeStateHistory:
|
||||||
|
return "state"
|
||||||
|
default:
|
||||||
|
return fmt.Sprintf("unknown type: %d", h)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// elementType represents the category of state element.
|
||||||
|
type elementType uint8
|
||||||
|
|
||||||
|
const (
|
||||||
|
typeAccount elementType = 0 // represents the account data
|
||||||
|
typeStorage elementType = 1 // represents the storage slot data
|
||||||
|
)
|
||||||
|
|
||||||
|
// String returns the string format representation.
|
||||||
|
func (e elementType) String() string {
|
||||||
|
switch e {
|
||||||
|
case typeAccount:
|
||||||
|
return "account"
|
||||||
|
case typeStorage:
|
||||||
|
return "storage"
|
||||||
|
default:
|
||||||
|
return fmt.Sprintf("unknown element type: %d", e)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// toHistoryType maps an element type to its corresponding history type.
|
||||||
|
func toHistoryType(typ elementType) historyType {
|
||||||
|
if typ == typeAccount || typ == typeStorage {
|
||||||
|
return typeStateHistory
|
||||||
|
}
|
||||||
|
panic(fmt.Sprintf("unknown element type %v", typ))
|
||||||
|
}
|
||||||
|
|
||||||
|
// stateIdent represents the identifier of a state element, which can be
|
||||||
|
// an account or a storage slot.
|
||||||
|
type stateIdent struct {
|
||||||
|
typ elementType
|
||||||
|
|
||||||
|
// The hash of the account address. This is used instead of the raw account
|
||||||
|
// address is to align the traversal order with the Merkle-Patricia-Trie.
|
||||||
|
addressHash common.Hash
|
||||||
|
|
||||||
|
// The hash of the storage slot key. This is used instead of the raw slot key
|
||||||
|
// because, in legacy state histories (prior to the Cancun fork), the slot
|
||||||
|
// identifier is the hash of the key, and the original key (preimage) cannot
|
||||||
|
// be recovered. To maintain backward compatibility, the key hash is used.
|
||||||
|
//
|
||||||
|
// Meanwhile, using the storage key hash also preserve the traversal order
|
||||||
|
// with Merkle-Patricia-Trie.
|
||||||
|
//
|
||||||
|
// This field is null if the identifier refers to an account or a trie node.
|
||||||
|
storageHash common.Hash
|
||||||
|
}
|
||||||
|
|
||||||
|
// String returns the string format state identifier.
|
||||||
|
func (ident stateIdent) String() string {
|
||||||
|
if ident.typ == typeAccount {
|
||||||
|
return ident.addressHash.Hex()
|
||||||
|
}
|
||||||
|
return ident.addressHash.Hex() + ident.storageHash.Hex()
|
||||||
|
}
|
||||||
|
|
||||||
|
// newAccountIdent constructs a state identifier for an account.
|
||||||
|
func newAccountIdent(addressHash common.Hash) stateIdent {
|
||||||
|
return stateIdent{
|
||||||
|
typ: typeAccount,
|
||||||
|
addressHash: addressHash,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// newStorageIdent constructs a state identifier for a storage slot.
|
||||||
|
// The address denotes the address hash of the associated account;
|
||||||
|
// the storageHash denotes the hash of the raw storage slot key;
|
||||||
|
func newStorageIdent(addressHash common.Hash, storageHash common.Hash) stateIdent {
|
||||||
|
return stateIdent{
|
||||||
|
typ: typeStorage,
|
||||||
|
addressHash: addressHash,
|
||||||
|
storageHash: storageHash,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// stateIdentQuery is the extension of stateIdent by adding the account address
|
||||||
|
// and raw storage key.
|
||||||
|
type stateIdentQuery struct {
|
||||||
|
stateIdent
|
||||||
|
|
||||||
|
address common.Address
|
||||||
|
storageKey common.Hash
|
||||||
|
}
|
||||||
|
|
||||||
|
// newAccountIdentQuery constructs a state identifier for an account.
|
||||||
|
func newAccountIdentQuery(address common.Address, addressHash common.Hash) stateIdentQuery {
|
||||||
|
return stateIdentQuery{
|
||||||
|
stateIdent: newAccountIdent(addressHash),
|
||||||
|
address: address,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// newStorageIdentQuery constructs a state identifier for a storage slot.
|
||||||
|
// the address denotes the address of the associated account;
|
||||||
|
// the addressHash denotes the address hash of the associated account;
|
||||||
|
// the storageKey denotes the raw storage slot key;
|
||||||
|
// the storageHash denotes the hash of the raw storage slot key;
|
||||||
|
func newStorageIdentQuery(address common.Address, addressHash common.Hash, storageKey common.Hash, storageHash common.Hash) stateIdentQuery {
|
||||||
|
return stateIdentQuery{
|
||||||
|
stateIdent: newStorageIdent(addressHash, storageHash),
|
||||||
|
address: address,
|
||||||
|
storageKey: storageKey,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// history defines the interface of historical data, implemented by stateHistory
|
||||||
|
// and trienodeHistory (in the near future).
|
||||||
|
type history interface {
|
||||||
|
// typ returns the historical data type held in the history.
|
||||||
|
typ() historyType
|
||||||
|
|
||||||
|
// forEach returns an iterator to traverse the state entries in the history.
|
||||||
|
forEach() iter.Seq[stateIdent]
|
||||||
|
}
|
||||||
|
|
||||||
var (
|
var (
|
||||||
errHeadTruncationOutOfRange = errors.New("history head truncation out of range")
|
errHeadTruncationOutOfRange = errors.New("history head truncation out of range")
|
||||||
errTailTruncationOutOfRange = errors.New("history tail truncation out of range")
|
errTailTruncationOutOfRange = errors.New("history tail truncation out of range")
|
||||||
|
|
@ -31,7 +167,7 @@ var (
|
||||||
|
|
||||||
// truncateFromHead removes excess elements from the head of the freezer based
|
// truncateFromHead removes excess elements from the head of the freezer based
|
||||||
// on the given parameters. It returns the number of items that were removed.
|
// on the given parameters. It returns the number of items that were removed.
|
||||||
func truncateFromHead(store ethdb.AncientStore, nhead uint64) (int, error) {
|
func truncateFromHead(store ethdb.AncientStore, typ historyType, nhead uint64) (int, error) {
|
||||||
ohead, err := store.Ancients()
|
ohead, err := store.Ancients()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, err
|
return 0, err
|
||||||
|
|
@ -40,11 +176,11 @@ func truncateFromHead(store ethdb.AncientStore, nhead uint64) (int, error) {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, err
|
return 0, err
|
||||||
}
|
}
|
||||||
log.Info("Truncating from head", "ohead", ohead, "tail", otail, "nhead", nhead)
|
log.Info("Truncating from head", "type", typ.String(), "ohead", ohead, "tail", otail, "nhead", nhead)
|
||||||
|
|
||||||
// Ensure that the truncation target falls within the valid range.
|
// Ensure that the truncation target falls within the valid range.
|
||||||
if ohead < nhead || nhead < otail {
|
if ohead < nhead || nhead < otail {
|
||||||
return 0, fmt.Errorf("%w, tail: %d, head: %d, target: %d", errHeadTruncationOutOfRange, otail, ohead, nhead)
|
return 0, fmt.Errorf("%w, %s, tail: %d, head: %d, target: %d", errHeadTruncationOutOfRange, typ, otail, ohead, nhead)
|
||||||
}
|
}
|
||||||
// Short circuit if nothing to truncate.
|
// Short circuit if nothing to truncate.
|
||||||
if ohead == nhead {
|
if ohead == nhead {
|
||||||
|
|
@ -61,7 +197,7 @@ func truncateFromHead(store ethdb.AncientStore, nhead uint64) (int, error) {
|
||||||
|
|
||||||
// truncateFromTail removes excess elements from the end of the freezer based
|
// truncateFromTail removes excess elements from the end of the freezer based
|
||||||
// on the given parameters. It returns the number of items that were removed.
|
// on the given parameters. It returns the number of items that were removed.
|
||||||
func truncateFromTail(store ethdb.AncientStore, ntail uint64) (int, error) {
|
func truncateFromTail(store ethdb.AncientStore, typ historyType, ntail uint64) (int, error) {
|
||||||
ohead, err := store.Ancients()
|
ohead, err := store.Ancients()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, err
|
return 0, err
|
||||||
|
|
@ -72,7 +208,7 @@ func truncateFromTail(store ethdb.AncientStore, ntail uint64) (int, error) {
|
||||||
}
|
}
|
||||||
// Ensure that the truncation target falls within the valid range.
|
// Ensure that the truncation target falls within the valid range.
|
||||||
if otail > ntail || ntail > ohead {
|
if otail > ntail || ntail > ohead {
|
||||||
return 0, fmt.Errorf("%w, tail: %d, head: %d, target: %d", errTailTruncationOutOfRange, otail, ohead, ntail)
|
return 0, fmt.Errorf("%w, %s, tail: %d, head: %d, target: %d", errTailTruncationOutOfRange, typ, otail, ohead, ntail)
|
||||||
}
|
}
|
||||||
// Short circuit if nothing to truncate.
|
// Short circuit if nothing to truncate.
|
||||||
if otail == ntail {
|
if otail == ntail {
|
||||||
|
|
|
||||||
|
|
@ -78,12 +78,7 @@ type indexReader struct {
|
||||||
|
|
||||||
// loadIndexData loads the index data associated with the specified state.
|
// loadIndexData loads the index data associated with the specified state.
|
||||||
func loadIndexData(db ethdb.KeyValueReader, state stateIdent) ([]*indexBlockDesc, error) {
|
func loadIndexData(db ethdb.KeyValueReader, state stateIdent) ([]*indexBlockDesc, error) {
|
||||||
var blob []byte
|
blob := readStateIndex(state, db)
|
||||||
if state.account {
|
|
||||||
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
|
||||||
} else {
|
|
||||||
blob = rawdb.ReadStorageHistoryIndex(db, state.addressHash, state.storageHash)
|
|
||||||
}
|
|
||||||
if len(blob) == 0 {
|
if len(blob) == 0 {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
@ -137,15 +132,8 @@ func (r *indexReader) readGreaterThan(id uint64) (uint64, error) {
|
||||||
|
|
||||||
br, ok := r.readers[desc.id]
|
br, ok := r.readers[desc.id]
|
||||||
if !ok {
|
if !ok {
|
||||||
var (
|
var err error
|
||||||
err error
|
blob := readStateIndexBlock(r.state, r.db, desc.id)
|
||||||
blob []byte
|
|
||||||
)
|
|
||||||
if r.state.account {
|
|
||||||
blob = rawdb.ReadAccountHistoryIndexBlock(r.db, r.state.addressHash, desc.id)
|
|
||||||
} else {
|
|
||||||
blob = rawdb.ReadStorageHistoryIndexBlock(r.db, r.state.addressHash, r.state.storageHash, desc.id)
|
|
||||||
}
|
|
||||||
br, err = newBlockReader(blob)
|
br, err = newBlockReader(blob)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, err
|
return 0, err
|
||||||
|
|
@ -174,12 +162,7 @@ type indexWriter struct {
|
||||||
|
|
||||||
// newIndexWriter constructs the index writer for the specified state.
|
// newIndexWriter constructs the index writer for the specified state.
|
||||||
func newIndexWriter(db ethdb.KeyValueReader, state stateIdent) (*indexWriter, error) {
|
func newIndexWriter(db ethdb.KeyValueReader, state stateIdent) (*indexWriter, error) {
|
||||||
var blob []byte
|
blob := readStateIndex(state, db)
|
||||||
if state.account {
|
|
||||||
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
|
||||||
} else {
|
|
||||||
blob = rawdb.ReadStorageHistoryIndex(db, state.addressHash, state.storageHash)
|
|
||||||
}
|
|
||||||
if len(blob) == 0 {
|
if len(blob) == 0 {
|
||||||
desc := newIndexBlockDesc(0)
|
desc := newIndexBlockDesc(0)
|
||||||
bw, _ := newBlockWriter(nil, desc)
|
bw, _ := newBlockWriter(nil, desc)
|
||||||
|
|
@ -194,15 +177,8 @@ func newIndexWriter(db ethdb.KeyValueReader, state stateIdent) (*indexWriter, er
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
var (
|
lastDesc := descList[len(descList)-1]
|
||||||
indexBlock []byte
|
indexBlock := readStateIndexBlock(state, db, lastDesc.id)
|
||||||
lastDesc = descList[len(descList)-1]
|
|
||||||
)
|
|
||||||
if state.account {
|
|
||||||
indexBlock = rawdb.ReadAccountHistoryIndexBlock(db, state.addressHash, lastDesc.id)
|
|
||||||
} else {
|
|
||||||
indexBlock = rawdb.ReadStorageHistoryIndexBlock(db, state.addressHash, state.storageHash, lastDesc.id)
|
|
||||||
}
|
|
||||||
bw, err := newBlockWriter(indexBlock, lastDesc)
|
bw, err := newBlockWriter(indexBlock, lastDesc)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|
@ -270,11 +246,7 @@ func (w *indexWriter) finish(batch ethdb.Batch) {
|
||||||
return // nothing to commit
|
return // nothing to commit
|
||||||
}
|
}
|
||||||
for _, bw := range writers {
|
for _, bw := range writers {
|
||||||
if w.state.account {
|
writeStateIndexBlock(w.state, batch, bw.desc.id, bw.finish())
|
||||||
rawdb.WriteAccountHistoryIndexBlock(batch, w.state.addressHash, bw.desc.id, bw.finish())
|
|
||||||
} else {
|
|
||||||
rawdb.WriteStorageHistoryIndexBlock(batch, w.state.addressHash, w.state.storageHash, bw.desc.id, bw.finish())
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
w.frozen = nil // release all the frozen writers
|
w.frozen = nil // release all the frozen writers
|
||||||
|
|
||||||
|
|
@ -282,11 +254,7 @@ func (w *indexWriter) finish(batch ethdb.Batch) {
|
||||||
for _, desc := range descList {
|
for _, desc := range descList {
|
||||||
buf = append(buf, desc.encode()...)
|
buf = append(buf, desc.encode()...)
|
||||||
}
|
}
|
||||||
if w.state.account {
|
writeStateIndex(w.state, batch, buf)
|
||||||
rawdb.WriteAccountHistoryIndex(batch, w.state.addressHash, buf)
|
|
||||||
} else {
|
|
||||||
rawdb.WriteStorageHistoryIndex(batch, w.state.addressHash, w.state.storageHash, buf)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// indexDeleter is responsible for deleting index data for a specific state.
|
// indexDeleter is responsible for deleting index data for a specific state.
|
||||||
|
|
@ -301,12 +269,7 @@ type indexDeleter struct {
|
||||||
|
|
||||||
// newIndexDeleter constructs the index deleter for the specified state.
|
// newIndexDeleter constructs the index deleter for the specified state.
|
||||||
func newIndexDeleter(db ethdb.KeyValueReader, state stateIdent) (*indexDeleter, error) {
|
func newIndexDeleter(db ethdb.KeyValueReader, state stateIdent) (*indexDeleter, error) {
|
||||||
var blob []byte
|
blob := readStateIndex(state, db)
|
||||||
if state.account {
|
|
||||||
blob = rawdb.ReadAccountHistoryIndex(db, state.addressHash)
|
|
||||||
} else {
|
|
||||||
blob = rawdb.ReadStorageHistoryIndex(db, state.addressHash, state.storageHash)
|
|
||||||
}
|
|
||||||
if len(blob) == 0 {
|
if len(blob) == 0 {
|
||||||
// TODO(rjl493456442) we can probably return an error here,
|
// TODO(rjl493456442) we can probably return an error here,
|
||||||
// deleter with no data is meaningless.
|
// deleter with no data is meaningless.
|
||||||
|
|
@ -323,15 +286,8 @@ func newIndexDeleter(db ethdb.KeyValueReader, state stateIdent) (*indexDeleter,
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
var (
|
lastDesc := descList[len(descList)-1]
|
||||||
indexBlock []byte
|
indexBlock := readStateIndexBlock(state, db, lastDesc.id)
|
||||||
lastDesc = descList[len(descList)-1]
|
|
||||||
)
|
|
||||||
if state.account {
|
|
||||||
indexBlock = rawdb.ReadAccountHistoryIndexBlock(db, state.addressHash, lastDesc.id)
|
|
||||||
} else {
|
|
||||||
indexBlock = rawdb.ReadStorageHistoryIndexBlock(db, state.addressHash, state.storageHash, lastDesc.id)
|
|
||||||
}
|
|
||||||
bw, err := newBlockWriter(indexBlock, lastDesc)
|
bw, err := newBlockWriter(indexBlock, lastDesc)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|
@ -376,15 +332,8 @@ func (d *indexDeleter) pop(id uint64) error {
|
||||||
d.descList = d.descList[:len(d.descList)-1]
|
d.descList = d.descList[:len(d.descList)-1]
|
||||||
|
|
||||||
// Open the previous block writer for deleting
|
// Open the previous block writer for deleting
|
||||||
var (
|
lastDesc := d.descList[len(d.descList)-1]
|
||||||
indexBlock []byte
|
indexBlock := readStateIndexBlock(d.state, d.db, lastDesc.id)
|
||||||
lastDesc = d.descList[len(d.descList)-1]
|
|
||||||
)
|
|
||||||
if d.state.account {
|
|
||||||
indexBlock = rawdb.ReadAccountHistoryIndexBlock(d.db, d.state.addressHash, lastDesc.id)
|
|
||||||
} else {
|
|
||||||
indexBlock = rawdb.ReadStorageHistoryIndexBlock(d.db, d.state.addressHash, d.state.storageHash, lastDesc.id)
|
|
||||||
}
|
|
||||||
bw, err := newBlockWriter(indexBlock, lastDesc)
|
bw, err := newBlockWriter(indexBlock, lastDesc)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
|
|
@ -399,38 +348,100 @@ func (d *indexDeleter) pop(id uint64) error {
|
||||||
// This function is safe to be called multiple times.
|
// This function is safe to be called multiple times.
|
||||||
func (d *indexDeleter) finish(batch ethdb.Batch) {
|
func (d *indexDeleter) finish(batch ethdb.Batch) {
|
||||||
for _, id := range d.dropped {
|
for _, id := range d.dropped {
|
||||||
if d.state.account {
|
deleteStateIndexBlock(d.state, batch, id)
|
||||||
rawdb.DeleteAccountHistoryIndexBlock(batch, d.state.addressHash, id)
|
|
||||||
} else {
|
|
||||||
rawdb.DeleteStorageHistoryIndexBlock(batch, d.state.addressHash, d.state.storageHash, id)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
d.dropped = nil
|
d.dropped = nil
|
||||||
|
|
||||||
// Flush the content of last block writer, regardless it's dirty or not
|
// Flush the content of last block writer, regardless it's dirty or not
|
||||||
if !d.bw.empty() {
|
if !d.bw.empty() {
|
||||||
if d.state.account {
|
writeStateIndexBlock(d.state, batch, d.bw.desc.id, d.bw.finish())
|
||||||
rawdb.WriteAccountHistoryIndexBlock(batch, d.state.addressHash, d.bw.desc.id, d.bw.finish())
|
|
||||||
} else {
|
|
||||||
rawdb.WriteStorageHistoryIndexBlock(batch, d.state.addressHash, d.state.storageHash, d.bw.desc.id, d.bw.finish())
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
// Flush the index metadata into the supplied batch
|
// Flush the index metadata into the supplied batch
|
||||||
if d.empty() {
|
if d.empty() {
|
||||||
if d.state.account {
|
deleteStateIndex(d.state, batch)
|
||||||
rawdb.DeleteAccountHistoryIndex(batch, d.state.addressHash)
|
|
||||||
} else {
|
|
||||||
rawdb.DeleteStorageHistoryIndex(batch, d.state.addressHash, d.state.storageHash)
|
|
||||||
}
|
|
||||||
} else {
|
} else {
|
||||||
buf := make([]byte, 0, indexBlockDescSize*len(d.descList))
|
buf := make([]byte, 0, indexBlockDescSize*len(d.descList))
|
||||||
for _, desc := range d.descList {
|
for _, desc := range d.descList {
|
||||||
buf = append(buf, desc.encode()...)
|
buf = append(buf, desc.encode()...)
|
||||||
}
|
}
|
||||||
if d.state.account {
|
writeStateIndex(d.state, batch, buf)
|
||||||
rawdb.WriteAccountHistoryIndex(batch, d.state.addressHash, buf)
|
|
||||||
} else {
|
|
||||||
rawdb.WriteStorageHistoryIndex(batch, d.state.addressHash, d.state.storageHash, buf)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// readStateIndex retrieves the index metadata for the given state identifier.
|
||||||
|
// This function is shared by accounts and storage slots.
|
||||||
|
func readStateIndex(ident stateIdent, db ethdb.KeyValueReader) []byte {
|
||||||
|
switch ident.typ {
|
||||||
|
case typeAccount:
|
||||||
|
return rawdb.ReadAccountHistoryIndex(db, ident.addressHash)
|
||||||
|
case typeStorage:
|
||||||
|
return rawdb.ReadStorageHistoryIndex(db, ident.addressHash, ident.storageHash)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown type: %v", ident.typ))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// writeStateIndex writes the provided index metadata into database with the
|
||||||
|
// given state identifier. This function is shared by accounts and storage slots.
|
||||||
|
func writeStateIndex(ident stateIdent, db ethdb.KeyValueWriter, data []byte) {
|
||||||
|
switch ident.typ {
|
||||||
|
case typeAccount:
|
||||||
|
rawdb.WriteAccountHistoryIndex(db, ident.addressHash, data)
|
||||||
|
case typeStorage:
|
||||||
|
rawdb.WriteStorageHistoryIndex(db, ident.addressHash, ident.storageHash, data)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown type: %v", ident.typ))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// deleteStateIndex removes the index metadata for the given state identifier.
|
||||||
|
// This function is shared by accounts and storage slots.
|
||||||
|
func deleteStateIndex(ident stateIdent, db ethdb.KeyValueWriter) {
|
||||||
|
switch ident.typ {
|
||||||
|
case typeAccount:
|
||||||
|
rawdb.DeleteAccountHistoryIndex(db, ident.addressHash)
|
||||||
|
case typeStorage:
|
||||||
|
rawdb.DeleteStorageHistoryIndex(db, ident.addressHash, ident.storageHash)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown type: %v", ident.typ))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// readStateIndexBlock retrieves the index block for the given state identifier
|
||||||
|
// and block ID. This function is shared by accounts and storage slots.
|
||||||
|
func readStateIndexBlock(ident stateIdent, db ethdb.KeyValueReader, id uint32) []byte {
|
||||||
|
switch ident.typ {
|
||||||
|
case typeAccount:
|
||||||
|
return rawdb.ReadAccountHistoryIndexBlock(db, ident.addressHash, id)
|
||||||
|
case typeStorage:
|
||||||
|
return rawdb.ReadStorageHistoryIndexBlock(db, ident.addressHash, ident.storageHash, id)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown type: %v", ident.typ))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// writeStateIndexBlock writes the provided index block into database with the
|
||||||
|
// given state identifier. This function is shared by accounts and storage slots.
|
||||||
|
func writeStateIndexBlock(ident stateIdent, db ethdb.KeyValueWriter, id uint32, data []byte) {
|
||||||
|
switch ident.typ {
|
||||||
|
case typeAccount:
|
||||||
|
rawdb.WriteAccountHistoryIndexBlock(db, ident.addressHash, id, data)
|
||||||
|
case typeStorage:
|
||||||
|
rawdb.WriteStorageHistoryIndexBlock(db, ident.addressHash, ident.storageHash, id, data)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown type: %v", ident.typ))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// deleteStateIndexBlock removes the index block from database with the given
|
||||||
|
// state identifier. This function is shared by accounts and storage slots.
|
||||||
|
func deleteStateIndexBlock(ident stateIdent, db ethdb.KeyValueWriter, id uint32) {
|
||||||
|
switch ident.typ {
|
||||||
|
case typeAccount:
|
||||||
|
rawdb.DeleteAccountHistoryIndexBlock(db, ident.addressHash, id)
|
||||||
|
case typeStorage:
|
||||||
|
rawdb.DeleteStorageHistoryIndexBlock(db, ident.addressHash, ident.storageHash, id)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown type: %v", ident.typ))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -179,7 +179,7 @@ func TestIndexWriterDelete(t *testing.T) {
|
||||||
func TestBatchIndexerWrite(t *testing.T) {
|
func TestBatchIndexerWrite(t *testing.T) {
|
||||||
var (
|
var (
|
||||||
db = rawdb.NewMemoryDatabase()
|
db = rawdb.NewMemoryDatabase()
|
||||||
batch = newBatchIndexer(db, false)
|
batch = newBatchIndexer(db, false, typeStateHistory)
|
||||||
histories = makeStateHistories(10)
|
histories = makeStateHistories(10)
|
||||||
)
|
)
|
||||||
for i, h := range histories {
|
for i, h := range histories {
|
||||||
|
|
@ -190,7 +190,7 @@ func TestBatchIndexerWrite(t *testing.T) {
|
||||||
if err := batch.finish(true); err != nil {
|
if err := batch.finish(true); err != nil {
|
||||||
t.Fatalf("Failed to finish batch indexer, %v", err)
|
t.Fatalf("Failed to finish batch indexer, %v", err)
|
||||||
}
|
}
|
||||||
metadata := loadIndexMetadata(db)
|
metadata := loadIndexMetadata(db, typeStateHistory)
|
||||||
if metadata == nil || metadata.Last != uint64(10) {
|
if metadata == nil || metadata.Last != uint64(10) {
|
||||||
t.Fatal("Unexpected index position")
|
t.Fatal("Unexpected index position")
|
||||||
}
|
}
|
||||||
|
|
@ -256,7 +256,7 @@ func TestBatchIndexerWrite(t *testing.T) {
|
||||||
func TestBatchIndexerDelete(t *testing.T) {
|
func TestBatchIndexerDelete(t *testing.T) {
|
||||||
var (
|
var (
|
||||||
db = rawdb.NewMemoryDatabase()
|
db = rawdb.NewMemoryDatabase()
|
||||||
bw = newBatchIndexer(db, false)
|
bw = newBatchIndexer(db, false, typeStateHistory)
|
||||||
histories = makeStateHistories(10)
|
histories = makeStateHistories(10)
|
||||||
)
|
)
|
||||||
// Index histories
|
// Index histories
|
||||||
|
|
@ -270,7 +270,7 @@ func TestBatchIndexerDelete(t *testing.T) {
|
||||||
}
|
}
|
||||||
|
|
||||||
// Unindex histories
|
// Unindex histories
|
||||||
bd := newBatchIndexer(db, true)
|
bd := newBatchIndexer(db, true, typeStateHistory)
|
||||||
for i := len(histories) - 1; i >= 0; i-- {
|
for i := len(histories) - 1; i >= 0; i-- {
|
||||||
if err := bd.process(histories[i], uint64(i+1)); err != nil {
|
if err := bd.process(histories[i], uint64(i+1)); err != nil {
|
||||||
t.Fatalf("Failed to process history, %v", err)
|
t.Fatalf("Failed to process history, %v", err)
|
||||||
|
|
@ -280,7 +280,7 @@ func TestBatchIndexerDelete(t *testing.T) {
|
||||||
t.Fatalf("Failed to finish batch indexer, %v", err)
|
t.Fatalf("Failed to finish batch indexer, %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
metadata := loadIndexMetadata(db)
|
metadata := loadIndexMetadata(db, typeStateHistory)
|
||||||
if metadata != nil {
|
if metadata != nil {
|
||||||
t.Fatal("Unexpected index position")
|
t.Fatal("Unexpected index position")
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -26,7 +26,6 @@ import (
|
||||||
|
|
||||||
"github.com/ethereum/go-ethereum/common"
|
"github.com/ethereum/go-ethereum/common"
|
||||||
"github.com/ethereum/go-ethereum/core/rawdb"
|
"github.com/ethereum/go-ethereum/core/rawdb"
|
||||||
"github.com/ethereum/go-ethereum/crypto"
|
|
||||||
"github.com/ethereum/go-ethereum/ethdb"
|
"github.com/ethereum/go-ethereum/ethdb"
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
"github.com/ethereum/go-ethereum/rlp"
|
"github.com/ethereum/go-ethereum/rlp"
|
||||||
|
|
@ -41,13 +40,32 @@ const (
|
||||||
stateIndexVersion = stateIndexV0 // the current state index version
|
stateIndexVersion = stateIndexV0 // the current state index version
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// indexVersion returns the latest index version for the given history type.
|
||||||
|
// It panics if the history type is unknown.
|
||||||
|
func indexVersion(typ historyType) uint8 {
|
||||||
|
switch typ {
|
||||||
|
case typeStateHistory:
|
||||||
|
return stateIndexVersion
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown history type: %d", typ))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// indexMetadata describes the metadata of the historical data index.
|
||||||
type indexMetadata struct {
|
type indexMetadata struct {
|
||||||
Version uint8
|
Version uint8
|
||||||
Last uint64
|
Last uint64
|
||||||
}
|
}
|
||||||
|
|
||||||
func loadIndexMetadata(db ethdb.KeyValueReader) *indexMetadata {
|
// loadIndexMetadata reads the metadata of the specific history index.
|
||||||
blob := rawdb.ReadStateHistoryIndexMetadata(db)
|
func loadIndexMetadata(db ethdb.KeyValueReader, typ historyType) *indexMetadata {
|
||||||
|
var blob []byte
|
||||||
|
switch typ {
|
||||||
|
case typeStateHistory:
|
||||||
|
blob = rawdb.ReadStateHistoryIndexMetadata(db)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown history type %d", typ))
|
||||||
|
}
|
||||||
if len(blob) == 0 {
|
if len(blob) == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -59,91 +77,94 @@ func loadIndexMetadata(db ethdb.KeyValueReader) *indexMetadata {
|
||||||
return &m
|
return &m
|
||||||
}
|
}
|
||||||
|
|
||||||
func storeIndexMetadata(db ethdb.KeyValueWriter, last uint64) {
|
// storeIndexMetadata stores the metadata of the specific history index.
|
||||||
var m indexMetadata
|
func storeIndexMetadata(db ethdb.KeyValueWriter, typ historyType, last uint64) {
|
||||||
m.Version = stateIndexVersion
|
m := indexMetadata{
|
||||||
m.Last = last
|
Version: indexVersion(typ),
|
||||||
|
Last: last,
|
||||||
|
}
|
||||||
blob, err := rlp.EncodeToBytes(m)
|
blob, err := rlp.EncodeToBytes(m)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Crit("Failed to encode index metadata", "err", err)
|
panic(fmt.Errorf("fail to encode index metadata, %v", err))
|
||||||
}
|
}
|
||||||
|
switch typ {
|
||||||
|
case typeStateHistory:
|
||||||
rawdb.WriteStateHistoryIndexMetadata(db, blob)
|
rawdb.WriteStateHistoryIndexMetadata(db, blob)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown history type %d", typ))
|
||||||
|
}
|
||||||
|
log.Debug("Written index metadata", "type", typ, "last", last)
|
||||||
}
|
}
|
||||||
|
|
||||||
// batchIndexer is a structure designed to perform batch indexing or unindexing
|
// deleteIndexMetadata deletes the metadata of the specific history index.
|
||||||
// of state histories atomically.
|
func deleteIndexMetadata(db ethdb.KeyValueWriter, typ historyType) {
|
||||||
|
switch typ {
|
||||||
|
case typeStateHistory:
|
||||||
|
rawdb.DeleteStateHistoryIndexMetadata(db)
|
||||||
|
default:
|
||||||
|
panic(fmt.Errorf("unknown history type %d", typ))
|
||||||
|
}
|
||||||
|
log.Debug("Deleted index metadata", "type", typ)
|
||||||
|
}
|
||||||
|
|
||||||
|
// batchIndexer is responsible for performing batch indexing or unindexing
|
||||||
|
// of historical data (e.g., state or trie node changes) atomically.
|
||||||
type batchIndexer struct {
|
type batchIndexer struct {
|
||||||
accounts map[common.Hash][]uint64 // History ID list, Keyed by the hash of account address
|
index map[stateIdent][]uint64 // List of history IDs for tracked state entry
|
||||||
storages map[common.Hash]map[common.Hash][]uint64 // History ID list, Keyed by the hash of account address and the hash of raw storage key
|
pending int // Number of entries processed in the current batch.
|
||||||
counter int // The counter of processed states
|
delete bool // Operation mode: true for unindex, false for index.
|
||||||
delete bool // Index or unindex mode
|
lastID uint64 // ID of the most recently processed history.
|
||||||
lastID uint64 // The ID of latest processed history
|
typ historyType // Type of history being processed (e.g., state or trienode).
|
||||||
db ethdb.KeyValueStore
|
db ethdb.KeyValueStore // Key-value database used to store or delete index data.
|
||||||
}
|
}
|
||||||
|
|
||||||
// newBatchIndexer constructs the batch indexer with the supplied mode.
|
// newBatchIndexer constructs the batch indexer with the supplied mode.
|
||||||
func newBatchIndexer(db ethdb.KeyValueStore, delete bool) *batchIndexer {
|
func newBatchIndexer(db ethdb.KeyValueStore, delete bool, typ historyType) *batchIndexer {
|
||||||
return &batchIndexer{
|
return &batchIndexer{
|
||||||
accounts: make(map[common.Hash][]uint64),
|
index: make(map[stateIdent][]uint64),
|
||||||
storages: make(map[common.Hash]map[common.Hash][]uint64),
|
|
||||||
delete: delete,
|
delete: delete,
|
||||||
|
typ: typ,
|
||||||
db: db,
|
db: db,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// process iterates through the accounts and their associated storage slots in the
|
// process traverses the state entries within the provided history and tracks the mutation
|
||||||
// state history, tracking the mapping between state and history IDs.
|
// records for them.
|
||||||
func (b *batchIndexer) process(h *stateHistory, historyID uint64) error {
|
func (b *batchIndexer) process(h history, id uint64) error {
|
||||||
for _, address := range h.accountList {
|
for ident := range h.forEach() {
|
||||||
addrHash := crypto.Keccak256Hash(address.Bytes())
|
b.index[ident] = append(b.index[ident], id)
|
||||||
b.counter += 1
|
b.pending++
|
||||||
b.accounts[addrHash] = append(b.accounts[addrHash], historyID)
|
}
|
||||||
|
b.lastID = id
|
||||||
|
|
||||||
for _, slotKey := range h.storageList[address] {
|
|
||||||
b.counter += 1
|
|
||||||
if _, ok := b.storages[addrHash]; !ok {
|
|
||||||
b.storages[addrHash] = make(map[common.Hash][]uint64)
|
|
||||||
}
|
|
||||||
// The hash of the storage slot key is used as the identifier because the
|
|
||||||
// legacy history does not include the raw storage key, therefore, the
|
|
||||||
// conversion from storage key to hash is necessary for non-v0 histories.
|
|
||||||
slotHash := slotKey
|
|
||||||
if h.meta.version != stateHistoryV0 {
|
|
||||||
slotHash = crypto.Keccak256Hash(slotKey.Bytes())
|
|
||||||
}
|
|
||||||
b.storages[addrHash][slotHash] = append(b.storages[addrHash][slotHash], historyID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
b.lastID = historyID
|
|
||||||
return b.finish(false)
|
return b.finish(false)
|
||||||
}
|
}
|
||||||
|
|
||||||
// finish writes the accumulated state indexes into the disk if either the
|
// finish writes the accumulated state indexes into the disk if either the
|
||||||
// memory limitation is reached or it's requested forcibly.
|
// memory limitation is reached or it's requested forcibly.
|
||||||
func (b *batchIndexer) finish(force bool) error {
|
func (b *batchIndexer) finish(force bool) error {
|
||||||
if b.counter == 0 {
|
if b.pending == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
if !force && b.counter < historyIndexBatch {
|
if !force && b.pending < historyIndexBatch {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
var (
|
var (
|
||||||
batch = b.db.NewBatch()
|
batch = b.db.NewBatch()
|
||||||
batchMu sync.RWMutex
|
batchMu sync.RWMutex
|
||||||
storages int
|
|
||||||
start = time.Now()
|
start = time.Now()
|
||||||
eg errgroup.Group
|
eg errgroup.Group
|
||||||
)
|
)
|
||||||
eg.SetLimit(runtime.NumCPU())
|
eg.SetLimit(runtime.NumCPU())
|
||||||
|
|
||||||
for addrHash, idList := range b.accounts {
|
for ident, list := range b.index {
|
||||||
eg.Go(func() error {
|
eg.Go(func() error {
|
||||||
if !b.delete {
|
if !b.delete {
|
||||||
iw, err := newIndexWriter(b.db, newAccountIdent(addrHash))
|
iw, err := newIndexWriter(b.db, ident)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
for _, n := range idList {
|
for _, n := range list {
|
||||||
if err := iw.append(n); err != nil {
|
if err := iw.append(n); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -152,11 +173,11 @@ func (b *batchIndexer) finish(force bool) error {
|
||||||
iw.finish(batch)
|
iw.finish(batch)
|
||||||
batchMu.Unlock()
|
batchMu.Unlock()
|
||||||
} else {
|
} else {
|
||||||
id, err := newIndexDeleter(b.db, newAccountIdent(addrHash))
|
id, err := newIndexDeleter(b.db, ident)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
for _, n := range idList {
|
for _, n := range list {
|
||||||
if err := id.pop(n); err != nil {
|
if err := id.pop(n); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -168,72 +189,36 @@ func (b *batchIndexer) finish(force bool) error {
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
for addrHash, slots := range b.storages {
|
|
||||||
storages += len(slots)
|
|
||||||
for storageHash, idList := range slots {
|
|
||||||
eg.Go(func() error {
|
|
||||||
if !b.delete {
|
|
||||||
iw, err := newIndexWriter(b.db, newStorageIdent(addrHash, storageHash))
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
for _, n := range idList {
|
|
||||||
if err := iw.append(n); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
batchMu.Lock()
|
|
||||||
iw.finish(batch)
|
|
||||||
batchMu.Unlock()
|
|
||||||
} else {
|
|
||||||
id, err := newIndexDeleter(b.db, newStorageIdent(addrHash, storageHash))
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
for _, n := range idList {
|
|
||||||
if err := id.pop(n); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
batchMu.Lock()
|
|
||||||
id.finish(batch)
|
|
||||||
batchMu.Unlock()
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if err := eg.Wait(); err != nil {
|
if err := eg.Wait(); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// Update the position of last indexed state history
|
// Update the position of last indexed state history
|
||||||
if !b.delete {
|
if !b.delete {
|
||||||
storeIndexMetadata(batch, b.lastID)
|
storeIndexMetadata(batch, b.typ, b.lastID)
|
||||||
} else {
|
} else {
|
||||||
if b.lastID == 1 {
|
if b.lastID == 1 {
|
||||||
rawdb.DeleteStateHistoryIndexMetadata(batch)
|
deleteIndexMetadata(batch, b.typ)
|
||||||
} else {
|
} else {
|
||||||
storeIndexMetadata(batch, b.lastID-1)
|
storeIndexMetadata(batch, b.typ, b.lastID-1)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := batch.Write(); err != nil {
|
if err := batch.Write(); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
log.Debug("Committed batch indexer", "accounts", len(b.accounts), "storages", storages, "records", b.counter, "elapsed", common.PrettyDuration(time.Since(start)))
|
log.Debug("Committed batch indexer", "type", b.typ, "entries", len(b.index), "records", b.pending, "elapsed", common.PrettyDuration(time.Since(start)))
|
||||||
b.counter = 0
|
b.pending = 0
|
||||||
b.accounts = make(map[common.Hash][]uint64)
|
b.index = make(map[stateIdent][]uint64)
|
||||||
b.storages = make(map[common.Hash]map[common.Hash][]uint64)
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// indexSingle processes the state history with the specified ID for indexing.
|
// indexSingle processes the state history with the specified ID for indexing.
|
||||||
func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader) error {
|
func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader, typ historyType) error {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
defer func() {
|
defer func() {
|
||||||
indexHistoryTimer.UpdateSince(start)
|
indexHistoryTimer.UpdateSince(start)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
metadata := loadIndexMetadata(db)
|
metadata := loadIndexMetadata(db, typ)
|
||||||
if metadata == nil || metadata.Last+1 != historyID {
|
if metadata == nil || metadata.Last+1 != historyID {
|
||||||
last := "null"
|
last := "null"
|
||||||
if metadata != nil {
|
if metadata != nil {
|
||||||
|
|
@ -241,29 +226,37 @@ func indexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.Ancient
|
||||||
}
|
}
|
||||||
return fmt.Errorf("history indexing is out of order, last: %s, requested: %d", last, historyID)
|
return fmt.Errorf("history indexing is out of order, last: %s, requested: %d", last, historyID)
|
||||||
}
|
}
|
||||||
h, err := readStateHistory(freezer, historyID)
|
var (
|
||||||
|
err error
|
||||||
|
h history
|
||||||
|
b = newBatchIndexer(db, false, typ)
|
||||||
|
)
|
||||||
|
if typ == typeStateHistory {
|
||||||
|
h, err = readStateHistory(freezer, historyID)
|
||||||
|
} else {
|
||||||
|
// h, err = readTrienodeHistory(freezer, historyID)
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
b := newBatchIndexer(db, false)
|
|
||||||
if err := b.process(h, historyID); err != nil {
|
if err := b.process(h, historyID); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := b.finish(true); err != nil {
|
if err := b.finish(true); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
log.Debug("Indexed state history", "id", historyID, "elapsed", common.PrettyDuration(time.Since(start)))
|
log.Debug("Indexed history", "type", typ, "id", historyID, "elapsed", common.PrettyDuration(time.Since(start)))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// unindexSingle processes the state history with the specified ID for unindexing.
|
// unindexSingle processes the state history with the specified ID for unindexing.
|
||||||
func unindexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader) error {
|
func unindexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.AncientReader, typ historyType) error {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
defer func() {
|
defer func() {
|
||||||
unindexHistoryTimer.UpdateSince(start)
|
unindexHistoryTimer.UpdateSince(start)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
metadata := loadIndexMetadata(db)
|
metadata := loadIndexMetadata(db, typ)
|
||||||
if metadata == nil || metadata.Last != historyID {
|
if metadata == nil || metadata.Last != historyID {
|
||||||
last := "null"
|
last := "null"
|
||||||
if metadata != nil {
|
if metadata != nil {
|
||||||
|
|
@ -271,18 +264,26 @@ func unindexSingle(historyID uint64, db ethdb.KeyValueStore, freezer ethdb.Ancie
|
||||||
}
|
}
|
||||||
return fmt.Errorf("history unindexing is out of order, last: %s, requested: %d", last, historyID)
|
return fmt.Errorf("history unindexing is out of order, last: %s, requested: %d", last, historyID)
|
||||||
}
|
}
|
||||||
h, err := readStateHistory(freezer, historyID)
|
var (
|
||||||
|
err error
|
||||||
|
h history
|
||||||
|
)
|
||||||
|
b := newBatchIndexer(db, true, typ)
|
||||||
|
if typ == typeStateHistory {
|
||||||
|
h, err = readStateHistory(freezer, historyID)
|
||||||
|
} else {
|
||||||
|
// h, err = readTrienodeHistory(freezer, historyID)
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
b := newBatchIndexer(db, true)
|
|
||||||
if err := b.process(h, historyID); err != nil {
|
if err := b.process(h, historyID); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := b.finish(true); err != nil {
|
if err := b.finish(true); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
log.Debug("Unindexed state history", "id", historyID, "elapsed", common.PrettyDuration(time.Since(start)))
|
log.Debug("Unindexed history", "type", typ, "id", historyID, "elapsed", common.PrettyDuration(time.Since(start)))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -305,6 +306,8 @@ type indexIniter struct {
|
||||||
interrupt chan *interruptSignal
|
interrupt chan *interruptSignal
|
||||||
done chan struct{}
|
done chan struct{}
|
||||||
closed chan struct{}
|
closed chan struct{}
|
||||||
|
typ historyType
|
||||||
|
log log.Logger // Contextual logger with the history type injected
|
||||||
|
|
||||||
// indexing progress
|
// indexing progress
|
||||||
indexed atomic.Uint64 // the id of latest indexed state
|
indexed atomic.Uint64 // the id of latest indexed state
|
||||||
|
|
@ -313,18 +316,20 @@ type indexIniter struct {
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
}
|
}
|
||||||
|
|
||||||
func newIndexIniter(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, lastID uint64) *indexIniter {
|
func newIndexIniter(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, typ historyType, lastID uint64) *indexIniter {
|
||||||
initer := &indexIniter{
|
initer := &indexIniter{
|
||||||
disk: disk,
|
disk: disk,
|
||||||
freezer: freezer,
|
freezer: freezer,
|
||||||
interrupt: make(chan *interruptSignal),
|
interrupt: make(chan *interruptSignal),
|
||||||
done: make(chan struct{}),
|
done: make(chan struct{}),
|
||||||
closed: make(chan struct{}),
|
closed: make(chan struct{}),
|
||||||
|
typ: typ,
|
||||||
|
log: log.New("type", typ.String()),
|
||||||
}
|
}
|
||||||
// Load indexing progress
|
// Load indexing progress
|
||||||
var recover bool
|
var recover bool
|
||||||
initer.last.Store(lastID)
|
initer.last.Store(lastID)
|
||||||
metadata := loadIndexMetadata(disk)
|
metadata := loadIndexMetadata(disk, typ)
|
||||||
if metadata != nil {
|
if metadata != nil {
|
||||||
initer.indexed.Store(metadata.Last)
|
initer.indexed.Store(metadata.Last)
|
||||||
recover = metadata.Last > lastID
|
recover = metadata.Last > lastID
|
||||||
|
|
@ -371,7 +376,7 @@ func (i *indexIniter) remain() uint64 {
|
||||||
default:
|
default:
|
||||||
last, indexed := i.last.Load(), i.indexed.Load()
|
last, indexed := i.last.Load(), i.indexed.Load()
|
||||||
if last < indexed {
|
if last < indexed {
|
||||||
log.Warn("State indexer is in recovery", "indexed", indexed, "last", last)
|
i.log.Warn("State indexer is in recovery", "indexed", indexed, "last", last)
|
||||||
return indexed - last
|
return indexed - last
|
||||||
}
|
}
|
||||||
return last - indexed
|
return last - indexed
|
||||||
|
|
@ -389,7 +394,7 @@ func (i *indexIniter) run(lastID uint64) {
|
||||||
// checkDone indicates whether all requested state histories
|
// checkDone indicates whether all requested state histories
|
||||||
// have been fully indexed.
|
// have been fully indexed.
|
||||||
checkDone = func() bool {
|
checkDone = func() bool {
|
||||||
metadata := loadIndexMetadata(i.disk)
|
metadata := loadIndexMetadata(i.disk, i.typ)
|
||||||
return metadata != nil && metadata.Last == lastID
|
return metadata != nil && metadata.Last == lastID
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
|
@ -411,7 +416,7 @@ func (i *indexIniter) run(lastID uint64) {
|
||||||
if newLastID == lastID+1 {
|
if newLastID == lastID+1 {
|
||||||
lastID = newLastID
|
lastID = newLastID
|
||||||
signal.result <- nil
|
signal.result <- nil
|
||||||
log.Debug("Extended state history range", "last", lastID)
|
i.log.Debug("Extended history range", "last", lastID)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
// The index limit is shortened by one, interrupt the current background
|
// The index limit is shortened by one, interrupt the current background
|
||||||
|
|
@ -422,14 +427,14 @@ func (i *indexIniter) run(lastID uint64) {
|
||||||
// If all state histories, including the one to be reverted, have
|
// If all state histories, including the one to be reverted, have
|
||||||
// been fully indexed, unindex it here and shut down the initializer.
|
// been fully indexed, unindex it here and shut down the initializer.
|
||||||
if checkDone() {
|
if checkDone() {
|
||||||
log.Info("Truncate the extra history", "id", lastID)
|
i.log.Info("Truncate the extra history", "id", lastID)
|
||||||
if err := unindexSingle(lastID, i.disk, i.freezer); err != nil {
|
if err := unindexSingle(lastID, i.disk, i.freezer, i.typ); err != nil {
|
||||||
signal.result <- err
|
signal.result <- err
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
close(i.done)
|
close(i.done)
|
||||||
signal.result <- nil
|
signal.result <- nil
|
||||||
log.Info("State histories have been fully indexed", "last", lastID-1)
|
i.log.Info("Histories have been fully indexed", "last", lastID-1)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Adjust the indexing target and relaunch the process
|
// Adjust the indexing target and relaunch the process
|
||||||
|
|
@ -438,12 +443,12 @@ func (i *indexIniter) run(lastID uint64) {
|
||||||
|
|
||||||
done, interrupt = make(chan struct{}), new(atomic.Int32)
|
done, interrupt = make(chan struct{}), new(atomic.Int32)
|
||||||
go i.index(done, interrupt, lastID)
|
go i.index(done, interrupt, lastID)
|
||||||
log.Debug("Shortened state history range", "last", lastID)
|
i.log.Debug("Shortened history range", "last", lastID)
|
||||||
|
|
||||||
case <-done:
|
case <-done:
|
||||||
if checkDone() {
|
if checkDone() {
|
||||||
close(i.done)
|
close(i.done)
|
||||||
log.Info("State histories have been fully indexed", "last", lastID)
|
i.log.Info("Histories have been fully indexed", "last", lastID)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Relaunch the background runner if some tasks are left
|
// Relaunch the background runner if some tasks are left
|
||||||
|
|
@ -452,7 +457,7 @@ func (i *indexIniter) run(lastID uint64) {
|
||||||
|
|
||||||
case <-i.closed:
|
case <-i.closed:
|
||||||
interrupt.Store(1)
|
interrupt.Store(1)
|
||||||
log.Info("Waiting background history index initer to exit")
|
i.log.Info("Waiting background history index initer to exit")
|
||||||
<-done
|
<-done
|
||||||
|
|
||||||
if checkDone() {
|
if checkDone() {
|
||||||
|
|
@ -472,14 +477,14 @@ func (i *indexIniter) next() (uint64, error) {
|
||||||
tailID := tail + 1 // compute the id of the oldest history
|
tailID := tail + 1 // compute the id of the oldest history
|
||||||
|
|
||||||
// Start indexing from scratch if nothing has been indexed
|
// Start indexing from scratch if nothing has been indexed
|
||||||
metadata := loadIndexMetadata(i.disk)
|
metadata := loadIndexMetadata(i.disk, i.typ)
|
||||||
if metadata == nil {
|
if metadata == nil {
|
||||||
log.Debug("Initialize state history indexing from scratch", "id", tailID)
|
i.log.Debug("Initialize history indexing from scratch", "id", tailID)
|
||||||
return tailID, nil
|
return tailID, nil
|
||||||
}
|
}
|
||||||
// Resume indexing from the last interrupted position
|
// Resume indexing from the last interrupted position
|
||||||
if metadata.Last+1 >= tailID {
|
if metadata.Last+1 >= tailID {
|
||||||
log.Debug("Resume state history indexing", "id", metadata.Last+1, "tail", tailID)
|
i.log.Debug("Resume history indexing", "id", metadata.Last+1, "tail", tailID)
|
||||||
return metadata.Last + 1, nil
|
return metadata.Last + 1, nil
|
||||||
}
|
}
|
||||||
// History has been shortened without indexing. Discard the gapped segment
|
// History has been shortened without indexing. Discard the gapped segment
|
||||||
|
|
@ -487,7 +492,7 @@ func (i *indexIniter) next() (uint64, error) {
|
||||||
//
|
//
|
||||||
// The missing indexes corresponding to the gapped histories won't be visible.
|
// The missing indexes corresponding to the gapped histories won't be visible.
|
||||||
// It's fine to leave them unindexed.
|
// It's fine to leave them unindexed.
|
||||||
log.Info("History gap detected, discard old segment", "oldHead", metadata.Last, "newHead", tailID)
|
i.log.Info("History gap detected, discard old segment", "oldHead", metadata.Last, "newHead", tailID)
|
||||||
return tailID, nil
|
return tailID, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -496,7 +501,7 @@ func (i *indexIniter) index(done chan struct{}, interrupt *atomic.Int32, lastID
|
||||||
|
|
||||||
beginID, err := i.next()
|
beginID, err := i.next()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("Failed to find next state history for indexing", "err", err)
|
i.log.Error("Failed to find next history for indexing", "err", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// All available state histories have been indexed, and the last indexed one
|
// All available state histories have been indexed, and the last indexed one
|
||||||
|
|
@ -511,36 +516,47 @@ func (i *indexIniter) index(done chan struct{}, interrupt *atomic.Int32, lastID
|
||||||
//
|
//
|
||||||
// This step is essential to avoid spinning up indexing thread
|
// This step is essential to avoid spinning up indexing thread
|
||||||
// endlessly until a history object is produced.
|
// endlessly until a history object is produced.
|
||||||
storeIndexMetadata(i.disk, 0)
|
storeIndexMetadata(i.disk, i.typ, 0)
|
||||||
log.Info("Initialized history indexing flag")
|
i.log.Info("Initialized history indexing flag")
|
||||||
} else {
|
} else {
|
||||||
log.Debug("State history is fully indexed", "last", lastID)
|
i.log.Debug("History is fully indexed", "last", lastID)
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
log.Info("Start history indexing", "beginID", beginID, "lastID", lastID)
|
i.log.Info("Start history indexing", "beginID", beginID, "lastID", lastID)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
current = beginID
|
current = beginID
|
||||||
start = time.Now()
|
start = time.Now()
|
||||||
logged = time.Now()
|
logged = time.Now()
|
||||||
batch = newBatchIndexer(i.disk, false)
|
batch = newBatchIndexer(i.disk, false, i.typ)
|
||||||
)
|
)
|
||||||
for current <= lastID {
|
for current <= lastID {
|
||||||
count := lastID - current + 1
|
count := lastID - current + 1
|
||||||
if count > historyReadBatch {
|
if count > historyReadBatch {
|
||||||
count = historyReadBatch
|
count = historyReadBatch
|
||||||
}
|
}
|
||||||
histories, err := readStateHistories(i.freezer, current, count)
|
var histories []history
|
||||||
|
if i.typ == typeStateHistory {
|
||||||
|
histories, err = readStateHistories(i.freezer, current, count)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// The history read might fall if the history is truncated from
|
// The history read might fall if the history is truncated from
|
||||||
// head due to revert operation.
|
// head due to revert operation.
|
||||||
log.Error("Failed to read history for indexing", "current", current, "count", count, "err", err)
|
i.log.Error("Failed to read history for indexing", "current", current, "count", count, "err", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
} else {
|
||||||
|
// histories, err = readTrienodeHistories(i.freezer, current, count)
|
||||||
|
// if err != nil {
|
||||||
|
// // The history read might fall if the history is truncated from
|
||||||
|
// // head due to revert operation.
|
||||||
|
// i.log.Error("Failed to read history for indexing", "current", current, "count", count, "err", err)
|
||||||
|
// return
|
||||||
|
// }
|
||||||
|
}
|
||||||
for _, h := range histories {
|
for _, h := range histories {
|
||||||
if err := batch.process(h, current); err != nil {
|
if err := batch.process(h, current); err != nil {
|
||||||
log.Error("Failed to index history", "err", err)
|
i.log.Error("Failed to index history", "err", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
current += 1
|
current += 1
|
||||||
|
|
@ -554,7 +570,7 @@ func (i *indexIniter) index(done chan struct{}, interrupt *atomic.Int32, lastID
|
||||||
done = current - beginID
|
done = current - beginID
|
||||||
)
|
)
|
||||||
eta := common.CalculateETA(done, left, time.Since(start))
|
eta := common.CalculateETA(done, left, time.Since(start))
|
||||||
log.Info("Indexing state history", "processed", done, "left", left, "elapsed", common.PrettyDuration(time.Since(start)), "eta", common.PrettyDuration(eta))
|
i.log.Info("Indexing state history", "processed", done, "left", left, "elapsed", common.PrettyDuration(time.Since(start)), "eta", common.PrettyDuration(eta))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
i.indexed.Store(current - 1) // update indexing progress
|
i.indexed.Store(current - 1) // update indexing progress
|
||||||
|
|
@ -563,7 +579,7 @@ func (i *indexIniter) index(done chan struct{}, interrupt *atomic.Int32, lastID
|
||||||
if interrupt != nil {
|
if interrupt != nil {
|
||||||
if signal := interrupt.Load(); signal != 0 {
|
if signal := interrupt.Load(); signal != 0 {
|
||||||
if err := batch.finish(true); err != nil {
|
if err := batch.finish(true); err != nil {
|
||||||
log.Error("Failed to flush index", "err", err)
|
i.log.Error("Failed to flush index", "err", err)
|
||||||
}
|
}
|
||||||
log.Info("State indexing interrupted")
|
log.Info("State indexing interrupted")
|
||||||
return
|
return
|
||||||
|
|
@ -571,9 +587,9 @@ func (i *indexIniter) index(done chan struct{}, interrupt *atomic.Int32, lastID
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := batch.finish(true); err != nil {
|
if err := batch.finish(true); err != nil {
|
||||||
log.Error("Failed to flush index", "err", err)
|
i.log.Error("Failed to flush index", "err", err)
|
||||||
}
|
}
|
||||||
log.Info("Indexed state history", "from", beginID, "to", lastID, "elapsed", common.PrettyDuration(time.Since(start)))
|
i.log.Info("Indexed history", "from", beginID, "to", lastID, "elapsed", common.PrettyDuration(time.Since(start)))
|
||||||
}
|
}
|
||||||
|
|
||||||
// recover handles unclean shutdown recovery. After an unclean shutdown, any
|
// recover handles unclean shutdown recovery. After an unclean shutdown, any
|
||||||
|
|
@ -631,21 +647,31 @@ func (i *indexIniter) recover(lastID uint64) {
|
||||||
// state history.
|
// state history.
|
||||||
type historyIndexer struct {
|
type historyIndexer struct {
|
||||||
initer *indexIniter
|
initer *indexIniter
|
||||||
|
typ historyType
|
||||||
disk ethdb.KeyValueStore
|
disk ethdb.KeyValueStore
|
||||||
freezer ethdb.AncientStore
|
freezer ethdb.AncientStore
|
||||||
}
|
}
|
||||||
|
|
||||||
// checkVersion checks whether the index data in the database matches the version.
|
// checkVersion checks whether the index data in the database matches the version.
|
||||||
func checkVersion(disk ethdb.KeyValueStore) {
|
func checkVersion(disk ethdb.KeyValueStore, typ historyType) {
|
||||||
blob := rawdb.ReadStateHistoryIndexMetadata(disk)
|
var blob []byte
|
||||||
|
if typ == typeStateHistory {
|
||||||
|
blob = rawdb.ReadStateHistoryIndexMetadata(disk)
|
||||||
|
} else {
|
||||||
|
panic(fmt.Errorf("unknown history type: %v", typ))
|
||||||
|
}
|
||||||
|
// Short circuit if metadata is not found, re-index is required
|
||||||
|
// from scratch.
|
||||||
if len(blob) == 0 {
|
if len(blob) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
// Short circuit if the metadata is found and the version is matched
|
||||||
var m indexMetadata
|
var m indexMetadata
|
||||||
err := rlp.DecodeBytes(blob, &m)
|
err := rlp.DecodeBytes(blob, &m)
|
||||||
if err == nil && m.Version == stateIndexVersion {
|
if err == nil && m.Version == stateIndexVersion {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
// Version is not matched, prune the existing data and re-index from scratch
|
||||||
version := "unknown"
|
version := "unknown"
|
||||||
if err == nil {
|
if err == nil {
|
||||||
version = fmt.Sprintf("%d", m.Version)
|
version = fmt.Sprintf("%d", m.Version)
|
||||||
|
|
@ -662,10 +688,11 @@ func checkVersion(disk ethdb.KeyValueStore) {
|
||||||
|
|
||||||
// newHistoryIndexer constructs the history indexer and launches the background
|
// newHistoryIndexer constructs the history indexer and launches the background
|
||||||
// initer to complete the indexing of any remaining state histories.
|
// initer to complete the indexing of any remaining state histories.
|
||||||
func newHistoryIndexer(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, lastHistoryID uint64) *historyIndexer {
|
func newHistoryIndexer(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, lastHistoryID uint64, typ historyType) *historyIndexer {
|
||||||
checkVersion(disk)
|
checkVersion(disk, typ)
|
||||||
return &historyIndexer{
|
return &historyIndexer{
|
||||||
initer: newIndexIniter(disk, freezer, lastHistoryID),
|
initer: newIndexIniter(disk, freezer, typ, lastHistoryID),
|
||||||
|
typ: typ,
|
||||||
disk: disk,
|
disk: disk,
|
||||||
freezer: freezer,
|
freezer: freezer,
|
||||||
}
|
}
|
||||||
|
|
@ -693,7 +720,7 @@ func (i *historyIndexer) extend(historyID uint64) error {
|
||||||
case <-i.initer.closed:
|
case <-i.initer.closed:
|
||||||
return errors.New("indexer is closed")
|
return errors.New("indexer is closed")
|
||||||
case <-i.initer.done:
|
case <-i.initer.done:
|
||||||
return indexSingle(historyID, i.disk, i.freezer)
|
return indexSingle(historyID, i.disk, i.freezer, i.typ)
|
||||||
case i.initer.interrupt <- signal:
|
case i.initer.interrupt <- signal:
|
||||||
return <-signal.result
|
return <-signal.result
|
||||||
}
|
}
|
||||||
|
|
@ -710,7 +737,7 @@ func (i *historyIndexer) shorten(historyID uint64) error {
|
||||||
case <-i.initer.closed:
|
case <-i.initer.closed:
|
||||||
return errors.New("indexer is closed")
|
return errors.New("indexer is closed")
|
||||||
case <-i.initer.done:
|
case <-i.initer.done:
|
||||||
return unindexSingle(historyID, i.disk, i.freezer)
|
return unindexSingle(historyID, i.disk, i.freezer, i.typ)
|
||||||
case i.initer.interrupt <- signal:
|
case i.initer.interrupt <- signal:
|
||||||
return <-signal.result
|
return <-signal.result
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -38,7 +38,7 @@ func TestHistoryIndexerShortenDeadlock(t *testing.T) {
|
||||||
rawdb.WriteStateHistory(freezer, uint64(i+1), h.meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
rawdb.WriteStateHistory(freezer, uint64(i+1), h.meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
||||||
}
|
}
|
||||||
// As a workaround, assign a future block to keep the initer running indefinitely
|
// As a workaround, assign a future block to keep the initer running indefinitely
|
||||||
indexer := newHistoryIndexer(db, freezer, 200)
|
indexer := newHistoryIndexer(db, freezer, 200, typeStateHistory)
|
||||||
defer indexer.close()
|
defer indexer.close()
|
||||||
|
|
||||||
done := make(chan error, 1)
|
done := make(chan error, 1)
|
||||||
|
|
|
||||||
|
|
@ -29,88 +29,6 @@ import (
|
||||||
"github.com/ethereum/go-ethereum/ethdb"
|
"github.com/ethereum/go-ethereum/ethdb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// stateIdent represents the identifier of a state element, which can be
|
|
||||||
// either an account or a storage slot.
|
|
||||||
type stateIdent struct {
|
|
||||||
account bool
|
|
||||||
|
|
||||||
// The hash of the account address. This is used instead of the raw account
|
|
||||||
// address is to align the traversal order with the Merkle-Patricia-Trie.
|
|
||||||
addressHash common.Hash
|
|
||||||
|
|
||||||
// The hash of the storage slot key. This is used instead of the raw slot key
|
|
||||||
// because, in legacy state histories (prior to the Cancun fork), the slot
|
|
||||||
// identifier is the hash of the key, and the original key (preimage) cannot
|
|
||||||
// be recovered. To maintain backward compatibility, the key hash is used.
|
|
||||||
//
|
|
||||||
// Meanwhile, using the storage key hash also preserve the traversal order
|
|
||||||
// with Merkle-Patricia-Trie.
|
|
||||||
//
|
|
||||||
// This field is null if the identifier refers to account data.
|
|
||||||
storageHash common.Hash
|
|
||||||
}
|
|
||||||
|
|
||||||
// String returns the string format state identifier.
|
|
||||||
func (ident stateIdent) String() string {
|
|
||||||
if ident.account {
|
|
||||||
return ident.addressHash.Hex()
|
|
||||||
}
|
|
||||||
return ident.addressHash.Hex() + ident.storageHash.Hex()
|
|
||||||
}
|
|
||||||
|
|
||||||
// newAccountIdent constructs a state identifier for an account.
|
|
||||||
func newAccountIdent(addressHash common.Hash) stateIdent {
|
|
||||||
return stateIdent{
|
|
||||||
account: true,
|
|
||||||
addressHash: addressHash,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// newStorageIdent constructs a state identifier for a storage slot.
|
|
||||||
// The address denotes the address of the associated account;
|
|
||||||
// the storageHash denotes the hash of the raw storage slot key;
|
|
||||||
func newStorageIdent(addressHash common.Hash, storageHash common.Hash) stateIdent {
|
|
||||||
return stateIdent{
|
|
||||||
addressHash: addressHash,
|
|
||||||
storageHash: storageHash,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// stateIdentQuery is the extension of stateIdent by adding the raw storage key.
|
|
||||||
type stateIdentQuery struct {
|
|
||||||
stateIdent
|
|
||||||
|
|
||||||
address common.Address
|
|
||||||
storageKey common.Hash
|
|
||||||
}
|
|
||||||
|
|
||||||
// newAccountIdentQuery constructs a state identifier for an account.
|
|
||||||
func newAccountIdentQuery(address common.Address, addressHash common.Hash) stateIdentQuery {
|
|
||||||
return stateIdentQuery{
|
|
||||||
stateIdent: stateIdent{
|
|
||||||
account: true,
|
|
||||||
addressHash: addressHash,
|
|
||||||
},
|
|
||||||
address: address,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// newStorageIdentQuery constructs a state identifier for a storage slot.
|
|
||||||
// the address denotes the address of the associated account;
|
|
||||||
// the addressHash denotes the address hash of the associated account;
|
|
||||||
// the storageKey denotes the raw storage slot key;
|
|
||||||
// the storageHash denotes the hash of the raw storage slot key;
|
|
||||||
func newStorageIdentQuery(address common.Address, addressHash common.Hash, storageKey common.Hash, storageHash common.Hash) stateIdentQuery {
|
|
||||||
return stateIdentQuery{
|
|
||||||
stateIdent: stateIdent{
|
|
||||||
addressHash: addressHash,
|
|
||||||
storageHash: storageHash,
|
|
||||||
},
|
|
||||||
address: address,
|
|
||||||
storageKey: storageKey,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// indexReaderWithLimitTag is a wrapper around indexReader that includes an
|
// indexReaderWithLimitTag is a wrapper around indexReader that includes an
|
||||||
// additional index position. This position represents the ID of the last
|
// additional index position. This position represents the ID of the last
|
||||||
// indexed state history at the time the reader was created, implying that
|
// indexed state history at the time the reader was created, implying that
|
||||||
|
|
@ -169,7 +87,7 @@ func (r *indexReaderWithLimitTag) readGreaterThan(id uint64, lastID uint64) (uin
|
||||||
// Given that it's very unlikely to occur and users try to perform historical
|
// Given that it's very unlikely to occur and users try to perform historical
|
||||||
// state queries while reverting the states at the same time. Simply returning
|
// state queries while reverting the states at the same time. Simply returning
|
||||||
// an error should be sufficient for now.
|
// an error should be sufficient for now.
|
||||||
metadata := loadIndexMetadata(r.db)
|
metadata := loadIndexMetadata(r.db, toHistoryType(r.reader.state.typ))
|
||||||
if metadata == nil || metadata.Last < lastID {
|
if metadata == nil || metadata.Last < lastID {
|
||||||
return 0, errors.New("state history hasn't been indexed yet")
|
return 0, errors.New("state history hasn't been indexed yet")
|
||||||
}
|
}
|
||||||
|
|
@ -331,7 +249,7 @@ func (r *historyReader) read(state stateIdentQuery, stateID uint64, lastID uint6
|
||||||
// To serve the request, all state histories from stateID+1 to lastID
|
// To serve the request, all state histories from stateID+1 to lastID
|
||||||
// must be indexed. It's not supposed to happen unless system is very
|
// must be indexed. It's not supposed to happen unless system is very
|
||||||
// wrong.
|
// wrong.
|
||||||
metadata := loadIndexMetadata(r.disk)
|
metadata := loadIndexMetadata(r.disk, toHistoryType(state.typ))
|
||||||
if metadata == nil || metadata.Last < lastID {
|
if metadata == nil || metadata.Last < lastID {
|
||||||
indexed := "null"
|
indexed := "null"
|
||||||
if metadata != nil {
|
if metadata != nil {
|
||||||
|
|
@ -364,7 +282,7 @@ func (r *historyReader) read(state stateIdentQuery, stateID uint64, lastID uint6
|
||||||
// that the associated state histories are no longer available due to a rollback.
|
// that the associated state histories are no longer available due to a rollback.
|
||||||
// Such truncation should be captured by the state resolver below, rather than returning
|
// Such truncation should be captured by the state resolver below, rather than returning
|
||||||
// invalid data.
|
// invalid data.
|
||||||
if state.account {
|
if state.typ == typeAccount {
|
||||||
return r.readAccount(state.address, historyID)
|
return r.readAccount(state.address, historyID)
|
||||||
}
|
}
|
||||||
return r.readStorage(state.address, state.storageKey, state.storageHash, historyID)
|
return r.readStorage(state.address, state.storageKey, state.storageHash, historyID)
|
||||||
|
|
|
||||||
|
|
@ -29,7 +29,7 @@ import (
|
||||||
|
|
||||||
func waitIndexing(db *Database) {
|
func waitIndexing(db *Database) {
|
||||||
for {
|
for {
|
||||||
metadata := loadIndexMetadata(db.diskdb)
|
metadata := loadIndexMetadata(db.diskdb, typeStateHistory)
|
||||||
if metadata != nil && metadata.Last >= db.tree.bottom().stateID() {
|
if metadata != nil && metadata.Last >= db.tree.bottom().stateID() {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@ import (
|
||||||
"encoding/binary"
|
"encoding/binary"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"iter"
|
||||||
"maps"
|
"maps"
|
||||||
"slices"
|
"slices"
|
||||||
"time"
|
"time"
|
||||||
|
|
@ -275,6 +276,36 @@ func newStateHistory(root common.Hash, parent common.Hash, block uint64, account
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// typ implements the history interface, returning the historical data type held.
|
||||||
|
func (h *stateHistory) typ() historyType {
|
||||||
|
return typeStateHistory
|
||||||
|
}
|
||||||
|
|
||||||
|
// forEach implements the history interface, returning an iterator to traverse the
|
||||||
|
// state entries in the history.
|
||||||
|
func (h *stateHistory) forEach() iter.Seq[stateIdent] {
|
||||||
|
return func(yield func(stateIdent) bool) {
|
||||||
|
for _, addr := range h.accountList {
|
||||||
|
addrHash := crypto.Keccak256Hash(addr.Bytes())
|
||||||
|
if !yield(newAccountIdent(addrHash)) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
for _, slotKey := range h.storageList[addr] {
|
||||||
|
// The hash of the storage slot key is used as the identifier because the
|
||||||
|
// legacy history does not include the raw storage key, therefore, the
|
||||||
|
// conversion from storage key to hash is necessary for non-v0 histories.
|
||||||
|
slotHash := slotKey
|
||||||
|
if h.meta.version != stateHistoryV0 {
|
||||||
|
slotHash = crypto.Keccak256Hash(slotKey.Bytes())
|
||||||
|
}
|
||||||
|
if !yield(newStorageIdent(addrHash, slotHash)) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// stateSet returns the state set, keyed by the hash of the account address
|
// stateSet returns the state set, keyed by the hash of the account address
|
||||||
// and the hash of the storage slot key.
|
// and the hash of the storage slot key.
|
||||||
func (h *stateHistory) stateSet() (map[common.Hash][]byte, map[common.Hash]map[common.Hash][]byte) {
|
func (h *stateHistory) stateSet() (map[common.Hash][]byte, map[common.Hash]map[common.Hash][]byte) {
|
||||||
|
|
@ -536,8 +567,8 @@ func readStateHistory(reader ethdb.AncientReader, id uint64) (*stateHistory, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// readStateHistories reads a list of state history records within the specified range.
|
// readStateHistories reads a list of state history records within the specified range.
|
||||||
func readStateHistories(freezer ethdb.AncientReader, start uint64, count uint64) ([]*stateHistory, error) {
|
func readStateHistories(freezer ethdb.AncientReader, start uint64, count uint64) ([]history, error) {
|
||||||
var histories []*stateHistory
|
var histories []history
|
||||||
metaList, aIndexList, sIndexList, aDataList, sDataList, err := rawdb.ReadStateHistoryList(freezer, start, count)
|
metaList, aIndexList, sIndexList, aDataList, sDataList, err := rawdb.ReadStateHistoryList(freezer, start, count)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|
|
||||||
|
|
@ -137,7 +137,7 @@ func TestTruncateHeadStateHistory(t *testing.T) {
|
||||||
rawdb.WriteStateHistory(freezer, uint64(i+1), hs[i].meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
rawdb.WriteStateHistory(freezer, uint64(i+1), hs[i].meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
||||||
}
|
}
|
||||||
for size := len(hs); size > 0; size-- {
|
for size := len(hs); size > 0; size-- {
|
||||||
pruned, err := truncateFromHead(freezer, uint64(size-1))
|
pruned, err := truncateFromHead(freezer, typeStateHistory, uint64(size-1))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Failed to truncate from head %v", err)
|
t.Fatalf("Failed to truncate from head %v", err)
|
||||||
}
|
}
|
||||||
|
|
@ -161,7 +161,7 @@ func TestTruncateTailStateHistory(t *testing.T) {
|
||||||
rawdb.WriteStateHistory(freezer, uint64(i+1), hs[i].meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
rawdb.WriteStateHistory(freezer, uint64(i+1), hs[i].meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
||||||
}
|
}
|
||||||
for newTail := 1; newTail < len(hs); newTail++ {
|
for newTail := 1; newTail < len(hs); newTail++ {
|
||||||
pruned, _ := truncateFromTail(freezer, uint64(newTail))
|
pruned, _ := truncateFromTail(freezer, typeStateHistory, uint64(newTail))
|
||||||
if pruned != 1 {
|
if pruned != 1 {
|
||||||
t.Error("Unexpected pruned items", "want", 1, "got", pruned)
|
t.Error("Unexpected pruned items", "want", 1, "got", pruned)
|
||||||
}
|
}
|
||||||
|
|
@ -209,7 +209,7 @@ func TestTruncateTailStateHistories(t *testing.T) {
|
||||||
accountData, storageData, accountIndex, storageIndex := hs[i].encode()
|
accountData, storageData, accountIndex, storageIndex := hs[i].encode()
|
||||||
rawdb.WriteStateHistory(freezer, uint64(i+1), hs[i].meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
rawdb.WriteStateHistory(freezer, uint64(i+1), hs[i].meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
||||||
}
|
}
|
||||||
pruned, _ := truncateFromTail(freezer, uint64(10)-c.limit)
|
pruned, _ := truncateFromTail(freezer, typeStateHistory, uint64(10)-c.limit)
|
||||||
if pruned != c.expPruned {
|
if pruned != c.expPruned {
|
||||||
t.Error("Unexpected pruned items", "want", c.expPruned, "got", pruned)
|
t.Error("Unexpected pruned items", "want", c.expPruned, "got", pruned)
|
||||||
}
|
}
|
||||||
|
|
@ -233,7 +233,7 @@ func TestTruncateOutOfRange(t *testing.T) {
|
||||||
accountData, storageData, accountIndex, storageIndex := hs[i].encode()
|
accountData, storageData, accountIndex, storageIndex := hs[i].encode()
|
||||||
rawdb.WriteStateHistory(freezer, uint64(i+1), hs[i].meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
rawdb.WriteStateHistory(freezer, uint64(i+1), hs[i].meta.encode(), accountIndex, storageIndex, accountData, storageData)
|
||||||
}
|
}
|
||||||
truncateFromTail(freezer, uint64(len(hs)/2))
|
truncateFromTail(freezer, typeStateHistory, uint64(len(hs)/2))
|
||||||
|
|
||||||
// Ensure of-out-range truncations are rejected correctly.
|
// Ensure of-out-range truncations are rejected correctly.
|
||||||
head, _ := freezer.Ancients()
|
head, _ := freezer.Ancients()
|
||||||
|
|
@ -254,9 +254,9 @@ func TestTruncateOutOfRange(t *testing.T) {
|
||||||
for _, c := range cases {
|
for _, c := range cases {
|
||||||
var gotErr error
|
var gotErr error
|
||||||
if c.mode == 0 {
|
if c.mode == 0 {
|
||||||
_, gotErr = truncateFromHead(freezer, c.target)
|
_, gotErr = truncateFromHead(freezer, typeStateHistory, c.target)
|
||||||
} else {
|
} else {
|
||||||
_, gotErr = truncateFromTail(freezer, c.target)
|
_, gotErr = truncateFromTail(freezer, typeStateHistory, c.target)
|
||||||
}
|
}
|
||||||
if !errors.Is(gotErr, c.expErr) {
|
if !errors.Is(gotErr, c.expErr) {
|
||||||
t.Errorf("Unexpected error, want: %v, got: %v", c.expErr, gotErr)
|
t.Errorf("Unexpected error, want: %v, got: %v", c.expErr, gotErr)
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue