Cerc refactor (#43)

* pointing to unpublished versions

* pointing to unpublished versions

* updated for bumped go-eth-state-node-iterator and other cerc refactors

README.md

@@ -2,7 +2,7 @@
 
 [![Go Report Card](https://goreportcard.com/badge/github.com/vulcanize/eth-ipfs-state-validator)](https://goreportcard.com/report/github.com/vulcanize/eth-ipfs-state-validator)
 
-> Uses [ipfs-ethdb](https://github.com/vulcanize/ipfs-ethdb/tree/master/postgres) to validate completeness of IPFS Ethereum state data
+> Uses [ipfs-ethdb](https://github.com/cerc-io/ipfs-ethdb/tree/master/postgres) to validate completeness of IPFS Ethereum state data
 
 ## Background
 

cmd/validateTrie.go

@@ -24,7 +24,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
-	validator "github.com/vulcanize/eth-ipfs-state-validator/v4/pkg"
+	validator "github.com/cerc-io/eth-ipfs-state-validator/v4/pkg"
 )
 
 // validateTrieCmd represents the validateTrie command
@@ -59,14 +59,19 @@ It can operate at three levels:
 }
 
 func validateTrie() {
-	v, err := newValidator()
+	params := validator.Params{
+		Workers:        viper.GetUint("validator.workers"),
+		RecoveryFormat: viper.GetString("validator.recoveryFormat"),
+	}
+	v, err := newValidator(params)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 	stateRootStr := viper.GetString("validator.stateRoot")
 	storageRootStr := viper.GetString("validator.storageRoot")
 	contractAddrStr := viper.GetString("validator.address")
-	switch strings.ToLower(viper.GetString("validator.type")) {
+	traversal := strings.ToLower(viper.GetString("validator.type"))
+	switch traversal {
 	case "f", "full":
 		if stateRootStr == "" {
 			logWithCommand.Fatal("must provide a state root for full state validation")
@@ -98,26 +103,28 @@ func validateTrie() {
 			logWithCommand.Fatalf("Storage trie for contract %s and root %s not complete\r\nerr: %v", addr.String(), storageRoot.String(), err)
 		}
 		logWithCommand.Infof("Storage trie for contract %s and root %s is complete", addr.String(), storageRoot.String())
+	default:
+		logWithCommand.Fatalf("Invalid traversal level: '%s'", traversal)
 	}
 
 	stats := v.GetCacheStats()
 	logWithCommand.Debugf("groupcache stats %+v", stats)
 }
 
-func newValidator() (*validator.Validator, error) {
+func newValidator(params validator.Params) (*validator.Validator, error) {
 	ipfsPath := viper.GetString("ipfs.path")
 	if ipfsPath == "" {
 		db, err := validator.NewDB()
 		if err != nil {
 			logWithCommand.Fatal(err)
 		}
-		return validator.NewPGIPFSValidator(db), nil
+		return validator.NewPGIPFSValidator(db, params), nil
 	}
 	bs, err := validator.InitIPFSBlockService(ipfsPath)
 	if err != nil {
 		return nil, err
 	}
-	return validator.NewIPFSValidator(bs), nil
+	return validator.NewIPFSValidator(bs, params), nil
 }
 
 func init() {
@@ -128,10 +135,14 @@ func init() {
 	validateTrieCmd.PersistentFlags().String("storage-root", "", "Root of the storage trie we wish to validate; for storage validation")
 	validateTrieCmd.PersistentFlags().String("address", "", "Contract address for the storage trie we wish to validate; for storage validation")
 	validateTrieCmd.PersistentFlags().String("ipfs-path", "", "Path to IPFS repository; if provided operations move through the IPFS repo otherwise Postgres connection params are expected in the provided config")
+	validateTrieCmd.PersistentFlags().Int("workers", 4, "number of concurrent workers to use")
+	validateTrieCmd.PersistentFlags().String("recovery-format", validator.DefaultRecoveryFormat, "format pattern for recovery files")
 
 	viper.BindPFlag("validator.stateRoot", validateTrieCmd.PersistentFlags().Lookup("state-root"))
 	viper.BindPFlag("validator.type", validateTrieCmd.PersistentFlags().Lookup("type"))
 	viper.BindPFlag("validator.storageRoot", validateTrieCmd.PersistentFlags().Lookup("storage-root"))
 	viper.BindPFlag("validator.address", validateTrieCmd.PersistentFlags().Lookup("address"))
+	viper.BindPFlag("validator.workers", validateTrieCmd.PersistentFlags().Lookup("workers"))
+	viper.BindPFlag("validator.recoveryFormat", validateTrieCmd.PersistentFlags().Lookup("recovery-format"))
 	viper.BindPFlag("ipfs.path", validateTrieCmd.PersistentFlags().Lookup("ipfs-path"))
 }

go.mod

@@ -1,24 +1,26 @@
-module github.com/vulcanize/eth-ipfs-state-validator/v4
+module github.com/cerc-io/eth-ipfs-state-validator/v4
 
 go 1.18
 
 require (
-	github.com/ethereum/go-ethereum v1.10.21
+	github.com/cerc-io/go-eth-state-node-iterator v1.1.7
+	github.com/cerc-io/ipfs-ethdb/v4 v4.0.8-alpha
+	github.com/ethereum/go-ethereum v1.10.23
 	github.com/ipfs/go-blockservice v0.4.0
 	github.com/ipfs/go-cid v0.2.0
 	github.com/ipfs/go-ipfs-blockstore v1.2.0
 	github.com/ipfs/go-ipfs-ds-help v1.1.0
 	github.com/ipfs/kubo v0.14.0
 	github.com/jmoiron/sqlx v1.3.5
-	github.com/lib/pq v1.10.5
+	github.com/lib/pq v1.10.6
 	github.com/mailgun/groupcache/v2 v2.3.0
 	github.com/multiformats/go-multihash v0.2.0
 	github.com/onsi/ginkgo v1.16.5
 	github.com/onsi/gomega v1.19.0
-	github.com/sirupsen/logrus v1.8.1
+	github.com/sirupsen/logrus v1.9.0
 	github.com/spf13/cobra v1.4.0
 	github.com/spf13/viper v1.11.0
-	github.com/vulcanize/ipfs-ethdb/v4 v4.0.5-alpha
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 )
 
 require (
@@ -219,11 +221,10 @@ require (
 	go.uber.org/multierr v1.8.0 // indirect
 	go.uber.org/zap v1.21.0 // indirect
 	go4.org v0.0.0-20200411211856-f5505b9728dd // indirect
-	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
+	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
 	golang.org/x/net v0.0.0-20220607020251-c690dde0001d // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
+	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.10 // indirect
 	golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df // indirect
@@ -235,4 +236,4 @@ require (
 	lukechampine.com/blake3 v1.1.7 // indirect
 )
 
-replace github.com/ethereum/go-ethereum v1.10.21 => github.com/vulcanize/go-ethereum v1.10.21-statediff-4.1.2-alpha
+replace github.com/ethereum/go-ethereum v1.10.23 => github.com/cerc-io/go-ethereum v1.10.23-statediff-4.2.0-alpha

go.sum

@@ -132,6 +132,12 @@ github.com/cenkalti/backoff/v4 v4.1.3 h1:cFAlzYUlVYDysBEH2T5hyJZMh3+5+WCBvSnK6Q8
 github.com/cenkalti/backoff/v4 v4.1.3/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/ceramicnetwork/go-dag-jose v0.1.0 h1:yJ/HVlfKpnD3LdYP03AHyTvbm3BpPiz2oZiOeReJRdU=
+github.com/cerc-io/go-eth-state-node-iterator v1.1.7 h1:i4zCKkfowJ4aUVpqVyMpOZIbD5WVm/iijhMAPgcLJ44=
+github.com/cerc-io/go-eth-state-node-iterator v1.1.7/go.mod h1:BBh4BetpBoHUbAiKAdatR3ij7IzDlgN9nPzZ7snn/lo=
+github.com/cerc-io/go-ethereum v1.10.23-statediff-4.2.0-alpha h1:Na4mum/Qkw1mRHnq/TIRPIF5exZAEBznVs3G3/YSENo=
+github.com/cerc-io/go-ethereum v1.10.23-statediff-4.2.0-alpha/go.mod h1:lKBVBWksSwBDR/5D9CAxaGQzDPIS3ueWb6idy7X1Shg=
+github.com/cerc-io/ipfs-ethdb/v4 v4.0.8-alpha h1:+05O+US8WosZO9RP3i1j9B0QpBYwg9SlhJeBdRBgzgA=
+github.com/cerc-io/ipfs-ethdb/v4 v4.0.8-alpha/go.mod h1:3zlBgkDTgSLlgMgY8BttUzQxxVAzZQ1jZUH4qaP09Pc=
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -732,8 +738,8 @@ github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
-github.com/lib/pq v1.10.5 h1:J+gdV2cUmX7ZqL2B0lFcW0m+egaHC2V3lpO8nWxyYiQ=
-github.com/lib/pq v1.10.5/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/lib/pq v1.10.6 h1:jbk+ZieJ0D7EVGJYpL9QTz7/YW6UHbmdnZWYyK5cdBs=
+github.com/lib/pq v1.10.6/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/libp2p/go-addr-util v0.0.1/go.mod h1:4ac6O7n9rIAKB1dnd+s8IbbMXkt+oBpzX4/+RACcnlQ=
 github.com/libp2p/go-addr-util v0.0.2/go.mod h1:Ecd6Fb3yIuLzq4bD7VcywcVSBtefcAwnUISBM3WG15E=
 github.com/libp2p/go-addr-util v0.1.0/go.mod h1:6I3ZYuFr2O/9D+SoyM0zEw0EF3YkldtTX406BpdQMqw=
@@ -1438,8 +1444,9 @@ github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPx
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
-github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=
 github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
+github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
+github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/assertions v1.0.0 h1:UVQPSSmc3qtTi+zPPkCXvZX9VvW/xT/NsRvKfwY81a8=
 github.com/smartystreets/assertions v1.0.0/go.mod h1:kHHU4qYBaI3q23Pp3VPrmWhuIUrLW/7eUrw0BU5VaoM=
@@ -1519,10 +1526,6 @@ github.com/urfave/cli v1.22.2/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtX
 github.com/urfave/cli/v2 v2.0.0/go.mod h1:SE9GqnLQmjVa0iPEY0f1w3ygNIYcIJ0OKPMoW2caLfQ=
 github.com/viant/assertly v0.4.8/go.mod h1:aGifi++jvCrUaklKEKT0BU95igDNaqkvz+49uaYMPRU=
 github.com/viant/toolbox v0.24.0/go.mod h1:OxMCG57V0PXuIP2HNQrtJf2CjqdmbrOx5EkMILuUhzM=
-github.com/vulcanize/go-ethereum v1.10.21-statediff-4.1.2-alpha h1:ct+8FGuQnHA6SOGOQoAMINWdeexuSF40+IjF48J094A=
-github.com/vulcanize/go-ethereum v1.10.21-statediff-4.1.2-alpha/go.mod h1:dNJkmCSbaasX0zfQM6pm1g3rWlW3EGhLOEZMScyrRAs=
-github.com/vulcanize/ipfs-ethdb/v4 v4.0.5-alpha h1:NFRwWeMB3Q+QqLM9qdcHvfvWBxOk0lPwhOqXJpkIg30=
-github.com/vulcanize/ipfs-ethdb/v4 v4.0.5-alpha/go.mod h1:WvYj0m0cLPAtoytTbcbE2nZ3Hg9iuuF+lY14dBVRWZQ=
 github.com/wI2L/jsondiff v0.2.0 h1:dE00WemBa1uCjrzQUUTE/17I6m5qAaN0EMFOg2Ynr/k=
 github.com/wI2L/jsondiff v0.2.0/go.mod h1:axTcwtBkY4TsKuV+RgoMhHyHKKFRI6nnjRLi8LLYQnA=
 github.com/wangjia184/sortedset v0.0.0-20160527075905-f5d03557ba30/go.mod h1:YkocrP2K2tcw938x9gCOmT5G5eCD6jsTz0SZuyAqwIE=
@@ -1677,8 +1680,8 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210813211128-0a44fdfbc16e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20211108221036-ceb1ce70b4fa/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20220518034528-6f7dac969898/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e h1:T8NU3HyQ8ClP4SEE+KbFlg6n0NhuTsN4MyznaarGsZM=
-golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
+golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -1908,8 +1911,9 @@ golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220517195934-5e4e11fc645e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a h1:dGzPydgVsqGcTRVwiLJ1jVbufYwmzD3LfVPLKsKg+0k=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=

main.go

@@ -18,7 +18,7 @@ package main
 import (
 	"github.com/sirupsen/logrus"
 
-	"github.com/vulcanize/eth-ipfs-state-validator/v4/cmd"
+	"github.com/cerc-io/eth-ipfs-state-validator/v4/cmd"
 )
 
 func main() {

pkg/types.go

@@ -0,0 +1,9 @@
+package validator
+
+type TraversalType = string
+
+const (
+	fullTraversal    = "full"
+	stateTraversal   = "state"
+	storageTraversal = "storage"
+)

pkg/validator.go

@@ -17,19 +17,28 @@
 package validator
 
 import (
+	"bytes"
+	"context"
+	"fmt"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/trie"
 	"github.com/ipfs/go-blockservice"
 	"github.com/jmoiron/sqlx"
 	"github.com/mailgun/groupcache/v2"
+	log "github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
 
-	ipfsethdb "github.com/vulcanize/ipfs-ethdb/v4"
-	pgipfsethdb "github.com/vulcanize/ipfs-ethdb/v4/postgres"
+	nodeiter "github.com/cerc-io/go-eth-state-node-iterator"
+	"github.com/cerc-io/go-eth-state-node-iterator/tracker"
+	ipfsethdb "github.com/cerc-io/ipfs-ethdb/v4"
+	pgipfsethdb "github.com/cerc-io/ipfs-ethdb/v4/postgres"
 )
 
 // Validator is used for validating Ethereum state and storage tries on PG-IPFS
@@ -38,10 +47,22 @@ type Validator struct {
 	trieDB        *trie.Database
 	stateDatabase state.Database
 	db            *pgipfsethdb.Database
+
+	params Params
 }
 
+type Params struct {
+	Workers        uint
+	RecoveryFormat string // %s substituted with traversal type
+}
+
+var (
+	DefaultRecoveryFormat = "./recover_validate_%s"
+	emptyCodeHash         = crypto.Keccak256(nil)
+)
+
 // NewPGIPFSValidator returns a new trie validator ontop of a connection pool for an IPFS backing Postgres database
-func NewPGIPFSValidator(db *sqlx.DB) *Validator {
+func NewPGIPFSValidator(db *sqlx.DB, par Params) *Validator {
 	kvs := pgipfsethdb.NewKeyValueStore(db, pgipfsethdb.CacheConfig{
 		Name:           "kv",
 		Size:           16 * 1000 * 1000, // 16MB
@@ -54,11 +75,13 @@ func NewPGIPFSValidator(db *sqlx.DB) *Validator {
 		ExpiryDuration: time.Hour * 8,    // 8 hours
 	})
 
+	normalizeParams(&par)
 	return &Validator{
 		kvs:           kvs,
 		trieDB:        trie.NewDatabase(kvs),
 		stateDatabase: state.NewDatabase(database),
 		db:            database.(*pgipfsethdb.Database),
+		params:        par,
 	}
 }
 
@@ -67,13 +90,15 @@ func (v *Validator) GetCacheStats() groupcache.Stats {
 }
 
 // NewIPFSValidator returns a new trie validator ontop of an IPFS blockservice
-func NewIPFSValidator(bs blockservice.BlockService) *Validator {
+func NewIPFSValidator(bs blockservice.BlockService, par Params) *Validator {
 	kvs := ipfsethdb.NewKeyValueStore(bs)
 	database := ipfsethdb.NewDatabase(bs)
+	normalizeParams(&par)
 	return &Validator{
 		kvs:           kvs,
 		trieDB:        trie.NewDatabase(kvs),
 		stateDatabase: state.NewDatabase(database),
+		params:        par,
 	}
 }
 
@@ -88,21 +113,25 @@ func NewValidator(kvs ethdb.KeyValueStore, database ethdb.Database) *Validator {
 	}
 }
 
+// Ensure params are valid
+func normalizeParams(p *Params) {
+	if p.Workers == 0 {
+		p.Workers = 1
+	}
+	if len(p.RecoveryFormat) == 0 {
+		p.RecoveryFormat = DefaultRecoveryFormat
+	}
+}
+
 // ValidateTrie returns an error if the state and storage tries for the provided state root cannot be confirmed as complete
 // This does consider child storage tries
 func (v *Validator) ValidateTrie(stateRoot common.Hash) error {
-	// Generate the state.NodeIterator for this root
-	stateDB, err := state.New(stateRoot, v.stateDatabase, nil)
+	t, err := v.stateDatabase.OpenTrie(stateRoot)
 	if err != nil {
 		return err
 	}
-	it := state.NewNodeIterator(stateDB)
-	for it.Next() {
-		// iterate through entire state trie and descendent storage tries
-		// it.Next() will return false when we have either completed iteration of the entire trie or have ran into an error (e.g. a missing node)
-		// if we are able to iterate through the entire trie without error then the trie is complete
-	}
-	return it.Error
+	iterate := func(it trie.NodeIterator) error { return v.iterate(it, true) }
+	return iterateTracked(t, fmt.Sprintf(v.params.RecoveryFormat, fullTraversal), v.params.Workers, iterate)
 }
 
 // ValidateStateTrie returns an error if the state trie for the provided state root cannot be confirmed as complete
@@ -113,13 +142,8 @@ func (v *Validator) ValidateStateTrie(stateRoot common.Hash) error {
 	if err != nil {
 		return err
 	}
-	it := t.NodeIterator(nil)
-	for it.Next(true) {
-		// iterate through entire state trie
-		// it.Next() will return false when we have either completed iteration of the entire trie or have ran into an error (e.g. a missing node)
-		// if we are able to iterate through the entire trie without error then the trie is complete
-	}
-	return it.Error()
+	iterate := func(it trie.NodeIterator) error { return v.iterate(it, false) }
+	return iterateTracked(t, fmt.Sprintf(v.params.RecoveryFormat, stateTraversal), v.params.Workers, iterate)
 }
 
 // ValidateStorageTrie returns an error if the storage trie for the provided storage root and contract address cannot be confirmed as complete
@@ -130,13 +154,8 @@ func (v *Validator) ValidateStorageTrie(address common.Address, storageRoot comm
 	if err != nil {
 		return err
 	}
-	it := t.NodeIterator(nil)
-	for it.Next(true) {
-		// iterate through entire storage trie
-		// it.Next() will return false when we have either completed iteration of the entire trie or have ran into an error (e.g. a missing node)
-		// if we are able to iterate through the entire trie without error then the trie is complete
-	}
-	return it.Error()
+	iterate := func(it trie.NodeIterator) error { return v.iterate(it, false) }
+	return iterateTracked(t, fmt.Sprintf(v.params.RecoveryFormat, storageTraversal), v.params.Workers, iterate)
 }
 
 // Close implements io.Closer
@@ -146,3 +165,79 @@ func (v *Validator) Close() error {
 	groupcache.DeregisterGroup("db")
 	return nil
 }
+
+// Traverses each iterator in a separate goroutine.
+// If storage = true, also traverse storage tries for each leaf.
+func (v *Validator) iterate(it trie.NodeIterator, storage bool) error {
+	// Iterate through entire state trie. it.Next() will return false when we have
+	// either completed iteration of the entire trie or run into an error (e.g. a
+	// missing node). If we are able to iterate through the entire trie without error
+	// then the trie is complete.
+	for it.Next(true) {
+		// This block adapted from geth - core/state/iterator.go
+		// If storage is not requested, or the state trie node is an internal entry, skip
+		if !storage || !it.Leaf() {
+			continue
+		}
+		// Otherwise we've reached an account node, initiate data iteration
+		var account types.StateAccount
+		if err := rlp.Decode(bytes.NewReader(it.LeafBlob()), &account); err != nil {
+			return err
+		}
+		dataTrie, err := v.stateDatabase.OpenStorageTrie(common.BytesToHash(it.LeafKey()), account.Root)
+		if err != nil {
+			return err
+		}
+		dataIt := dataTrie.NodeIterator(nil)
+		if !bytes.Equal(account.CodeHash, emptyCodeHash) {
+			addrHash := common.BytesToHash(it.LeafKey())
+			_, err := v.stateDatabase.ContractCode(addrHash, common.BytesToHash(account.CodeHash))
+			if err != nil {
+				return fmt.Errorf("code %x: %v", account.CodeHash, err)
+			}
+		}
+		for dataIt.Next(true) {
+		}
+		if dataIt.Error() != nil {
+			return dataIt.Error()
+		}
+	}
+	return it.Error()
+}
+
+func iterateTracked(tree state.Trie, recoveryFile string, iterCount uint, fn func(trie.NodeIterator) error) error {
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	tracker := tracker.New(recoveryFile, iterCount)
+	tracker.CaptureSignal(cancelCtx)
+	halt := func() {
+		if err := tracker.HaltAndDump(); err != nil {
+			log.Errorf("failed to write recovery file: %v", err)
+		}
+	}
+
+	// attempt to restore from recovery file if it exists
+	iters, err := tracker.Restore(tree)
+	if err != nil {
+		return err
+	}
+	if iterCount < uint(len(iters)) {
+		return fmt.Errorf("recovered too many iterators: got %d, expected %d", len(iters), iterCount)
+	}
+
+	if iters == nil { // nothing restored
+		iters = nodeiter.SubtrieIterators(tree, iterCount)
+		for i, it := range iters {
+			iters[i] = tracker.Tracked(it, nil)
+		}
+	}
+
+	g, ctx := errgroup.WithContext(ctx)
+	defer halt()
+
+	for _, it := range iters {
+		func(it trie.NodeIterator) {
+			g.Go(func() error { return fn(it) })
+		}(it)
+	}
+	return g.Wait()
+}

pkg/validator_test.go

@@ -19,6 +19,8 @@ package validator_test
 import (
 	"fmt"
 	"math/big"
+	"os"
+	"path/filepath"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -30,8 +32,8 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
-	validator "github.com/vulcanize/eth-ipfs-state-validator/v4/pkg"
-	pgipfsethdb "github.com/vulcanize/ipfs-ethdb/v4/postgres"
+	validator "github.com/cerc-io/eth-ipfs-state-validator/v4/pkg"
+	pgipfsethdb "github.com/cerc-io/ipfs-ethdb/v4/postgres"
 )
 
 var (
@@ -194,15 +196,20 @@ var (
 	v   *validator.Validator
 	db  *sqlx.DB
 	err error
+	tmp string
 )
 
 var _ = Describe("PG-IPFS Validator", func() {
 	BeforeEach(func() {
 		db, err = pgipfsethdb.TestDB()
 		Expect(err).ToNot(HaveOccurred())
-		v = validator.NewPGIPFSValidator(db)
+		tmp, err = os.MkdirTemp("", "test_validator")
+		Expect(err).ToNot(HaveOccurred())
+		params := validator.Params{Workers: 4, RecoveryFormat: filepath.Join(tmp, "recover_%s")}
+		v = validator.NewPGIPFSValidator(db, params)
 	})
 	AfterEach(func() {
+		os.RemoveAll(tmp)
 		v.Close()
 	})
 	Describe("ValidateTrie", func() {
@@ -210,7 +217,7 @@ var _ = Describe("PG-IPFS Validator", func() {
 			err = validator.ResetTestDB(db)
 			Expect(err).ToNot(HaveOccurred())
 		})
-		It("Returns an error the state root node is missing", func() {
+		It("Returns an error if the state root node is missing", func() {
 			// we write code to ethdb, there should probably be an EthCode IPLD codec
 			// but there isn't, and we don't need one here since blockstore keys are mh-derived
 			loadTrie(append(missingRootStateNodes, mockCode), trieStorageNodes)