cerc-io/plugeth-statediff
16
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

rework iterator state stuff
Some checks failed
Test / Run unit tests (pull_request) Failing after 15m46s
Details
Test / Run integration tests (pull_request) Successful in 30m5s
Details

Browse Source
This commit is contained in:
Roy Crihfield 2023-09-26 18:07:25 +08:00
parent 5b51346a83
commit 100c106ee5
4 changed files with 43 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
builder.go
View File

@ -148,9 +148,9 @@ func (sdb *builder) WriteStateDiff(
func(subdiv uint) { func(subdiv uint) {
g.Go(func() error { g.Go(func() error {
a, b := subitersA[subdiv], subitersB[subdiv] a, b := subitersA[subdiv], subitersB[subdiv]
it, aux := utils.NewSymmetricDifferenceIterator(a, b) it := utils.NewSymmetricDifferenceIterator(a, b)
return sdb.processAccounts(ctx, return sdb.processAccounts(ctx,
it, aux, it, &it.SymmDiffState,
params.watchedAddressesLeafPaths, params.watchedAddressesLeafPaths,
nodeSink, ipldSink, logger, nodeSink, ipldSink, logger,
) )
@ -165,7 +165,7 @@ func (sdb *builder) WriteStateDiffTracked(
args Args, params Params, args Args, params Params,
nodeSink sdtypes.StateNodeSink, nodeSink sdtypes.StateNodeSink,
ipldSink sdtypes.IPLDSink, ipldSink sdtypes.IPLDSink,
tracker tracker.Tracker, tracker tracker.IteratorTracker,
) error { ) error {
defer metrics.UpdateDuration(time.Now(), metrics.IndexerMetrics.WriteStateDiffTimer) defer metrics.UpdateDuration(time.Now(), metrics.IndexerMetrics.WriteStateDiffTimer)
// Load tries for old and new states // Load tries for old and new states
@ -178,18 +178,14 @@ func (sdb *builder) WriteStateDiffTracked(
return fmt.Errorf("error opening new state trie: %w", err) return fmt.Errorf("error opening new state trie: %w", err)
} }
var subiters []trie.NodeIterator var subiters, bases []trie.NodeIterator
var auxes []*utils.SymmDiffAux
// Constructor for difference iterator at a specific (recovered) path // Constructor for difference iterator at a specific (recovered) path
makeIterator := func(key []byte) trie.NodeIterator { makeIterator := func(key []byte) trie.NodeIterator {
a := triea.NodeIterator(key) a := triea.NodeIterator(key)
b := trieb.NodeIterator(key) b := trieb.NodeIterator(key)
diffit, aux := utils.NewSymmetricDifferenceIterator(a, b) return utils.NewSymmetricDifferenceIterator(a, b)
// iterators are constructed in-order, so these will align
auxes = append(auxes, aux)
return diffit
} }
subiters, err = tracker.Restore(makeIterator) subiters, bases, err = tracker.Restore(makeIterator)
if err != nil { if err != nil {
return fmt.Errorf("error restoring iterators: %w", err) return fmt.Errorf("error restoring iterators: %w", err)
} }
@ -214,7 +210,7 @@ func (sdb *builder) WriteStateDiffTracked(
func(subdiv uint) { func(subdiv uint) {
g.Go(func() error { g.Go(func() error {
return sdb.processAccounts(ctx, return sdb.processAccounts(ctx,
subiters[subdiv], auxes[subdiv], subiters[subdiv], &bases[subdiv].(*utils.SymmDiffIterator).SymmDiffState,
params.watchedAddressesLeafPaths, params.watchedAddressesLeafPaths,
nodeSink, ipldSink, logger, nodeSink, ipldSink, logger,
) )
@ -225,9 +221,10 @@ func (sdb *builder) WriteStateDiffTracked(
} }
// processAccounts processes account creations, deletions, and updates // processAccounts processes account creations, deletions, and updates
// the NodeIterator and SymmDiffIterator instances should refer to the same object, will only be used
func (sdb *builder) processAccounts( func (sdb *builder) processAccounts(
ctx context.Context, ctx context.Context,
it trie.NodeIterator, aux *utils.SymmDiffAux, it trie.NodeIterator, symdiff *utils.SymmDiffState,
watchedAddressesLeafPaths [][]byte, watchedAddressesLeafPaths [][]byte,
nodeSink sdtypes.StateNodeSink, ipldSink sdtypes.IPLDSink, nodeSink sdtypes.StateNodeSink, ipldSink sdtypes.IPLDSink,
logger log.Logger, logger log.Logger,
@ -250,7 +247,7 @@ func (sdb *builder) processAccounts(
if !isWatchedPathPrefix(watchedAddressesLeafPaths, it.Path()) { if !isWatchedPathPrefix(watchedAddressesLeafPaths, it.Path()) {
continue continue
} }
if aux.FromA() { // Node exists in the old trie if symdiff.FromA() { // Node exists in the old trie
if it.Leaf() { if it.Leaf() {
var account types.StateAccount var account types.StateAccount
if err := rlp.DecodeBytes(it.LeafBlob(), &account); err != nil { if err := rlp.DecodeBytes(it.LeafBlob(), &account); err != nil {
@ -259,7 +256,7 @@ func (sdb *builder) processAccounts(
leafKey := make([]byte, len(it.LeafKey())) leafKey := make([]byte, len(it.LeafKey()))
copy(leafKey, it.LeafKey()) copy(leafKey, it.LeafKey())
if aux.CommonPath() { if symdiff.CommonPath() {
// If B also contains this leaf node, this is the old state of an updated account. // If B also contains this leaf node, this is the old state of an updated account.
if update, ok := updates[string(leafKey)]; ok { if update, ok := updates[string(leafKey)]; ok {
update.oldRoot = account.Root update.oldRoot = account.Root
@ -284,7 +281,7 @@ func (sdb *builder) processAccounts(
return err return err
} }
if aux.CommonPath() { if symdiff.CommonPath() {
// If A also contains this leaf node, this is the new state of an updated account. // If A also contains this leaf node, this is the new state of an updated account.
if update, ok := updates[string(accountW.LeafKey)]; ok { if update, ok := updates[string(accountW.LeafKey)]; ok {
update.new = *accountW update.new = *accountW
@ -354,7 +351,7 @@ func (sdb *builder) processAccounts(
} }
} }
metrics.IndexerMetrics.DifferenceIteratorCounter.Inc(int64(aux.Count())) metrics.IndexerMetrics.DifferenceIteratorCounter.Inc(int64(symdiff.Count()))
return it.Error() return it.Error()
} }
@ -482,10 +479,10 @@ func (sdb *builder) processStorageUpdates(
var prevBlob []byte var prevBlob []byte
a, b := oldTrie.NodeIterator(nil), newTrie.NodeIterator(nil) a, b := oldTrie.NodeIterator(nil), newTrie.NodeIterator(nil)
it, aux := utils.NewSymmetricDifferenceIterator(a, b) it := utils.NewSymmetricDifferenceIterator(a, b)
for it.Next(true) { for it.Next(true) {
if aux.FromA() { if it.FromA() {
if it.Leaf() && !aux.CommonPath() { if it.Leaf() && !it.CommonPath() {
// If this node's leaf key is absent from B, the storage slot was vacated. // If this node's leaf key is absent from B, the storage slot was vacated.
// In that case, emit an empty "removed" storage node record. // In that case, emit an empty "removed" storage node record.
if err := storageSink(sdtypes.StorageLeafNode{ if err := storageSink(sdtypes.StorageLeafNode{

2
go.mod
View File

@ -124,7 +124,7 @@ require (
) )
replace ( replace (
github.com/cerc-io/eth-iterator-utils => git.vdb.to/cerc-io/eth-iterator-utils v0.1.2-0.20230925184550-062eb329435f github.com/cerc-io/eth-iterator-utils => git.vdb.to/cerc-io/eth-iterator-utils v0.1.2-0.20230926100620-802551012643
github.com/cerc-io/eth-testing => git.vdb.to/cerc-io/eth-testing v0.3.1-0.20230925181540-2ea71042e7e0 github.com/cerc-io/eth-testing => git.vdb.to/cerc-io/eth-testing v0.3.1-0.20230925181540-2ea71042e7e0
github.com/ethereum/go-ethereum => git.vdb.to/cerc-io/plugeth v0.0.0-20230808125822-691dc334fab1 github.com/ethereum/go-ethereum => git.vdb.to/cerc-io/plugeth v0.0.0-20230808125822-691dc334fab1
github.com/openrelayxyz/plugeth-utils => git.vdb.to/cerc-io/plugeth-utils v0.0.0-20230706160122-cd41de354c46 github.com/openrelayxyz/plugeth-utils => git.vdb.to/cerc-io/plugeth-utils v0.0.0-20230706160122-cd41de354c46

4
go.sum
View File

@ -1,6 +1,6 @@
cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
git.vdb.to/cerc-io/eth-iterator-utils v0.1.2-0.20230925184550-062eb329435f h1:sIuSkD6U7uYD/FGfvWOBViIuaHd+YhLM0Hln+4BQM10= git.vdb.to/cerc-io/eth-iterator-utils v0.1.2-0.20230926100620-802551012643 h1:yJFyJgGVy1RMEJqPrTYyaB7fF1wpfx0Df5Bsunb+Lyg=
git.vdb.to/cerc-io/eth-iterator-utils v0.1.2-0.20230925184550-062eb329435f/go.mod h1:Xv+d7Q11qGJcggcfxoj2JEvJJBKj0C66I6PyG5/lz9o= git.vdb.to/cerc-io/eth-iterator-utils v0.1.2-0.20230926100620-802551012643/go.mod h1:Xv+d7Q11qGJcggcfxoj2JEvJJBKj0C66I6PyG5/lz9o=
git.vdb.to/cerc-io/eth-testing v0.3.1-0.20230925181540-2ea71042e7e0 h1:fWAvsSiuDqveuxwnfc8psInfLZhMqHlQnmOpOHsd8Tk= git.vdb.to/cerc-io/eth-testing v0.3.1-0.20230925181540-2ea71042e7e0 h1:fWAvsSiuDqveuxwnfc8psInfLZhMqHlQnmOpOHsd8Tk=
git.vdb.to/cerc-io/eth-testing v0.3.1-0.20230925181540-2ea71042e7e0/go.mod h1:qdvpc/W1xvf2MKx3rMOqvFvYaYIHG77Z1g0lwsmw0Uk= git.vdb.to/cerc-io/eth-testing v0.3.1-0.20230925181540-2ea71042e7e0/go.mod h1:qdvpc/W1xvf2MKx3rMOqvFvYaYIHG77Z1g0lwsmw0Uk=
git.vdb.to/cerc-io/plugeth v0.0.0-20230808125822-691dc334fab1 h1:KLjxHwp9Zp7xhECccmJS00RiL+VwTuUGLU7qeIctg8g= git.vdb.to/cerc-io/plugeth v0.0.0-20230808125822-691dc334fab1 h1:KLjxHwp9Zp7xhECccmJS00RiL+VwTuUGLU7qeIctg8g=

48
utils/iterator.go
View File

@ -7,9 +7,9 @@ import (
"github.com/ethereum/go-ethereum/trie" "github.com/ethereum/go-ethereum/trie"
) )
type symmDiffIterator struct { type SymmDiffIterator struct {
a, b iterState // Nodes returned are those in b - a and a - b (keys only) a, b iterState // Nodes returned are those in b - a and a - b (keys only)
SymmDiffAux SymmDiffState
} }
// pairs an iterator with a cache of its valid status // pairs an iterator with a cache of its valid status
@ -18,10 +18,10 @@ type iterState struct {
valid bool valid bool
} }
// SymmDiffAux exposes state specific to symmetric difference iteration, which is not accessible // SymmDiffState exposes state specific to symmetric difference iteration, which is not accessible
// from the NodeIterator interface. This includes the number of nodes seen, whether the current key // from the NodeIterator interface. This includes the number of nodes seen, whether the current key
// is common to both A and B, and whether the current node is sourced from A or B. // is common to both A and B, and whether the current node is sourced from A or B.
type SymmDiffAux struct { type SymmDiffState struct {
yieldFromA bool // Whether next node comes from a yieldFromA bool // Whether next node comes from a
count int // Number of nodes scanned on either trie count int // Number of nodes scanned on either trie
eqPathIndex int // Count index of last pair of equal paths, to detect an updated key eqPathIndex int // Count index of last pair of equal paths, to detect an updated key
@ -30,14 +30,14 @@ type SymmDiffAux struct {
// NewSymmetricDifferenceIterator constructs a trie.NodeIterator that iterates over the symmetric difference // NewSymmetricDifferenceIterator constructs a trie.NodeIterator that iterates over the symmetric difference
// of elements in a and b, i.e., the elements in a that are not in b, and vice versa. // of elements in a and b, i.e., the elements in a that are not in b, and vice versa.
// Returns the iterator, and a pointer to an auxiliary object for accessing the state not exposed by the NodeIterator interface recording the number of nodes seen. // Returns the iterator, and a pointer to an auxiliary object for accessing the state not exposed by the NodeIterator interface recording the number of nodes seen.
func NewSymmetricDifferenceIterator(a, b trie.NodeIterator) (trie.NodeIterator, *SymmDiffAux) { func NewSymmetricDifferenceIterator(a, b trie.NodeIterator) *SymmDiffIterator {
it := &symmDiffIterator{ it := &SymmDiffIterator{
a: iterState{a, true}, a: iterState{a, true},
b: iterState{b, true}, b: iterState{b, true},
// common paths are detected by a distance <=1 between count and this index, so we start at -2 // common paths are detected by a distance <=1 between count and this index, so we start at -2
SymmDiffAux: SymmDiffAux{eqPathIndex: -2}, SymmDiffState: SymmDiffState{eqPathIndex: -2},
} }
return it, &it.SymmDiffAux return it
} }
func (st *iterState) Next(descend bool) bool { func (st *iterState) Next(descend bool) bool {
@ -46,65 +46,65 @@ func (st *iterState) Next(descend bool) bool {
} }
// FromA returns true if the current node is sourced from A. // FromA returns true if the current node is sourced from A.
func (it *SymmDiffAux) FromA() bool { func (it *SymmDiffState) FromA() bool {
return it.yieldFromA return it.yieldFromA
} }
// CommonPath returns true if a node with the current path exists in each sub-iterator - i.e. it // CommonPath returns true if a node with the current path exists in each sub-iterator - i.e. it
// represents an updated node. // represents an updated node.
func (it *SymmDiffAux) CommonPath() bool { func (it *SymmDiffState) CommonPath() bool {
return it.count-it.eqPathIndex <= 1 return it.count-it.eqPathIndex <= 1
} }
// Count returns the number of nodes seen. // Count returns the number of nodes seen.
func (it *SymmDiffAux) Count() int { func (it *SymmDiffState) Count() int {
return it.count return it.count
} }
func (it *symmDiffIterator) curr() *iterState { func (it *SymmDiffIterator) curr() *iterState {
if it.yieldFromA { if it.yieldFromA {
return &it.a return &it.a
} }
return &it.b return &it.b
} }
func (it *symmDiffIterator) Hash() common.Hash { func (it *SymmDiffIterator) Hash() common.Hash {
return it.curr().Hash() return it.curr().Hash()
} }
func (it *symmDiffIterator) Parent() common.Hash { func (it *SymmDiffIterator) Parent() common.Hash {
return it.curr().Parent() return it.curr().Parent()
} }
func (it *symmDiffIterator) Leaf() bool { func (it *SymmDiffIterator) Leaf() bool {
return it.curr().Leaf() return it.curr().Leaf()
} }
func (it *symmDiffIterator) LeafKey() []byte { func (it *SymmDiffIterator) LeafKey() []byte {
return it.curr().LeafKey() return it.curr().LeafKey()
} }
func (it *symmDiffIterator) LeafBlob() []byte { func (it *SymmDiffIterator) LeafBlob() []byte {
return it.curr().LeafBlob() return it.curr().LeafBlob()
} }
func (it *symmDiffIterator) LeafProof() [][]byte { func (it *SymmDiffIterator) LeafProof() [][]byte {
return it.curr().LeafProof() return it.curr().LeafProof()
} }
func (it *symmDiffIterator) Path() []byte { func (it *SymmDiffIterator) Path() []byte {
return it.curr().Path() return it.curr().Path()
} }
func (it *symmDiffIterator) NodeBlob() []byte { func (it *SymmDiffIterator) NodeBlob() []byte {
return it.curr().NodeBlob() return it.curr().NodeBlob()
} }
func (it *symmDiffIterator) AddResolver(resolver trie.NodeResolver) { func (it *SymmDiffIterator) AddResolver(resolver trie.NodeResolver) {
panic("not implemented") panic("not implemented")
} }
func (it *symmDiffIterator) Next(bool) bool { func (it *SymmDiffIterator) Next(bool) bool {
// NodeIterators start in a "pre-valid" state, so the first Next advances to a valid node. // NodeIterators start in a "pre-valid" state, so the first Next advances to a valid node.
if it.count == 0 { if it.count == 0 {
if it.a.Next(true) { if it.a.Next(true) {
@ -122,7 +122,7 @@ func (it *symmDiffIterator) Next(bool) bool {
return it.a.valid || it.b.valid return it.a.valid || it.b.valid
} }
func (it *symmDiffIterator) seek() { func (it *SymmDiffIterator) seek() {
// Invariants: // Invariants:
// - At the end of the function, the sub-iterator with the lexically lesser path // - At the end of the function, the sub-iterator with the lexically lesser path
// points to the next element // points to the next element
@ -163,7 +163,7 @@ func (it *symmDiffIterator) seek() {
} }
} }
func (it *symmDiffIterator) Error() error { func (it *SymmDiffIterator) Error() error {
if err := it.a.Error(); err != nil { if err := it.a.Error(); err != nil {
return err return err
} }