Merge pull request #9 from deep-stack/pm-merge-perpetual

Merge changes to make validation process perpetual

cmd/state_validator.go

@@ -3,6 +3,9 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"os"
+	"os/signal"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/statediff"
 	log "github.com/sirupsen/logrus"
@@ -36,33 +39,39 @@ func stateValidator() {
 		logWithCommand.Fatalf("block height cannot be less the 1")
 	}
 	trail := viper.GetUint64("validate.trail")
+	sleepInterval := viper.GetUint("validate.sleepInterval")
 
 	chainConfigPath := viper.GetString("ethereum.chainConfig")
 	chainCfg, err := statediff.LoadConfig(chainConfigPath)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
-	srvc := validator.NewService(cfg.DB, height, trail, chainCfg)
+	service := validator.NewService(cfg.DB, height, trail, sleepInterval, chainCfg)
 
-	_, err = srvc.Start(context.Background())
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
+	wg := new(sync.WaitGroup)
+	wg.Add(1)
+	go service.Start(context.Background(), wg)
 
-	logWithCommand.Println("state validation complete")
+	shutdown := make(chan os.Signal, 1)
+	signal.Notify(shutdown, os.Interrupt)
+	<-shutdown
+	service.Stop()
+	wg.Wait()
 }
 
 func init() {
 	rootCmd.AddCommand(stateValidatorCmd)
 
 	stateValidatorCmd.PersistentFlags().String("block-height", "1", "block height to initiate state validation")
-	stateValidatorCmd.PersistentFlags().String("trail", "0", "trail of block height to validate")
+	stateValidatorCmd.PersistentFlags().String("trail", "16", "trail of block height to validate")
+	stateValidatorCmd.PersistentFlags().String("sleep-interval", "10", "sleep interval in seconds after validator has caught up to (head-trail) height")
 
 	stateValidatorCmd.PersistentFlags().String("chain-config", "", "path to chain config")
 
 	_ = viper.BindPFlag("validate.block-height", stateValidatorCmd.PersistentFlags().Lookup("block-height"))
 	_ = viper.BindPFlag("validate.trail", stateValidatorCmd.PersistentFlags().Lookup("trail"))
+	_ = viper.BindPFlag("validate.sleepInterval", stateValidatorCmd.PersistentFlags().Lookup("sleep-interval"))
 
 	_ = viper.BindPFlag("ethereum.chainConfig", stateValidatorCmd.PersistentFlags().Lookup("chain-config"))
 }

environments/example.toml

@@ -7,7 +7,8 @@
 
 [validate]
     block-height = 1
-    trail  = 10
+    trail  = 16
+    sleepInterval = 10
 
 [ethereum]
     chainConfig = "./chain.json"

pkg/validator/database.go

