Feature/update go geth sharding (#13)

* Upgrade Geth, Go and add CICD

* update ipld-eth-server version

* Track validation progress on a channel

* Add integration tests

* Setup validator config and update instructions to run tests locally

* Update readme and tests

* Update test to use v4 Infrastructure

* Inlcude the env file

* Fix config file write

* Update DB configuration

Co-authored-by: prathamesh0 <prathamesh.musale0@gmail.com>
Co-authored-by: Ashwin Phatak <ashwinpphatak@gmail.com>

pkg/validator/config.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/params"
+	"github.com/ethereum/go-ethereum/statediff"
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/jmoiron/sqlx"
 	"github.com/spf13/viper"
@@ -59,11 +60,35 @@ var TestChainConfig = &params.ChainConfig{
 type Config struct {
 	dbConfig postgres.Config
 	DB       *sqlx.DB
+
+	ChainCfg *params.ChainConfig
+
+	BlockNum, Trail uint64
+	SleepInterval   uint
 }
 
 func NewConfig() (*Config, error) {
 	cfg := new(Config)
-	return cfg, cfg.setupDB()
+	err := cfg.setupDB()
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.BlockNum = viper.GetUint64("validate.block-height")
+	if cfg.BlockNum < 1 {
+		return nil, fmt.Errorf("block height cannot be less the 1")
+	}
+
+	cfg.Trail = viper.GetUint64("validate.trail")
+	cfg.SleepInterval = viper.GetUint("validate.sleepInterval")
+
+	chainConfigPath := viper.GetString("ethereum.chainConfig")
+	cfg.ChainCfg, err = statediff.LoadConfig(chainConfigPath)
+	if err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
 }
 
 func (c *Config) setupDB() error {

pkg/validator/database.go

@@ -1,9 +1,13 @@
 package validator
 
 import (
+	"errors"
+
 	"github.com/ethereum/go-ethereum/ethdb"
 )
 
+var errNotSupported = errors.New("this operation is not supported")
+
 type database struct {
 	ethDB ethdb.Database
 }
@@ -105,7 +109,7 @@ func (d *database) NewBatchWithSize(size int) ethdb.Batch {
 	return d.ethDB.NewBatchWithSize(size)
 }
 
-func (d *database) ReadAncients(fn func(ethdb.AncientReader) error) (err error) {
+func (d *database) ReadAncients(fn func(ethdb.AncientReaderOp) error) (err error) {
 	return d.ethDB.ReadAncients(fn)
 }
 
@@ -117,3 +121,8 @@ func (d *database) Close() error {
 func (d *database) NewSnapshot() (ethdb.Snapshot, error) {
 	return d.NewSnapshot()
 }
+
+// NewSnapshot creates a database snapshot based on the current state.
+func (d *database) AncientDatadir() (string, error) {
+	return "", errNotSupported
+}

pkg/validator/validator.go

@@ -41,17 +41,19 @@ type service struct {
 	logger          *log.Logger
 	chainCfg        *params.ChainConfig
 	quitChan        chan bool
+	progressChan    chan uint64
 }
 
-func NewService(db *sqlx.DB, blockNum, trailNum uint64, sleepInterval uint, chainCfg *params.ChainConfig) *service {
+func NewService(cfg *Config, progressChan chan uint64) *service {
 	return &service{
-		db:            db,
-		blockNum:      blockNum,
-		trail:         trailNum,
-		sleepInterval: sleepInterval,
+		db:            cfg.DB,
+		blockNum:      cfg.BlockNum,
+		trail:         cfg.Trail,
+		sleepInterval: cfg.SleepInterval,
 		logger:        log.New(),
-		chainCfg:      chainCfg,
+		chainCfg:      cfg.ChainCfg,
 		quitChan:      make(chan bool),
+		progressChan:  progressChan,
 	}
 }
 
@@ -100,7 +102,9 @@ func (s *service) Start(ctx context.Context, wg *sync.WaitGroup) {
 		select {
 		case <-s.quitChan:
 			s.logger.Infof("last validated block %v", idxBlockNum-1)
-			s.logger.Info("stopping ipld-eth-db-validator process")
+			if s.progressChan != nil {
+				close(s.progressChan)
+			}
 			return
 		default:
 			idxBlockNum, err = s.Validate(ctx, api, idxBlockNum)
@@ -115,6 +119,7 @@ func (s *service) Start(ctx context.Context, wg *sync.WaitGroup) {
 
 // Stop is used to gracefully stop the service
 func (s *service) Stop() {
+	s.logger.Info("stopping ipld-eth-db-validator process")
 	close(s.quitChan)
 }
 
@@ -133,15 +138,10 @@ func (s *service) Validate(ctx context.Context, api *ipldEth.PublicEthAPI, idxBl
 		}
 
 		s.logger.Infof("state root verified for block %d", idxBlockNum)
-
-		err = ValidateReferentialIntegrity(s.db, idxBlockNum)
-		if err != nil {
-			s.logger.Errorf("failed to verify referential integrity at block %d", idxBlockNum)
-			return idxBlockNum, err
+		if s.progressChan != nil {
+			s.progressChan <- idxBlockNum
 		}
 
-		s.logger.Infof("referential integrity verified for block %d", idxBlockNum)
-
 		idxBlockNum++
 	} else {
 		// Sleep / wait for head to move ahead