core/rawdb, triedb/pathdb: implement history indexer

core/rawdb/accessors_history.go

@@ -0,0 +1,98 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package rawdb
+
+import (
+	"encoding/binary"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+// ReadStateHistoryIndexHead retrieves the number of latest indexed state history.
+func ReadStateHistoryIndexHead(db ethdb.KeyValueReader) *uint64 {
+	data, _ := db.Get(stateHistoryIndexHeadKey)
+	if len(data) != 8 {
+		return nil
+	}
+	number := binary.BigEndian.Uint64(data)
+	return &number
+}
+
+// WriteStateHistoryIndexHead stores the number of latest indexed state history
+// into database.
+func WriteStateHistoryIndexHead(db ethdb.KeyValueWriter, number uint64) {
+	if err := db.Put(stateHistoryIndexHeadKey, encodeBlockNumber(number)); err != nil {
+		log.Crit("Failed to store the state index tail", "err", err)
+	}
+}
+
+// DeleteStateHistoryIndexHead removes the number of latest indexed state history.
+func DeleteStateHistoryIndexHead(db ethdb.KeyValueWriter) {
+	if err := db.Delete(stateHistoryIndexHeadKey); err != nil {
+		log.Crit("Failed to delete the state index tail", "err", err)
+	}
+}
+
+// ReadStateIndex retrieves the state index with the provided account address
+// and state hash.
+func ReadStateIndex(db ethdb.KeyValueReader, address common.Address, state common.Hash) []byte {
+	data, err := db.Get(stateIndexKey(address, state))
+	if err != nil || len(data) == 0 {
+		return nil
+	}
+	return data
+}
+
+// WriteStateIndex writes the provided state index into database.
+func WriteStateIndex(db ethdb.KeyValueWriter, address common.Address, state common.Hash, data []byte) {
+	if err := db.Put(stateIndexKey(address, state), data); err != nil {
+		log.Crit("Failed to store state index", "err", err)
+	}
+}
+
+// DeleteStateIndex deletes the specified state index from the database.
+func DeleteStateIndex(db ethdb.KeyValueWriter, address common.Address, state common.Hash) {
+	if err := db.Delete(stateIndexKey(address, state)); err != nil {
+		log.Crit("Failed to delete state index", "err", err)
+	}
+}
+
+// ReadStateIndexBlock retrieves the state index block with the provided state
+// identifier along with the block id.
+func ReadStateIndexBlock(db ethdb.KeyValueReader, address common.Address, state common.Hash, id uint32) []byte {
+	data, err := db.Get(stateIndexBlockKey(address, state, id))
+	if err != nil || len(data) == 0 {
+		return nil
+	}
+	return data
+}
+
+// WriteStateIndexBlock writes the provided state index block into database.
+func WriteStateIndexBlock(db ethdb.KeyValueWriter, address common.Address, state common.Hash, id uint32, data []byte) {
+	if err := db.Put(stateIndexBlockKey(address, state, id), data); err != nil {
+		log.Crit("Failed to store state index", "err", err)
+	}
+}
+
+// DeleteStateIndexBlock deletes the specified state index block from the database.
+func DeleteStateIndexBlock(db ethdb.KeyValueWriter, address common.Address, state common.Hash, id uint32) {
+	if err := db.Delete(stateIndexBlockKey(address, state, id)); err != nil {
+		log.Crit("Failed to delete state index", "err", err)
+	}
+}

core/rawdb/accessors_state.go

@@ -251,6 +251,33 @@ func ReadStateHistory(db ethdb.AncientReaderOp, id uint64) ([]byte, []byte, []by
 	return meta, accountIndex, storageIndex, accountData, storageData, nil
 }
 
+// ReadStateHistoryList retrieves a list of state histories from database with
+// specific range. Compute the position of state history in freezer by minus one
+// since the id of first state history starts from one(zero for initial state).
+func ReadStateHistoryList(db ethdb.AncientReaderOp, start uint64, count uint64) ([][]byte, [][]byte, [][]byte, [][]byte, [][]byte, error) {
+	metaList, err := db.AncientRange(stateHistoryMeta, start-1, count, 0)
+	if err != nil {
+		return nil, nil, nil, nil, nil, err
+	}
+	aIndexList, err := db.AncientRange(stateHistoryAccountIndex, start-1, count, 0)
+	if err != nil {
+		return nil, nil, nil, nil, nil, err
+	}
+	sIndexList, err := db.AncientRange(stateHistoryStorageIndex, start-1, count, 0)
+	if err != nil {
+		return nil, nil, nil, nil, nil, err
+	}
+	aDataList, err := db.AncientRange(stateHistoryAccountData, start-1, count, 0)
+	if err != nil {
+		return nil, nil, nil, nil, nil, err
+	}
+	sDataList, err := db.AncientRange(stateHistoryStorageData, start-1, count, 0)
+	if err != nil {
+		return nil, nil, nil, nil, nil, err
+	}
+	return metaList, aIndexList, sIndexList, aDataList, sDataList, nil
+}
+
 // WriteStateHistory writes the provided state history to database. Compute the
 // position of state history in freezer by minus one since the id of first state
 // history starts from one(zero for initial state).

core/rawdb/schema.go

@@ -76,6 +76,9 @@ var (
 	// trieJournalKey tracks the in-memory trie node layers across restarts.
 	trieJournalKey = []byte("TrieJournal")
 
+	// stateHistoryIndexHeadKey tracks the ID of the latest state that has been indexed.
+	stateHistoryIndexHeadKey = []byte("StateHistoryIndexHead")
+
 	// txIndexTailKey tracks the oldest block whose transactions have been indexed.
 	txIndexTailKey = []byte("TransactionIndexTail")
 
@@ -117,6 +120,9 @@ var (
 	TrieNodeStoragePrefix = []byte("O") // TrieNodeStoragePrefix + accountHash + hexPath -> trie node
 	stateIDPrefix         = []byte("L") // stateIDPrefix + state root -> state id
 
+	// state history indexing within path-based storage scheme
+	stateIndexPrefix = []byte("M") // stateIndexPrefix + account address or (account address + slotHash) -> index
+
 	// VerklePrefix is the database prefix for Verkle trie data, which includes:
 	// (a) Trie nodes
 	// (b) In-memory trie node journal
@@ -346,3 +352,21 @@ func IsStorageTrieNode(key []byte) bool {
 	ok, _, _ := ResolveStorageTrieNode(key)
 	return ok
 }
+
+// stateIndexKey = stateIndexPrefix + address + state
+func stateIndexKey(address common.Address, state common.Hash) []byte {
+	if state == (common.Hash{}) {
+		return append(stateIndexPrefix, address.Bytes()...)
+	}
+	return append(append(stateIndexPrefix, address.Bytes()...), state.Bytes()...)
+}
+
+// stateIndexBlockKey = stateIndexPrefix + address + state + id
+func stateIndexBlockKey(address common.Address, state common.Hash, id uint32) []byte {
+	var buf [4]byte
+	binary.BigEndian.PutUint32(buf[:], id)
+	if state == (common.Hash{}) {
+		return append(append(stateIndexPrefix, address.Bytes()...), buf[:]...)
+	}
+	return append(append(append(stateIndexPrefix, address.Bytes()...), state.Bytes()...), buf[:]...)
+}

triedb/pathdb/database.go

@@ -161,6 +161,7 @@ type Database struct {
 	diskdb     ethdb.Database               // Persistent storage for matured trie nodes
 	tree       *layerTree                   // The group for all known layers
 	freezer    ethdb.ResettableAncientStore // Freezer for storing trie histories, nil possible in tests
+	indexer    *historyIndexer              // History indexer
 	lock       sync.RWMutex                 // Lock to prevent mutations from happening at the same time
 }
 
@@ -257,6 +258,8 @@ func (db *Database) repairHistory() error {
 	if pruned != 0 {
 		log.Warn("Truncated extra state histories", "number", pruned)
 	}
+	// TODO read-only mode?
+	db.indexer = newHistoryIndexer(db.diskdb, db.freezer, db.tree.bottom().stateID())
 	return nil
 }
 

triedb/pathdb/disklayer.go

@@ -240,6 +240,9 @@ func (dl *diskLayer) commit(bottom *diffLayer, force bool) (*diskLayer, error) {
 			overflow = true
 			oldest = bottom.stateID() - limit + 1 // track the id of history **after truncation**
 		}
+		if dl.db.indexer != nil {
+			dl.db.indexer.notify(bottom.stateID())
+		}
 	}
 	// Mark the diskLayer as stale before applying any mutations on top.
 	dl.markStale()

triedb/pathdb/history.go

@@ -490,6 +490,32 @@ func readHistory(reader ethdb.AncientReader, id uint64) (*history, error) {
 	return &dec, nil
 }
 
+// readHistories reads and decodes a list of state histories with specific
+// history range.
+func readHistories(freezer ethdb.AncientReader, start uint64, count uint64) ([]*history, error) {
+	metaList, aIndexList, sIndexList, aDataList, sDataList, err := rawdb.ReadStateHistoryList(freezer, start, count)
+	if err != nil {
+		return nil, err
+	}
+	number := len(metaList)
+	if number != len(aIndexList) || number != len(sIndexList) || number != len(aDataList) || number != len(sDataList) {
+		return nil, errors.New("corrupted state history")
+	}
+	var result []*history
+	for i := 0; i < number; i++ {
+		var m meta
+		if err := m.decode(metaList[i]); err != nil {
+			return nil, err
+		}
+		dec := history{meta: &m}
+		if err := dec.decode(aDataList[i], sDataList[i], aIndexList[i], sIndexList[i]); err != nil {
+			return nil, err
+		}
+		result = append(result, &dec)
+	}
+	return result, nil
+}
+
 // writeHistory persists the state history with the provided state set.
 func writeHistory(writer ethdb.AncientWriter, dl *diffLayer) error {
 	// Short circuit if state set is not available.

triedb/pathdb/history_index_reader.go

@@ -0,0 +1,339 @@
+// Copyright 2023 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/
+
+package pathdb
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"math"
+	"sort"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/ethdb"
+)
+
+type blockReader struct {
+	restarts []uint32
+	buf      []byte
+}
+
+func parseIndexBlock(blob []byte) ([]uint32, []byte, error) {
+	if len(blob) < 4 {
+		return nil, nil, fmt.Errorf("corrupted index block, len: %d", len(blob))
+	}
+	restartLen := binary.BigEndian.Uint32(blob[len(blob)-4:])
+	if restartLen == 0 {
+		return nil, nil, errors.New("corrupted index block, no restart")
+	}
+	tailLen := int(restartLen+1) * 4
+	if len(blob) < tailLen {
+		return nil, nil, fmt.Errorf("truncated restarts, size: %d, restarts: %d", len(blob), restartLen)
+	}
+	restarts := make([]uint32, 0, restartLen)
+	for i := restartLen; i > 0; i-- {
+		restart := binary.BigEndian.Uint32(blob[len(blob)-int(i+1)*4:])
+		restarts = append(restarts, restart)
+	}
+	prev := restarts[0]
+	for i := 1; i < len(restarts); i++ {
+		if restarts[i] <= prev {
+			return nil, nil, fmt.Errorf("restart out of order, prev: %d, next: %d", prev, restarts[i])
+		}
+		if int(restarts[i]) >= len(blob)-tailLen {
+			return nil, nil, fmt.Errorf("invalid restart position, restart: %d, size: %d", restarts[i], len(blob)-tailLen)
+		}
+		prev = restarts[i]
+	}
+	return restarts, blob[:len(blob)-tailLen], nil
+}
+
+func newBlockReader(disk ethdb.KeyValueReader, addr common.Address, state common.Hash, id uint32) (*blockReader, error) {
+	blob := rawdb.ReadStateIndexBlock(disk, addr, state, id)
+	if len(blob) == 0 {
+		return nil, errors.New("index block is not present")
+	}
+	restarts, data, err := parseIndexBlock(blob)
+	if err != nil {
+		return nil, err
+	}
+	return &blockReader{
+		restarts: restarts,
+		buf:      data, // safe to own the slice
+	}, nil
+}
+
+func (br *blockReader) readGreaterThan(id uint64) (uint64, error) {
+	var err error
+	index := sort.Search(len(br.restarts), func(i int) bool {
+		item, n := binary.Uvarint(br.buf[br.restarts[i]:])
+		if n <= 0 {
+			err = errors.New("failed to decode item at restart point")
+		}
+		return item > id
+	})
+	if err != nil {
+		return 0, err
+	}
+	if index == 0 {
+		item, _ := binary.Uvarint(br.buf[br.restarts[0]:])
+		return item, nil
+	}
+	var (
+		start  int
+		limit  int
+		result uint64
+	)
+	if index == len(br.restarts) {
+		start = int(br.restarts[len(br.restarts)-1])
+		limit = len(br.buf)
+	} else {
+		start = int(br.restarts[index-1])
+		limit = int(br.restarts[index])
+	}
+	pos := start
+	for pos < limit {
+		x, n := binary.Uvarint(br.buf[pos:])
+		if pos == start {
+			result = x
+		} else {
+			result += x
+		}
+		if result > id {
+			return result, nil
+		}
+		pos += n
+	}
+	return 0, errors.New("not found")
+}
+
+type indexReader struct {
+	disk     ethdb.KeyValueReader
+	descList []*indexBlockDesc
+	readers  map[uint32]*blockReader
+	owner    common.Address
+	state    common.Hash
+}
+
+func parseIndex(blob []byte) ([]*indexBlockDesc, error) {
+	if len(blob) == 0 {
+		return nil, errors.New("state index not found")
+	}
+	if len(blob)%indexBlockDescSize != 0 {
+		return nil, fmt.Errorf("corrupted state index, len: %d", len(blob))
+	}
+	var descList []*indexBlockDesc
+	for i := 0; i < len(blob)/indexBlockDescSize; i++ {
+		var desc indexBlockDesc
+		desc.decode(blob[i*indexBlockDescSize : (i+1)*indexBlockDescSize])
+		if desc.empty() {
+			return nil, errors.New("empty state index block")
+		}
+		descList = append(descList, &desc)
+	}
+	return descList, nil
+}
+
+func newIndexReader(disk ethdb.KeyValueReader, owner common.Address, state common.Hash) (*indexReader, error) {
+	descList, err := parseIndex(rawdb.ReadStateIndex(disk, owner, state))
+	if err != nil {
+		return nil, err
+	}
+	return &indexReader{
+		descList: descList,
+		readers:  make(map[uint32]*blockReader),
+		disk:     disk,
+		owner:    owner,
+		state:    state,
+	}, nil
+}
+
+func (r *indexReader) readGreaterThan(id uint64) (uint64, error) {
+	index := sort.Search(len(r.descList), func(i int) bool {
+		return id < r.descList[i].max
+	})
+	if index == len(r.descList) {
+		return math.MaxUint64, nil
+	}
+	desc := r.descList[index]
+
+	br, ok := r.readers[desc.id]
+	if !ok {
+		var err error
+		br, err = newBlockReader(r.disk, r.owner, r.state, desc.id)
+		if err != nil {
+			return 0, err
+		}
+		r.readers[desc.id] = br
+	}
+	return br.readGreaterThan(id)
+}
+
+type historyReader struct {
+	disk    ethdb.KeyValueReader
+	freezer ethdb.AncientReader
+	readers map[string]*indexReader
+}
+
+func newHistoryReader(disk ethdb.KeyValueReader, freezer ethdb.AncientReader) *historyReader {
+	return &historyReader{
+		disk:    disk,
+		freezer: freezer,
+		readers: make(map[string]*indexReader),
+	}
+}
+
+func (r *historyReader) findAccount(account common.Address, id uint64, resolve func([]byte)) error {
+	blob := rawdb.ReadStateAccountIndex(r.freezer, id)
+	if len(blob)%accountIndexSize != 0 {
+		return errors.New("corrupted account index")
+	}
+	n := len(blob) / accountIndexSize
+
+	index := sort.Search(n, func(i int) bool {
+		h := blob[accountIndexSize*i : accountIndexSize*i+common.HashLength]
+		return bytes.Compare(h, account.Bytes()) >= 0
+	})
+	if index == n {
+		return errors.New("account is not found")
+	}
+	if account != common.BytesToAddress(blob[accountIndexSize*index:accountIndexSize*index+common.AddressLength]) {
+		return errors.New("account is not found")
+	}
+	resolve(blob[accountIndexSize*index : accountIndexSize*(index+1)])
+	return nil
+}
+
+func (r *historyReader) findStorage(storageHash common.Hash, id uint64, slotOffset, slotLength int, resolve func([]byte)) error {
+	blob := rawdb.ReadStateStorageIndex(r.freezer, id)
+	if len(blob)%slotIndexSize != 0 {
+		return errors.New("storage indices are not corrupted")
+	}
+	if slotIndexSize*(slotOffset+slotLength) > len(blob) {
+		return errors.New("out of slice")
+	}
+	subSlice := blob[slotIndexSize*slotOffset : slotIndexSize*(slotOffset+slotLength)]
+
+	index := sort.Search(slotLength, func(i int) bool {
+		slotHash := subSlice[slotIndexSize*i : slotIndexSize*i+common.HashLength]
+		return bytes.Compare(slotHash, storageHash.Bytes()) >= 0
+	})
+	if index == slotLength {
+		return errors.New("storage is not found")
+	}
+	if storageHash != common.BytesToHash(subSlice[slotIndexSize*index:slotIndexSize*index+common.HashLength]) {
+		return errors.New("storage is not found")
+	}
+	resolve(subSlice[slotIndexSize*index : slotIndexSize*(index+1)])
+	return nil
+}
+
+func (r *historyReader) resolveAccount(accountHash common.Address, id uint64) ([]byte, error) {
+	var (
+		offset int
+		length int
+	)
+	err := r.findAccount(accountHash, id, func(blob []byte) {
+		length = int(blob[common.AddressLength])
+		offset = int(binary.BigEndian.Uint32(blob[common.AddressLength+1 : common.AddressLength+5]))
+	})
+	if err != nil {
+		return nil, err
+	}
+	// TODO(rj493456442) optimize it with partial read
+	data := rawdb.ReadStateAccountHistory(r.freezer, id)
+	if len(data) < offset+length {
+		return nil, errors.New("corrupted account data")
+	}
+	return data[offset : offset+length], nil
+}
+
+func (r *historyReader) resolveStorage(account common.Address, storageHash common.Hash, id uint64) ([]byte, error) {
+	var (
+		slotOffset int
+		slotLength int
+		offset     int
+		length     int
+	)
+	err := r.findAccount(account, id, func(blob []byte) {
+		slotOffset = int(binary.BigEndian.Uint32(blob[common.AddressLength+5 : common.AddressLength+9]))
+		slotLength = int(binary.BigEndian.Uint32(blob[common.AddressLength+9 : common.AddressLength+13]))
+	})
+	if err != nil {
+		return nil, err
+	}
+	err = r.findStorage(storageHash, id, slotOffset, slotLength, func(blob []byte) {
+		length = int(blob[common.HashLength])
+		offset = int(binary.BigEndian.Uint32(blob[common.HashLength+1 : common.HashLength+5]))
+	})
+	if err != nil {
+		return nil, err
+	}
+	// TODO(rj493456442) optimize it with partial read
+	data := rawdb.ReadStateStorageHistory(r.freezer, id)
+	if len(data) < offset+length {
+		return nil, errors.New("corrupted storage data")
+	}
+	return data[offset : offset+length], nil
+}
+
+func (r *historyReader) resolve(owner common.Address, state common.Hash, id uint64) ([]byte, error) {
+	if state == (common.Hash{}) {
+		return r.resolveAccount(owner, id)
+	}
+	return r.resolveStorage(owner, state, id)
+}
+
+func (r *historyReader) read(owner common.Address, state common.Hash, id uint64, latest uint64) ([]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 {
+		return nil, errors.New("historic state is pruned")
+	}
+	head := rawdb.ReadStateHistoryIndexHead(r.disk)
+
+	/*
+		the available range of histories is [tail+1, head]
+	*/
+	if head == nil || *head <= id {
+		return nil, errors.New("state history is not fully indexed")
+	}
+	ir, ok := r.readers[owner.Hex()+state.Hex()]
+	if !ok {
+		ir, err = newIndexReader(r.disk, owner, state)
+		if err != nil {
+			return nil, err
+		}
+		r.readers[owner.Hex()+state.Hex()] = ir
+	}
+	id, err = ir.readGreaterThan(id)
+	if err != nil {
+		return nil, err
+	}
+	if id == math.MaxUint64 {
+		if *head < latest {
+			return nil, errors.New("state history is not fully indexed")
+		}
+		return nil, errors.New("not found")
+	}
+	return r.resolve(owner, state, id)
+}

triedb/pathdb/history_index_writer.go

@@ -0,0 +1,330 @@
+// 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 <http://www.gnu.org/licenses/
+
+package pathdb
+
+import (
+	"encoding/binary"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+const (
+	indexBlockDescSize   = 24   // The size of index block descriptor
+	indexBlockEntriesCap = 4096 // The maximum number of entries can be grouped in a block
+	indexBlockRestartLen = 256  // The restart interval length of index block
+
+	// stateWriteBatch is the number of states for constructing indexes together.
+	// In the worst case, the database write caused by each state is roughly 4KB,
+	// 256MB in total is still acceptable.
+	stateWriteBatch = 65536
+)
+
+// indexBlockDesc is the descriptor of an index block that contains a list of
+// state mutation records belonging to a specific state (account or storage slot).
+type indexBlockDesc struct {
+	min     uint64
+	max     uint64
+	entries uint32
+	id      uint32
+}
+
+func newIndexBlockDesc(id uint32) *indexBlockDesc {
+	return &indexBlockDesc{id: id}
+}
+
+func (d *indexBlockDesc) empty() bool {
+	return d.entries == 0
+}
+
+func (d *indexBlockDesc) full() bool {
+	return d.entries >= indexBlockEntriesCap
+}
+
+// encode packs index block descriptor into byte stream.
+func (d *indexBlockDesc) encode() []byte {
+	var buf [indexBlockDescSize]byte
+	binary.BigEndian.PutUint64(buf[:8], d.min)
+	binary.BigEndian.PutUint64(buf[8:16], d.max)
+	binary.BigEndian.PutUint32(buf[16:20], d.entries)
+	binary.BigEndian.PutUint32(buf[20:24], d.id)
+	return buf[:]
+}
+
+// decode unpacks index block descriptor from byte stream.
+func (d *indexBlockDesc) decode(blob []byte) {
+	d.min = binary.BigEndian.Uint64(blob[:8])
+	d.max = binary.BigEndian.Uint64(blob[8:16])
+	d.entries = binary.BigEndian.Uint32(blob[16:20])
+	d.id = binary.BigEndian.Uint32(blob[20:24])
+}
+
+type blockWriter struct {
+	desc     *indexBlockDesc
+	restarts []uint32
+	scratch  []byte
+	buf      []byte
+}
+
+func newBlockWriter(blob []byte, desc *indexBlockDesc) (*blockWriter, error) {
+	scratch := make([]byte, binary.MaxVarintLen64)
+	if len(blob) == 0 {
+		return &blockWriter{
+			desc:    desc,
+			scratch: scratch,
+			buf:     make([]byte, 0, 1024),
+		}, nil
+	}
+	restarts, data, err := parseIndexBlock(blob)
+	if err != nil {
+		return nil, err
+	}
+	return &blockWriter{
+		desc:     desc,
+		restarts: restarts,
+		scratch:  scratch,
+		buf:      data, // safe to own the slice
+	}, nil
+}
+
+func (b *blockWriter) append(id uint64) error {
+	if id <= b.desc.max {
+		return fmt.Errorf("element out of order, last: %d, this: %d", b.desc.max, id)
+	}
+	if b.desc.entries%indexBlockRestartLen == 0 {
+		b.restarts = append(b.restarts, uint32(len(b.buf)))
+
+		// The restart point item can be either encoded in variable
+		// size or fixed size. Although variable-size encoding is
+		// slightly slower (2ns per operation), it is still relatively
+		// fast, therefore, it's picked for better space efficiency.
+		n := binary.PutUvarint(b.scratch[0:], id)
+		b.buf = append(b.buf, b.scratch[:n]...)
+	} else {
+		n := binary.PutUvarint(b.scratch[0:], id-b.desc.max)
+		b.buf = append(b.buf, b.scratch[:n]...)
+	}
+	b.desc.entries++
+	if b.desc.min == 0 {
+		b.desc.min = id
+	}
+	b.desc.max = id
+	return nil
+}
+
+func (b *blockWriter) empty() bool {
+	return b.desc.empty()
+}
+
+func (b *blockWriter) full() bool {
+	return b.desc.full()
+}
+
+func (b *blockWriter) finish() error {
+	b.restarts = append(b.restarts, uint32(len(b.restarts)))
+	for _, number := range b.restarts {
+		binary.BigEndian.PutUint32(b.scratch[:4], number)
+		b.buf = append(b.buf, b.scratch[:4]...)
+	}
+	return nil
+}
+
+type indexWriter struct {
+	descList []*indexBlockDesc
+	last     uint64
+	bw       *blockWriter
+	frozen   []*blockWriter
+
+	db    ethdb.KeyValueStore
+	addr  common.Address
+	state common.Hash
+}
+
+func newIndexWriter(db ethdb.KeyValueStore, addr common.Address, state common.Hash) (*indexWriter, error) {
+	blob := rawdb.ReadStateIndex(db, addr, state)
+	if len(blob) == 0 {
+		desc := &indexBlockDesc{}
+		bw, _ := newBlockWriter(nil, desc)
+		return &indexWriter{
+			descList: []*indexBlockDesc{desc},
+			bw:       bw,
+			db:       db,
+			addr:     addr,
+			state:    state,
+		}, nil
+	}
+	descList, err := parseIndex(blob)
+	if err != nil {
+		return nil, err
+	}
+	// Open the last block writer, or create a new one in case
+	// it's already full.
+	var (
+		lastDesc = descList[len(descList)-1]
+		lastElem = lastDesc.max
+	)
+	if lastDesc.full() {
+		descList = append(descList, &indexBlockDesc{id: lastDesc.id + 1})
+		lastDesc = descList[len(descList)-1]
+	}
+
+	indexBlock := rawdb.ReadStateIndexBlock(db, addr, state, lastDesc.id)
+	bw, err := newBlockWriter(indexBlock, lastDesc)
+	if err != nil {
+		return nil, err
+	}
+	return &indexWriter{
+		descList: descList,
+		last:     lastElem,
+		bw:       bw,
+		db:       db,
+		addr:     addr,
+		state:    state,
+	}, nil
+}
+
+func (w *indexWriter) append(id uint64) error {
+	if id <= w.last {
+		return fmt.Errorf("element out of order, last: %d, this: %d", w.last, id)
+	}
+	if err := w.bw.append(id); err != nil {
+		return err
+	}
+	w.last = id
+
+	if w.bw.full() {
+		w.rotate()
+	}
+	return nil
+}
+
+func (w *indexWriter) rotate() {
+	w.frozen = append(w.frozen, w.bw)
+	desc := newIndexBlockDesc(w.bw.desc.id + 1)
+	w.bw, _ = newBlockWriter(nil, desc)
+	w.descList = append(w.descList, desc)
+}
+
+func (w *indexWriter) finish(batch ethdb.Batch) error {
+	var (
+		writers  = append(w.frozen, w.bw)
+		descList = w.descList
+	)
+	// Chop the last block if it's empty
+	if w.bw.empty() {
+		writers = writers[:len(writers)-1]
+		descList = descList[:len(descList)-1]
+	}
+	if len(writers) == 0 {
+		return nil
+	}
+	for _, bw := range writers {
+		if err := bw.finish(); err != nil {
+			return err
+		}
+		rawdb.WriteStateIndexBlock(batch, w.addr, w.state, bw.desc.id, bw.buf)
+	}
+	buf := make([]byte, 0, indexBlockDescSize*len(descList))
+	for _, desc := range descList {
+		buf = append(buf, desc.encode()...)
+	}
+	rawdb.WriteStateIndex(batch, w.addr, w.state, buf)
+	return nil
+}
+
+type historyWriter struct {
+	accounts map[common.Address][]uint64
+	storages map[common.Address]map[common.Hash][]uint64
+	total    int
+}
+
+func newHistoryWriter() *historyWriter {
+	return &historyWriter{
+		accounts: make(map[common.Address][]uint64),
+		storages: make(map[common.Address]map[common.Hash][]uint64),
+	}
+}
+
+func (w *historyWriter) reset() {
+	w.total = 0
+	w.accounts = make(map[common.Address][]uint64)
+	w.storages = make(map[common.Address]map[common.Hash][]uint64)
+}
+
+func (w *historyWriter) addAccount(addr common.Address, number uint64) {
+	w.total += 1
+	w.accounts[addr] = append(w.accounts[addr], number)
+}
+
+func (w *historyWriter) addSlot(addr common.Address, hash common.Hash, number uint64) {
+	w.total += 1
+	if _, ok := w.storages[addr]; !ok {
+		w.storages[addr] = make(map[common.Hash][]uint64)
+	}
+	w.storages[addr][hash] = append(w.storages[addr][hash], number)
+}
+
+func (w *historyWriter) finish(db ethdb.KeyValueStore, force bool, head uint64) error {
+	if !force && w.total < stateWriteBatch {
+		return nil
+	}
+	var (
+		s     = time.Now()
+		batch = db.NewBatch()
+	)
+	for account, idList := range w.accounts {
+		iw, err := newIndexWriter(db, account, common.Hash{})
+		if err != nil {
+			return err
+		}
+		for _, id := range idList {
+			if err := iw.append(id); err != nil {
+				return err
+			}
+		}
+		if err := iw.finish(batch); err != nil {
+			return err
+		}
+	}
+	for account, slots := range w.storages {
+		for slot, idList := range slots {
+			iw, err := newIndexWriter(db, account, slot)
+			if err != nil {
+				return err
+			}
+			for _, id := range idList {
+				if err := iw.append(id); err != nil {
+					return err
+				}
+			}
+			if err := iw.finish(batch); err != nil {
+				return err
+			}
+		}
+	}
+	rawdb.WriteStateHistoryIndexHead(batch, head)
+	if err := batch.Write(); err != nil {
+		return err
+	}
+	w.reset()
+	log.Info("Written state history indexes", "head", head, "size", common.StorageSize(batch.ValueSize()), "elapsed", common.PrettyDuration(time.Since(s)))
+	return nil
+}

triedb/pathdb/history_indexer.go

@@ -0,0 +1,210 @@
+// 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 <http://www.gnu.org/licenses/
+
+package pathdb
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+const historyReadBatch = 1000 // The batch size for reading state history
+
+type historyIndexer struct {
+	disk    ethdb.KeyValueStore
+	freezer ethdb.AncientStore
+	headCh  chan uint64
+	closeCh chan struct{}
+	wg      sync.WaitGroup
+}
+
+func newHistoryIndexer(disk ethdb.KeyValueStore, freezer ethdb.AncientStore, head uint64) *historyIndexer {
+	indexer := &historyIndexer{
+		disk:    disk,
+		freezer: freezer,
+		headCh:  make(chan uint64),
+		closeCh: make(chan struct{}),
+	}
+	indexer.wg.Add(1)
+	go indexer.loop(head)
+	return indexer
+}
+
+func (i *historyIndexer) close() {
+	select {
+	case <-i.closeCh:
+		return
+	default:
+		close(i.closeCh)
+		i.wg.Wait()
+	}
+}
+
+func (i *historyIndexer) notify(head uint64) error {
+	select {
+	case <-i.closeCh:
+		return errors.New("closed")
+	case i.headCh <- head:
+	}
+	return nil
+}
+
+func (i *historyIndexer) process(w *historyWriter, h *history, id uint64) error {
+	for _, account := range h.accountList {
+		w.addAccount(account, id)
+		for _, slot := range h.storageList[account] {
+			w.addSlot(account, slot, id)
+		}
+	}
+	return w.finish(i.disk, false, id)
+}
+
+func (i *historyIndexer) next() (uint64, error) {
+	tail, err := i.freezer.Tail()
+	if err != nil {
+		return 0, err
+	}
+	tailID := tail + 1 // compute the real history id
+
+	// Start indexing from scratch if nothing has been indexed
+	head := rawdb.ReadStateHistoryIndexHead(i.disk)
+	if head == nil {
+		return tailID, nil
+	}
+	// Resume indexing from the last interrupted position
+	if *head+1 >= tailID {
+		return *head + 1, nil
+	}
+	// History has been shortened without indexing. Discard the gapped segment
+	// in the history and shift to the first available element.
+	//
+	// The missing indexes corresponding to the gapped histories won't be visible.
+	// It's fine to leave them unindexed.
+	log.Info("History gap detected, discard old segment", "oldHead", *head, "newHead", tailID)
+	return tailID, nil
+}
+
+func (i *historyIndexer) run(done chan struct{}, head uint64, interrupt *atomic.Int32) {
+	defer close(done)
+
+	begin, err := i.next()
+	if err != nil {
+		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 {
+	//
+	//}
+	log.Info("Start history indexing", "begin", begin, "head", head)
+
+	var (
+		current = begin
+		writer  = newHistoryWriter()
+		start   = time.Now()
+		logged  = time.Now()
+	)
+	for current <= head {
+		count := head - current + 1
+		if count > historyReadBatch {
+			count = historyReadBatch
+		}
+		s := time.Now()
+		result, err := readHistories(i.freezer, current, count)
+		if err != nil {
+			log.Error("Failed to read history", "err", err)
+			return
+		}
+		log.Debug("Loaded histories", "number", len(result), "elapsed", common.PrettyDuration(time.Since(s)))
+
+		for _, h := range result {
+			if err := i.process(writer, h, current); err != nil {
+				log.Error("Failed to index history", "err", err)
+				return
+			}
+			current += 1
+
+			if time.Since(logged) > time.Second*8 {
+				logged = time.Now()
+
+				var (
+					left  = head - current
+					done  = current - begin
+					speed = done/uint64(time.Since(start)/time.Millisecond+1) + 1 // +1s to avoid division by zero
+				)
+				// Override the ETA if larger than the largest until now
+				eta := time.Duration(left/speed) * time.Millisecond
+				log.Info("Indexing state history", "processed", current-begin+1, "remain", head-current, "eta", common.PrettyDuration(eta))
+			}
+		}
+		// Check interruption signal and abort process if it's fired
+		if interrupt != nil {
+			if signal := interrupt.Load(); signal != 0 {
+				if err := writer.finish(i.disk, true, current-1); err != nil {
+					log.Error("Failed to flush index", "err", err)
+				}
+				log.Info("State indexing interrupted")
+				return
+			}
+		}
+	}
+	if err := writer.finish(i.disk, true, head); err != nil {
+		log.Error("Failed to flush index", "err", err)
+	}
+	log.Info("Indexed state history", "from", begin, "to", head, "elapsed", common.PrettyDuration(time.Since(start)))
+}
+
+func (i *historyIndexer) loop(head uint64) {
+	defer i.wg.Done()
+
+	// Launch background indexing thread
+	done, interrupt := make(chan struct{}), new(atomic.Int32)
+	go i.run(done, head, interrupt)
+
+	for {
+		select {
+		case newHead := <-i.headCh:
+			if newHead <= head {
+				// TODO, reorg??
+				continue
+			}
+			head = newHead
+
+			if done == nil {
+				done, interrupt = make(chan struct{}), new(atomic.Int32)
+				go i.run(done, head, interrupt)
+			}
+		case <-done:
+			done, interrupt = nil, nil
+
+		case <-i.closeCh:
+			if done != nil {
+				interrupt.Store(1)
+				log.Info("Waiting background history indexer to exit")
+				<-done
+			}
+			return
+		}
+	}
+}