Integrate state validator into server

cmd/serve.go

@@ -23,6 +23,7 @@ import (
 	"os/signal"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/mailgun/groupcache/v2"
@@ -92,6 +93,13 @@ func serve() {
 		logWithCommand.Fatal(err)
 	}
 
+	if serverConfig.StateValidationEnabled {
+		go startStateTrieValidator(server, serverConfig.StateValidationEveryNthBlock)
+		logWithCommand.Info("state validator enabled")
+	} else {
+		logWithCommand.Info("state validator disabled")
+	}
+
 	shutdown := make(chan os.Signal)
 	signal.Notify(shutdown, os.Interrupt)
 	<-shutdown
@@ -236,6 +244,46 @@ func startGroupCacheService(settings *s.Config) error {
 	return nil
 }
 
+func startStateTrieValidator(server s.Server, validateEveryNthBlock uint64) {
+	var lastBlockNumber uint64 = 0
+	backend := server.Backend()
+
+	for {
+		time.Sleep(10 * time.Second)
+
+		block, err := backend.CurrentBlock()
+		if err != nil {
+			log.Errorln("Error fetching current block for state trie validator")
+			continue
+		}
+
+		stateRoot := block.Root()
+		blockNumber := block.NumberU64()
+		blockHash := block.Hash()
+
+		if (blockNumber > lastBlockNumber) && (blockNumber%validateEveryNthBlock == 0) {
+			// The validate trie call will take a long time on mainnet, e.g. a few hours.
+			err = backend.ValidateTrie(stateRoot)
+			if err != nil {
+				// Log an error and exit.
+				log.Fatalf("Error validating state trie for block %s (%d), state root %s",
+					blockHash,
+					blockNumber,
+					stateRoot,
+				)
+			} else {
+				log.Infoln("Successfully validated state trie for block %s (%d), state root %s",
+					blockHash,
+					blockNumber,
+					stateRoot,
+				)
+			}
+
+			lastBlockNumber = blockNumber
+		}
+	}
+}
+
 func parseRpcAddresses(value string) ([]*rpc.Client, error) {
 	rpcAddresses := strings.Split(value, ",")
 	rpcClients := make([]*rpc.Client, 0, len(rpcAddresses))
@@ -304,6 +352,10 @@ func init() {
 	serveCmd.PersistentFlags().Int("gcache-statedb-cache-expiry", 60, "state DB cache expiry time in mins")
 	serveCmd.PersistentFlags().Int("gcache-statedb-log-stats-interval", 60, "state DB cache stats log interval in secs")
 
+	// state validator flags
+	serveCmd.PersistentFlags().Bool("validator-enabled", false, "turn on the state validator")
+	serveCmd.PersistentFlags().Uint("validator-every-nth-block", 1500, "only validate every Nth block")
+
 	// and their bindings
 	// database
 	viper.BindPFlag("database.name", serveCmd.PersistentFlags().Lookup("database-name"))
@@ -353,4 +405,8 @@ func init() {
 	viper.BindPFlag("groupcache.statedb.cacheSizeInMB", serveCmd.PersistentFlags().Lookup("gcache-statedb-cache-size"))
 	viper.BindPFlag("groupcache.statedb.cacheExpiryInMins", serveCmd.PersistentFlags().Lookup("gcache-statedb-cache-expiry"))
 	viper.BindPFlag("groupcache.statedb.logStatsIntervalInSecs", serveCmd.PersistentFlags().Lookup("gcache-statedb-log-stats-interval"))
+
+	// state validator flags
+	viper.BindPFlag("validator.enabled", serveCmd.PersistentFlags().Lookup("validator-enabled"))
+	viper.BindPFlag("validator.everyNthBlock", serveCmd.PersistentFlags().Lookup("validator-every-nth-block"))
 }

pkg/eth/backend.go

@@ -120,7 +120,7 @@ func NewEthBackend(db *postgres.DB, c *Config) (*Backend, error) {
 	r := NewCIDRetriever(db)
 	ethDB := ipfsethdb.NewDatabase(db.DB, ipfsethdb.CacheConfig{
 		Name:           StateDBGroupCacheName,
-		Size:           gcc.StateDB.CacheSizeInMB,
+		Size:           gcc.StateDB.CacheSizeInMB * 1024 * 1024,
 		ExpiryDuration: time.Minute * time.Duration(gcc.StateDB.CacheExpiryInMins),
 	})
 
@@ -809,6 +809,26 @@ func (b *Backend) GetHeader(hash common.Hash, height uint64) *types.Header {
 	return header
 }
 
+// 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 (b *Backend) ValidateTrie(stateRoot common.Hash) error {
+	// Generate the state.NodeIterator for this root
+	stateDB, err := state.New(stateRoot, b.StateDatabase, nil)
+
+	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
+}
+
 // RPCGasCap returns the configured gas cap for the rpc server
 func (b *Backend) RPCGasCap() *big.Int {
 	return b.Config.RPCGasCap

pkg/serve/config.go

@@ -48,6 +48,9 @@ const (
 	ethRPCGasCap         = "ETH_RPC_GAS_CAP"
 	ethChainConfig       = "ETH_CHAIN_CONFIG"
 	ethSupportsStatediff = "ETH_SUPPORTS_STATEDIFF"
+
+	VALIDATOR_ENABLED         = "VALIDATOR_ENABLED"
+	VALIDATOR_EVERY_NTH_BLOCK = "VALIDATOR_EVERY_NTH_BLOCK"
 )
 
 // Config struct
@@ -83,6 +86,9 @@ type Config struct {
 
 	// Cache configuration.
 	GroupCache *ethServerShared.GroupCacheConfig
+
+	StateValidationEnabled       bool
+	StateValidationEveryNthBlock uint64
 }
 
 // NewConfig is used to initialize a watcher config from a .toml file
@@ -212,6 +218,8 @@ func NewConfig() (*Config, error) {
 
 	c.loadGroupCacheConfig()
 
+	c.loadValidatorConfig()
+
 	return c, err
 }
 
@@ -266,3 +274,11 @@ func (c *Config) loadGroupCacheConfig() {
 
 	c.GroupCache = &gcc
 }
+
+func (c *Config) loadValidatorConfig() {
+	viper.BindEnv("validator.enabled", VALIDATOR_ENABLED)
+	viper.BindEnv("validator.everyNthBlock", VALIDATOR_EVERY_NTH_BLOCK)
+
+	c.StateValidationEnabled = viper.GetBool("validator.enabled")
+	c.StateValidationEveryNthBlock = viper.GetUint64("validator.everyNthBlock")
+}