Implement ethereum state validator.

cmd/root.go

@@ -0,0 +1,88 @@
+package cmd
+
+import (
+	"os"
+	"strings"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+var (
+	cfgFile        string
+	subCommand     string
+	logWithCommand log.Entry
+)
+
+// rootCmd represents the base command when called without any subcommands
+var rootCmd = &cobra.Command{
+	Use:              "ipld-eth-db-validator",
+	Short:            "Validates each block state stored for state-diff service.",
+	Long:             `Validates each block state stored for state-diff service.`,
+	PersistentPreRun: initFunc,
+}
+
+func Execute() {
+	log.Info("----- Starting state validator -----")
+	if err := rootCmd.Execute(); err != nil {
+		os.Exit(1)
+	}
+}
+
+func initFunc(cmd *cobra.Command, args []string) {
+	logfile := viper.GetString("logfile")
+	if logfile != "" {
+		file, err := os.OpenFile(logfile,
+			os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
+		if err == nil {
+			log.Infof("Directing output to %s", logfile)
+			log.SetOutput(file)
+		} else {
+			log.SetOutput(os.Stdout)
+			log.Info("Failed to log to file, using default stdout")
+		}
+	} else {
+		log.SetOutput(os.Stdout)
+	}
+
+	if err := logLevel(); err != nil {
+		log.Fatal("Could not set log level: ", err)
+	}
+}
+
+func init() {
+	cobra.OnInitialize(initConfig)
+	viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
+	viper.AutomaticEnv()
+
+	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file location")
+	rootCmd.PersistentFlags().String("logfile", "", "file path for logging")
+	rootCmd.PersistentFlags().String("database-name", "vulcanize_public", "database name")
+	rootCmd.PersistentFlags().Int("database-port", 5432, "database port")
+	rootCmd.PersistentFlags().String("database-hostname", "localhost", "database hostname")
+	rootCmd.PersistentFlags().String("database-user", "", "database user")
+	rootCmd.PersistentFlags().String("database-password", "", "database password")
+	rootCmd.PersistentFlags().String("log-level", log.InfoLevel.String(), "Log level (trace, debug, info, warn, error, fatal, panic")
+
+	_ = viper.BindPFlag("logfile", rootCmd.PersistentFlags().Lookup("logfile"))
+	_ = viper.BindPFlag("database.name", rootCmd.PersistentFlags().Lookup("database-name"))
+	_ = viper.BindPFlag("database.port", rootCmd.PersistentFlags().Lookup("database-port"))
+	_ = viper.BindPFlag("database.hostname", rootCmd.PersistentFlags().Lookup("database-hostname"))
+	_ = viper.BindPFlag("database.user", rootCmd.PersistentFlags().Lookup("database-user"))
+	_ = viper.BindPFlag("database.password", rootCmd.PersistentFlags().Lookup("database-password"))
+	_ = viper.BindPFlag("log.level", rootCmd.PersistentFlags().Lookup("log-level"))
+}
+
+func logLevel() error {
+	lvl, err := log.ParseLevel(viper.GetString("log.level"))
+	if err != nil {
+		return err
+	}
+	log.SetLevel(lvl)
+	if lvl > log.InfoLevel {
+		log.SetReportCaller(true)
+	}
+	log.Info("Log level set to ", lvl.String())
+	return nil
+}

environments/example.toml

@@ -0,0 +1,10 @@
+[database]
+    name     = "vulcanize_public"
+    hostname = "localhost"
+    port     = 5432
+    password = "password"
+    user     = "vdbm"
+
+[validate]
+    block-height = 1
+    trail  = 3100

go.mod

@@ -1,3 +1,18 @@
 module github.com/Vulcanize/ipld-eth-db-validator
 
 go 1.17
+
+require (
+	github.com/ethereum/go-ethereum v1.10.14
+	github.com/jmoiron/sqlx v1.2.0
+	github.com/sirupsen/logrus v1.7.0
+	github.com/spf13/cobra v1.3.0
+	github.com/spf13/viper v1.10.0
+	github.com/vulcanize/ipfs-ethdb v0.0.6
+	github.com/vulcanize/ipld-eth-server v0.3.9
+)
+
+replace (
+	github.com/ethereum/go-ethereum v1.10.14 => github.com/vulcanize/go-ethereum v1.10.14-statediff-0.0.29
+	github.com/vulcanize/ipld-eth-server v0.3.9 => github.com/vulcanize/ipld-eth-server v0.0.0-20220117121622-5ea788912bd3
+)

main.go

@@ -0,0 +1,7 @@
+package main
+
+import "github.com/Vulcanize/ipld-eth-db-validator/cmd"
+
+func main() {
+	cmd.Execute()
+}

pkg/validator/config.go

@@ -0,0 +1,52 @@
+package validator
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/statediff/indexer/node"
+	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
+	"github.com/spf13/viper"
+)
+
+type Config struct {
+	dbParams postgres.ConnectionParams
+	dbConfig postgres.ConnectionConfig
+	DB       *postgres.DB
+}
+
+func NewConfig() (*Config, error) {
+	cfg := new(Config)
+	return cfg, cfg.setupDB()
+}
+
+func (c *Config) setupDB() error {
+	_ = viper.BindEnv("database.name", postgres.DATABASE_NAME)
+	_ = viper.BindEnv("database.hostname", postgres.DATABASE_HOSTNAME)
+	_ = viper.BindEnv("database.port", postgres.DATABASE_PORT)
+	_ = viper.BindEnv("database.user", postgres.DATABASE_USER)
+	_ = viper.BindEnv("database.password", postgres.DATABASE_PASSWORD)
+	_ = viper.BindEnv("database.maxIdle", postgres.DATABASE_MAX_IDLE_CONNECTIONS)
+	_ = viper.BindEnv("database.maxOpen", postgres.DATABASE_MAX_OPEN_CONNECTIONS)
+	_ = viper.BindEnv("database.maxLifetime", postgres.DATABASE_MAX_CONN_LIFETIME)
+
+	// DB params
+	c.dbParams.Name = viper.GetString("database.name")
+	c.dbParams.Hostname = viper.GetString("database.hostname")
+	c.dbParams.Port = viper.GetInt("database.port")
+	c.dbParams.User = viper.GetString("database.user")
+	c.dbParams.Password = viper.GetString("database.password")
+
+	// DB Config
+	c.dbConfig.MaxIdle = viper.GetInt("database.maxIdle")
+	c.dbConfig.MaxOpen = viper.GetInt("database.maxOpen")
+	c.dbConfig.MaxLifetime = viper.GetInt("database.maxLifetime")
+
+	// Create DB
+	db, err := NewDB(postgres.DbConnectionString(c.dbParams), postgres.ConnectionConfig{}, node.Info{})
+	if err != nil {
+		return fmt.Errorf("failed to create config: %w", err)
+	}
+
+	c.DB = db
+	return nil
+}

pkg/validator/validator.go

@@ -0,0 +1,191 @@
+package validator
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/core/vm"
+	"github.com/ethereum/go-ethereum/params"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/statediff/indexer/node"
+	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
+	"github.com/jmoiron/sqlx"
+	log "github.com/sirupsen/logrus"
+	ipfsethdb "github.com/vulcanize/ipfs-ethdb/postgres"
+	ipldEth "github.com/vulcanize/ipld-eth-server/pkg/eth"
+	ethServerShared "github.com/vulcanize/ipld-eth-server/pkg/shared"
+)
+
+type service struct {
+	db              *postgres.DB
+	blockNum, trail uint64
+	logger          *log.Logger
+}
+
+func NewService(db *postgres.DB, blockNum, trailNum uint64) *service {
+	return &service{
+		db:       db,
+		blockNum: blockNum,
+		trail:    trailNum,
+		logger:   log.New(),
+	}
+}
+
+func NewEthBackend(db *postgres.DB, c *ipldEth.Config) (*ipldEth.Backend, error) {
+	gcc := c.GroupCacheConfig
+
+	groupName := gcc.StateDB.Name
+	if groupName == "" {
+		groupName = ipldEth.StateDBGroupCacheName
+	}
+
+	r := ipldEth.NewCIDRetriever(db)
+	ethDB := ipfsethdb.NewDatabase(db.DB, ipfsethdb.CacheConfig{
+		Name:           groupName,
+		Size:           gcc.StateDB.CacheSizeInMB * 1024 * 1024,
+		ExpiryDuration: time.Minute * time.Duration(gcc.StateDB.CacheExpiryInMins),
+	})
+
+	customEthDB := NewDatabase(ethDB)
+
+	return &ipldEth.Backend{
+		DB:            db,
+		Retriever:     r,
+		Fetcher:       ipldEth.NewIPLDFetcher(db),
+		IPLDRetriever: ipldEth.NewIPLDRetriever(db),
+		EthDB:         customEthDB,
+		StateDatabase: state.NewDatabase(customEthDB),
+		Config:        c,
+	}, nil
+}
+
+func NewDB(connectString string, config postgres.ConnectionConfig, node node.Info) (*postgres.DB, error) {
+	db, connectErr := sqlx.Connect("postgres", connectString)
+	if connectErr != nil {
+		return &postgres.DB{}, postgres.ErrDBConnectionFailed(connectErr)
+	}
+	if config.MaxOpen > 0 {
+		db.SetMaxOpenConns(config.MaxOpen)
+	}
+	if config.MaxIdle > 0 {
+		db.SetMaxIdleConns(config.MaxIdle)
+	}
+	if config.MaxLifetime > 0 {
+		lifetime := time.Duration(config.MaxLifetime) * time.Second
+		db.SetConnMaxLifetime(lifetime)
+	}
+	pg := postgres.DB{DB: db, Node: node}
+	return &pg, nil
+}
+
+// Start is used to begin the service
+func (s *service) Start(ctx context.Context) (uint64, error) {
+	api, err := ethAPI(s.db)
+	if err != nil {
+		return 0, err
+	}
+
+	idxBlockNum := s.blockNum
+	headBlock, _ := api.B.BlockByNumber(ctx, rpc.LatestBlockNumber)
+	headBlockNum := headBlock.NumberU64()
+
+	for headBlockNum-s.trail >= idxBlockNum {
+		validateBlock, err := api.B.BlockByNumber(ctx, rpc.BlockNumber(idxBlockNum))
+		if err != nil {
+			return idxBlockNum, err
+		}
+
+		stateDB, err := applyTransaction(validateBlock, api.B)
+		if err != nil {
+			return idxBlockNum, err
+		}
+
+		blockStateRoot := validateBlock.Header().Root.String()
+		dbStateRoot := stateDB.IntermediateRoot(true).String()
+		if blockStateRoot != dbStateRoot {
+			s.logger.Errorf("failed to verify state root at block %d", idxBlockNum)
+			return idxBlockNum, fmt.Errorf("failed to verify state root at block")
+		}
+
+		s.logger.Infof("state root verified for block= %d", idxBlockNum)
+
+		// again fetch head block
+		headBlock, err = api.B.BlockByNumber(ctx, rpc.LatestBlockNumber)
+		if err != nil {
+			return idxBlockNum, err
+		}
+
+		headBlockNum = headBlock.NumberU64()
+		idxBlockNum++
+	}
+
+	s.logger.Infof("last validated block %v", idxBlockNum)
+
+	return idxBlockNum, nil
+}
+
+func ethAPI(db *postgres.DB) (*ipldEth.PublicEthAPI, error) {
+	// TODO: decide network for chainConfig.
+	backend, err := NewEthBackend(db, &ipldEth.Config{
+		ChainConfig: params.RinkebyChainConfig,
+		GroupCacheConfig: &ethServerShared.GroupCacheConfig{
+			StateDB: ethServerShared.GroupConfig{
+				Name: "vulcanize_validator",
+			},
+		},
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return ipldEth.NewPublicEthAPI(backend, nil, false, false, false)
+}
+
+// applyTransaction attempts to apply a transaction to the given state database
+// and uses the input parameters for its environment. It returns the stateDB of parent with applied transa
+func applyTransaction(block *types.Block, backend *ipldEth.Backend) (*state.StateDB, error) {
+	if block.NumberU64() == 0 {
+		return nil, errors.New("no transaction in genesis")
+	}
+
+	// Create the parent state database
+	parentHash := block.ParentHash()
+	parentRPCBlockHash := rpc.BlockNumberOrHash{
+		BlockHash:        &parentHash,
+		RequireCanonical: false,
+	}
+
+	parent, _ := backend.BlockByNumberOrHash(context.Background(), parentRPCBlockHash)
+	if parent == nil {
+		return nil, fmt.Errorf("parent %#x not found", block.ParentHash())
+	}
+
+	stateDB, _, err := backend.StateAndHeaderByNumberOrHash(context.Background(), parentRPCBlockHash)
+	if err != nil {
+		return nil, err
+	}
+
+	signer := types.MakeSigner(backend.Config.ChainConfig, block.Number())
+
+	for idx, tx := range block.Transactions() {
+		// Assemble the transaction call message and return if the requested offset
+		msg, _ := tx.AsMessage(signer, block.BaseFee())
+		txContext := core.NewEVMTxContext(msg)
+		ctx := core.NewEVMBlockContext(block.Header(), backend, nil)
+
+		// Not yet the searched for transaction, execute on top of the current state
+		newEVM := vm.NewEVM(ctx, txContext, stateDB, backend.Config.ChainConfig, vm.Config{})
+
+		stateDB.Prepare(tx.Hash(), idx)
+		if _, err := core.ApplyMessage(newEVM, msg, new(core.GasPool).AddGas(block.GasLimit())); err != nil {
+			return nil, fmt.Errorf("transaction %#x failed: %w", tx.Hash(), err)
+		}
+	}
+	return stateDB, nil
+}