Merge pull request #45 from roysc/improve-logging

Improve logging on failed traversal

pkg/database.go

@@ -18,6 +18,8 @@ package validator
 
 import (
 	"fmt"
+	"os"
+	"strconv"
 
 	"github.com/jmoiron/sqlx"
 	"github.com/spf13/viper"
@@ -32,13 +34,6 @@ const (
 	DATABASE_PASSWORD = "DATABASE_PASSWORD"
 )
 
-// NewDB returns a new sqlx.DB from config/cli/env variables
-func NewDB() (*sqlx.DB, error) {
-	c := Config{}
-	c.Init()
-	return sqlx.Connect("postgres", c.ConnString())
-}
-
 type Config struct {
 	Hostname string
 	Name     string
@@ -47,6 +42,13 @@ type Config struct {
 	Port     int
 }
 
+// NewDB returns a new sqlx.DB from config/cli/env variables
+func NewDB() (*sqlx.DB, error) {
+	c := Config{}
+	LoadViper(&c)
+	return sqlx.Connect("postgres", c.ConnString())
+}
+
 func (c *Config) ConnString() string {
 	if len(c.User) > 0 && len(c.Password) > 0 {
 		return fmt.Sprintf("postgresql://%s:%s@%s:%d/%s?sslmode=disable",
@@ -59,7 +61,30 @@ func (c *Config) ConnString() string {
 	return fmt.Sprintf("postgresql://%s:%d/%s?sslmode=disable", c.Hostname, c.Port, c.Name)
 }
 
-func (c *Config) Init() {
+func LoadEnv(c *Config) error {
+	if val := os.Getenv(DATABASE_NAME); val != "" {
+		c.Name = val
+	}
+	if val := os.Getenv(DATABASE_HOSTNAME); val != "" {
+		c.Hostname = val
+	}
+	if val := os.Getenv(DATABASE_PORT); val != "" {
+		port, err := strconv.Atoi(val)
+		if err != nil {
+			return err
+		}
+		c.Port = port
+	}
+	if val := os.Getenv(DATABASE_USER); val != "" {
+		c.User = val
+	}
+	if val := os.Getenv(DATABASE_PASSWORD); val != "" {
+		c.Password = val
+	}
+	return nil
+}
+
+func LoadViper(c *Config) {
 	viper.BindEnv("database.name", DATABASE_NAME)
 	viper.BindEnv("database.hostname", DATABASE_HOSTNAME)
 	viper.BindEnv("database.port", DATABASE_PORT)

pkg/util.go

@@ -20,41 +20,10 @@ import (
 	"context"
 
 	"github.com/ipfs/go-blockservice"
-	"github.com/ipfs/go-cid"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
-	dshelp "github.com/ipfs/go-ipfs-ds-help"
 	"github.com/ipfs/kubo/core"
 	"github.com/ipfs/kubo/repo/fsrepo"
-	"github.com/jmoiron/sqlx"
 )
 
-// PublishRaw derives a cid from raw bytes and provided codec and multihash type, and writes it to the db tx
-func PublishRaw(tx *sqlx.Tx, codec, mh uint64, raw []byte, blockNumber uint64) (string, error) {
-	c, err := RawdataToCid(codec, raw, mh)
-	if err != nil {
-		return "", err
-	}
-	dbKey := dshelp.MultihashToDsKey(c.Hash())
-	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
-	_, err = tx.Exec(`INSERT INTO public.blocks (key, data, block_number) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING`, prefixedKey, raw, blockNumber)
-	return c.String(), err
-}
-
-// RawdataToCid takes the desired codec, multihash type, and a slice of bytes
-// and returns the proper cid of the object.
-func RawdataToCid(codec uint64, rawdata []byte, multiHash uint64) (cid.Cid, error) {
-	c, err := cid.Prefix{
-		Codec:    codec,
-		Version:  1,
-		MhType:   multiHash,
-		MhLength: -1,
-	}.Sum(rawdata)
-	if err != nil {
-		return cid.Cid{}, err
-	}
-	return c, nil
-}
-
 // InitIPFSBlockService is used to configure and return a BlockService using an ipfs repo path (e.g. ~/.ipfs)
 func InitIPFSBlockService(ipfsPath string) (blockservice.BlockService, error) {
 	r, openErr := fsrepo.Open(ipfsPath)
@@ -72,9 +41,3 @@ func InitIPFSBlockService(ipfsPath string) (blockservice.BlockService, error) {
 	}
 	return ipfsNode.Blocks, nil
 }
-
-// ResetTestDB drops all rows in the test db public.blocks table
-func ResetTestDB(db *sqlx.DB) error {
-	_, err := db.Exec("DELETE FROM public.blocks")
-	return err
-}

pkg/validator.go

@@ -166,7 +166,7 @@ func (v *Validator) Close() error {
 	return nil
 }
 
-// Traverses each iterator in a separate goroutine.
+// Traverses one iterator fully
 // 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
@@ -193,18 +193,20 @@ func (v *Validator) iterate(it trie.NodeIterator, storage bool) error {
 			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)
+				return fmt.Errorf("code %x: %w (path %x)", account.CodeHash, err, nodeiter.HexToKeyBytes(it.Path()))
 			}
 		}
 		for dataIt.Next(true) {
 		}
 		if dataIt.Error() != nil {
-			return dataIt.Error()
+			return fmt.Errorf("data iterator error (path %x): %w", nodeiter.HexToKeyBytes(dataIt.Path()), dataIt.Error())
 		}
 	}
 	return it.Error()
 }
 
+// Traverses each iterator in a separate goroutine.
+// Dumps to a recovery file on failure or interrupt.
 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)

pkg/validator_test.go

@@ -33,7 +33,6 @@ import (
 	. "github.com/onsi/gomega"
 
 	validator "github.com/cerc-io/eth-ipfs-state-validator/v4/pkg"
-	pgipfsethdb "github.com/cerc-io/ipfs-ethdb/v4/postgres"
 )
 
 var (
@@ -68,6 +67,7 @@ var (
 
 	mockCode           = []byte{1, 2, 3, 4, 5}
 	codeHash           = crypto.Keccak256Hash(mockCode)
+	codePath           = common.Hex2Bytes("6114658a74d9cc9f7acf2c5cd696c3494d7c344d78bfec3add0d91ec4e8d1c45")
 	contractAccount, _ = rlp.EncodeToBytes(&types.StateAccount{
 		Nonce:    1,
 		Balance:  big.NewInt(0),
@@ -190,6 +190,9 @@ var (
 		storageBranchRootNode,
 		slot1StorageLeafNode,
 	}
+
+	missingStateNodePath   = common.Hex2Bytes("0e")
+	missingStorageNodePath = common.Hex2Bytes("02")
 )
 
 var (
@@ -197,11 +200,21 @@ var (
 	db  *sqlx.DB
 	err error
 	tmp string
+
+	config = validator.Config{
+		Hostname: "localhost",
+		Name:     "cerc_testing",
+		User:     "vdbm",
+		Password: "password",
+		Port:     8077,
+	}
 )
 
 var _ = Describe("PG-IPFS Validator", func() {
 	BeforeEach(func() {
-		db, err = pgipfsethdb.TestDB()
+		err = validator.LoadEnv(&config)
+		Expect(err).ToNot(HaveOccurred())
+		db, err = sqlx.Connect("postgres", config.ConnString())
 		Expect(err).ToNot(HaveOccurred())
 		tmp, err = os.MkdirTemp("", "test_validator")
 		Expect(err).ToNot(HaveOccurred())
@@ -211,10 +224,11 @@ var _ = Describe("PG-IPFS Validator", func() {
 	AfterEach(func() {
 		os.RemoveAll(tmp)
 		v.Close()
+		db.Close()
 	})
 	Describe("ValidateTrie", func() {
 		AfterEach(func() {
-			err = validator.ResetTestDB(db)
+			err = ResetTestDB(db)
 			Expect(err).ToNot(HaveOccurred())
 		})
 		It("Returns an error if the state root node is missing", func() {
@@ -236,19 +250,25 @@ var _ = Describe("PG-IPFS Validator", func() {
 			err = v.ValidateTrie(stateRoot)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("missing trie node"))
+			pathSubStr := fmt.Sprintf("path %x", missingStateNodePath)
+			Expect(err.Error()).To(ContainSubstring(pathSubStr))
 		})
 		It("Returns an error if the storage trie is missing node(s)", func() {
 			loadTrie(append(trieStateNodes, mockCode), missingNodeStorageNodes)
 			err = v.ValidateTrie(stateRoot)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("missing trie node"))
+			pathSubStr := fmt.Sprintf("path %x", missingStorageNodePath)
+			Expect(err.Error()).To(ContainSubstring(pathSubStr))
 		})
 		It("Returns an error if contract code is missing", func() {
 			loadTrie(trieStateNodes, trieStorageNodes)
 			err = v.ValidateTrie(stateRoot)
 			Expect(err).To(HaveOccurred())
-			subStr := fmt.Sprintf("code %s: not found", codeHash.Hex()[2:])
+			codeSubStr := fmt.Sprintf("code %s: not found", codeHash.Hex()[2:])
-			Expect(err.Error()).To(ContainSubstring(subStr))
+			Expect(err.Error()).To(ContainSubstring(codeSubStr))
+			pathSubStr := fmt.Sprintf("path %x", codePath)
+			Expect(err.Error()).To(ContainSubstring(pathSubStr))
 		})
 		It("Returns no error if the entire state (state trie and storage tries) can be validated", func() {
 			loadTrie(append(trieStateNodes, mockCode), trieStorageNodes)
@@ -259,7 +279,7 @@ var _ = Describe("PG-IPFS Validator", func() {
 
 	Describe("ValidateStateTrie", func() {
 		AfterEach(func() {
-			err = validator.ResetTestDB(db)
+			err = ResetTestDB(db)
 			Expect(err).ToNot(HaveOccurred())
 		})
 		It("Returns an error the state root node is missing", func() {
@@ -283,7 +303,7 @@ var _ = Describe("PG-IPFS Validator", func() {
 
 	Describe("ValidateStorageTrie", func() {
 		AfterEach(func() {
-			err = validator.ResetTestDB(db)
+			err = ResetTestDB(db)
 			Expect(err).ToNot(HaveOccurred())
 		})
 		It("Returns an error the storage root node is missing", func() {
@@ -310,11 +330,11 @@ func loadTrie(stateNodes, storageNodes [][]byte) {
 	tx, err := db.Beginx()
 	Expect(err).ToNot(HaveOccurred())
 	for _, node := range stateNodes {
-		_, err := validator.PublishRaw(tx, cid.EthStateTrie, multihash.KECCAK_256, node, blockNumber)
+		_, err := PublishRaw(tx, cid.EthStateTrie, multihash.KECCAK_256, node, blockNumber)
 		Expect(err).ToNot(HaveOccurred())
 	}
 	for _, node := range storageNodes {
-		_, err := validator.PublishRaw(tx, cid.EthStorageTrie, multihash.KECCAK_256, node, blockNumber)
+		_, err := PublishRaw(tx, cid.EthStorageTrie, multihash.KECCAK_256, node, blockNumber)
 		Expect(err).ToNot(HaveOccurred())
 	}
 	err = tx.Commit()