@@ -1,19 +1,3 @@
-// Copyright 2020 The go-ethereum Authors
-// This file is part of go-ethereum.
-//
-// go-ethereum is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// go-ethereum 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 General Public License for more details.
-//
-// You should have received a copy of the GNU General Public License
-// along with go-ethereum. If not, see <http://www.gnu.org/licenses/>.
-
 package validator
 
 import (

pkg/validator/validator.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -32,17 +33,21 @@ var (
 type service struct {
 	db              *sqlx.DB
 	blockNum, trail uint64
+	sleepInterval   uint
 	logger          *log.Logger
 	chainCfg        *params.ChainConfig
+	quitChan        chan bool
 }
 
-func NewService(db *sqlx.DB, blockNum, trailNum uint64, chainCfg *params.ChainConfig) *service {
+func NewService(db *sqlx.DB, blockNum, trailNum uint64, sleepInterval uint, chainCfg *params.ChainConfig) *service {
 	return &service{
-		db:       db,
-		blockNum: blockNum,
-		trail:    trailNum,
-		logger:   log.New(),
-		chainCfg: chainCfg,
+		db:            db,
+		blockNum:      blockNum,
+		trail:         trailNum,
+		sleepInterval: sleepInterval,
+		logger:        log.New(),
+		chainCfg:      chainCfg,
+		quitChan:      make(chan bool),
 	}
 }
 
@@ -76,51 +81,86 @@ func NewEthBackend(db *sqlx.DB, c *ipldEth.Config) (*ipldEth.Backend, error) {
 }
 
 // Start is used to begin the service
-func (s *service) Start(ctx context.Context) (uint64, error) {
-	api, err := ethAPI(ctx, s.db, s.chainCfg)
+func (s *service) Start(ctx context.Context, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	api, err := EthAPI(ctx, s.db, s.chainCfg)
 	if err != nil {
-		return 0, err
+		s.logger.Fatal(err)
+		return
 	}
 
 	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
+	for {
+		select {
+		case <-s.quitChan:
+			s.logger.Infof("last validated block %v", idxBlockNum-1)
+			s.logger.Info("stopping ipld-eth-db-validator process")
+			return
+		default:
+			idxBlockNum, err = s.Validate(ctx, api, idxBlockNum)
+			if err != nil {
+				s.logger.Infof("last validated block %v", idxBlockNum-1)
+				s.logger.Fatal(err)
+				return
+			}
 		}
+	}
+}
 
-		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")
-		}
+// Stop is used to gracefully stop the service
+func (s *service) Stop() {
+	close(s.quitChan)
+}
 
-		s.logger.Infof("state root verified for block %d", idxBlockNum)
+func (s *service) Validate(ctx context.Context, api *ipldEth.PublicEthAPI, idxBlockNum uint64) (uint64, error) {
+	headBlockNum, err := fetchHeadBlockNumber(ctx, api)
+	if err != nil {
+		return idxBlockNum, err
+	}
 
-		headBlock, err = api.B.BlockByNumber(ctx, rpc.LatestBlockNumber)
+	// Check if it block at height idxBlockNum can be validated
+	if idxBlockNum <= headBlockNum-s.trail {
+		err = ValidateBlock(ctx, api, idxBlockNum)
 		if err != nil {
+			s.logger.Errorf("failed to verify state root at block %d", idxBlockNum)
 			return idxBlockNum, err
 		}
 
-		headBlockNum = headBlock.NumberU64()
+		s.logger.Infof("state root verified for block %d", idxBlockNum)
 		idxBlockNum++
+	} else {
+		// Sleep / wait for head to move ahead
+		time.Sleep(time.Second * time.Duration(s.sleepInterval))
 	}
 
-	s.logger.Infof("last validated block %v", idxBlockNum-1)
 	return idxBlockNum, nil
 }
 
-func ethAPI(ctx context.Context, db *sqlx.DB, chainCfg *params.ChainConfig) (*ipldEth.PublicEthAPI, error) {
+// ValidateBlock validates block at the given height
+func ValidateBlock(ctx context.Context, api *ipldEth.PublicEthAPI, blockNumber uint64) error {
+	blockToBeValidated, err := api.B.BlockByNumber(ctx, rpc.BlockNumber(blockNumber))
+	if err != nil {
+		return err
+	}
+
+	stateDB, err := applyTransaction(blockToBeValidated, api.B)
+	if err != nil {
+		return err
+	}
+
+	blockStateRoot := blockToBeValidated.Header().Root.String()
+
+	dbStateRoot := stateDB.IntermediateRoot(true).String()
+	if blockStateRoot != dbStateRoot {
+		return fmt.Errorf("state roots do not match at block %d", blockNumber)
+	}
+
+	return nil
+}
+
+func EthAPI(ctx context.Context, db *sqlx.DB, chainCfg *params.ChainConfig) (*ipldEth.PublicEthAPI, error) {
 	// TODO: decide network for custom chainConfig.
 	backend, err := NewEthBackend(db, &ipldEth.Config{
 		ChainConfig: chainCfg,
@@ -147,6 +187,16 @@ func ethAPI(ctx context.Context, db *sqlx.DB, chainCfg *params.ChainConfig) (*ip
 	return ipldEth.NewPublicEthAPI(backend, nil, false, false, false)
 }
 
+// fetchHeadBlockNumber gets the latest block number from the db
+func fetchHeadBlockNumber(ctx context.Context, api *ipldEth.PublicEthAPI) (uint64, error) {
+	headBlock, err := api.B.BlockByNumber(ctx, rpc.LatestBlockNumber)
+	if err != nil {
+		return 0, err
+	}
+
+	return headBlock.NumberU64(), nil
+}
+
 // 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) {

test/integration_test.go

@@ -13,7 +13,10 @@ import (
 	integration "github.com/vulcanize/ipld-eth-server/v4/test"
 )
 
-const trail = 0
+const (
+	trail                  = 0
+	validatorSleepInterval = uint(5)
+)
 
 var _ = Describe("Integration test", func() {
 	ctx := context.Background()
@@ -36,9 +39,8 @@ var _ = Describe("Integration test", func() {
 			Expect(contractErr).ToNot(HaveOccurred())
 
 			db := shared.SetupDB()
-			srvc := validator.NewService(db, uint64(contract.BlockNumber), trail, validator.IntegrationTestChainConfig)
-			_, err := srvc.Start(ctx)
-			Expect(err).ToNot(HaveOccurred())
+			srvc := validator.NewService(db, uint64(contract.BlockNumber), trail, validatorSleepInterval, validator.IntegrationTestChainConfig)
+			srvc.Start(ctx, nil)
 		})
 	})
 })

validator_test/validator_test.go

@@ -114,10 +114,13 @@ var _ = Describe("eth state reading tests", func() {
 
 	Describe("state_validation", func() {
 		It("Validator", func() {
-			srvc := validator.NewService(db, blockHeight, trail, validator.TestChainConfig)
-
-			_, err := srvc.Start(context.Background())
+			api, err := validator.EthAPI(context.Background(), db, validator.TestChainConfig)
 			Expect(err).ToNot(HaveOccurred())
+
+			for i := uint64(blockHeight); i <= chainLength-trail; i++ {
+				err = validator.ValidateBlock(context.Background(), api, i)
+				Expect(err).ToNot(HaveOccurred())
+			}
 		})
 	})
 })