Add github ref name.

bump go-ethereum version

.github/workflows/publish.yaml

@@ -7,19 +7,19 @@ jobs:
     name: Push Docker image to Docker Hub
     runs-on: ubuntu-latest
     steps:
+      - uses: actions/checkout@v2
       - name: Get the version
-        id: vars
         run: |
-          echo ::set-output name=sha::$(echo ${GITHUB_SHA:0:7})
           echo ::set-output name=tag::$(echo ${GITHUB_REF#refs/tags/})
-      - name: Docker Login to Github Registry
+          echo "::set-output name=branch::$(echo ${github.event.release.target_commitish})"
+        id: vars
+      - name: Docker Login
         run: echo ${{ secrets.GITHUB_TOKEN }} | docker login https://docker.pkg.github.com -u vulcanize --password-stdin
-      - name: Docker Pull
-        run: docker pull docker.pkg.github.com/vulcanize/eth-statediff-service/eth-statediff-service:${{steps.vars.outputs.sha}}
-      - name: Docker Login to Docker Registry
-        run: echo ${{ secrets.VULCANIZEJENKINS_PAT }} | docker login -u vulcanizejenkins --password-stdin
-      - name: Tag docker image
-        run: docker tag docker.pkg.github.com/vulcanize/eth-statediff-service/eth-statediff-service:${{steps.vars.outputs.sha}} vulcanize/eth-statediff-service:${{steps.vars.outputs.tag}}
-      - name: Docker Push to Docker Hub
-        run: docker push vulcanize/eth-statediff-service:${{steps.vars.outputs.tag}}
-
+      - name: Build and push
+        uses: docker/build-push-action@v2
+        with:
+          context: .
+          push: true
+          tags: vulcanize/eth-statediff-service:${{ steps.vars.outputs.tag }}
+          build-args: |
+            BRANCH=${{ steps.vars.outputs.branch }}

Dockerfile

@@ -15,7 +15,7 @@ RUN GO111MODULE=on GCO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -ldfl
 FROM alpine
 
 ARG USER="vdm"
-ARG CONFIG_FILE="./environments/example.toml"
+ARG CONFIG_FILE="./environments/config.toml"
 ARG EXPOSE_PORT=8545
 
 RUN adduser -Du 5000 $USER adm

cmd/env.go

@@ -33,17 +33,34 @@ const (
 	TRIE_CACHE_SIZE_MB = "TRIE_CACHE_SIZE_MB"
 	LVLDB_PATH         = "LVLDB_PATH"
 	LVLDB_ANCIENT      = "LVLDB_ANCIENT"
-	STATEDIFF_WORKERS  = "STATEDIFF_WORKERS"
-	WRITE_SERVER       = "WRITE_SERVER"
+
+	STATEDIFF_PRERUN            = "STATEDIFF_PRERUN"
+	STATEDIFF_TRIE_WORKERS      = "STATEDIFF_TRIE_WORKERS"
+	STATEDIFF_SERVICE_WORKERS   = "STATEDIFF_SERVICE_WORKERS"
+	STATEDIFF_WORKER_QUEUE_SIZE = "STATEDIFF_WORKER_QUEUE_SIZE"
+
+	SERVICE_IPC_PATH  = "SERVICE_IPC_PATH"
+	SERVICE_HTTP_PATH = "SERVICE_HTTP_PATH"
 
 	PROM_METRICS   = "PROM_METRICS"
 	PROM_HTTP      = "PROM_HTTP"
 	PROM_HTTP_ADDR = "PROM_HTTP_ADDR"
 	PROM_HTTP_PORT = "PROM_HTTP_PORT"
+	PROM_DB_STATS  = "PROM_DB_STATS"
+
+	PRERUN_ONLY        = "PRERUN_ONLY"
+	PRERUN_RANGE_START = "PRERUN_RANGE_START"
+	PRERUN_RANGE_STOP  = "PRERUN_RANGE_STOP"
+
+	LOG_LEVEL     = "LOG_LEVEL"
+	LOG_FILE_PATH = "LOG_FILE_PATH"
 )
 
 // Bind env vars for eth node and DB configuration
 func init() {
+	viper.BindEnv("server.ipcPath", SERVICE_IPC_PATH)
+	viper.BindEnv("server.httpPath", SERVICE_HTTP_PATH)
+
 	viper.BindEnv("ethereum.nodeID", ETH_NODE_ID)
 	viper.BindEnv("ethereum.clientName", ETH_CLIENT_NAME)
 	viper.BindEnv("ethereum.genesisBlock", ETH_GENESIS_BLOCK)
@@ -69,6 +86,17 @@ func init() {
 	viper.BindEnv("prom.http", PROM_HTTP)
 	viper.BindEnv("prom.httpAddr", PROM_HTTP_ADDR)
 	viper.BindEnv("prom.httpPort", PROM_HTTP_PORT)
+	viper.BindEnv("prom.dbStats", PROM_DB_STATS)
 
-	viper.BindEnv("statediff.workers", STATEDIFF_WORKERS)
+	viper.BindEnv("statediff.serviceWorkers", STATEDIFF_SERVICE_WORKERS)
+	viper.BindEnv("statediff.trieWorkers", STATEDIFF_TRIE_WORKERS)
+	viper.BindEnv("statediff.workerQueueSize", STATEDIFF_WORKER_QUEUE_SIZE)
+
+	viper.BindEnv("statediff.prerun", STATEDIFF_PRERUN)
+	viper.BindEnv("prerun.only", PRERUN_ONLY)
+	viper.BindEnv("prerun.start", PRERUN_RANGE_START)
+	viper.BindEnv("prerun.stop", PRERUN_RANGE_STOP)
+
+	viper.BindEnv("log.level", LOG_LEVEL)
+	viper.BindEnv("log.file", LOG_FILE_PATH)
 }

cmd/root.go

@@ -18,8 +18,10 @@ package cmd
 
 import (
 	"fmt"
+	"math/rand"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/ethereum/go-ethereum/statediff/indexer/node"
 	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
@@ -68,6 +70,7 @@ func initFuncs(cmd *cobra.Command, args []string) {
 	}
 
 	if viper.GetBool("prom.metrics") {
+		log.Info("initializing prometheus metrics")
 		prom.Init()
 	}
 
@@ -77,12 +80,12 @@ func initFuncs(cmd *cobra.Command, args []string) {
 			viper.GetString("prom.httpAddr"),
 			viper.GetString("prom.httpPort"),
 		)
+		log.Info("starting prometheus server")
 		prom.Listen(addr)
 	}
 }
 
 func logLevel() error {
-	viper.BindEnv("log.level", "LOGRUS_LEVEL")
 	lvl, err := log.ParseLevel(viper.GetString("log.level"))
 	if err != nil {
 		return err
@@ -100,6 +103,8 @@ func init() {
 	viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
 	viper.AutomaticEnv()
 
+	rootCmd.PersistentFlags().String("http-path", "", "vdb server http path")
+	rootCmd.PersistentFlags().String("ipc-path", "", "vdb server ipc path")
 	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file location")
 	rootCmd.PersistentFlags().String("log-file", "", "file path for logging")
 	rootCmd.PersistentFlags().String("log-level", log.InfoLevel.String(),
@@ -107,7 +112,11 @@ func init() {
 
 	rootCmd.PersistentFlags().String("leveldb-path", "", "path to primary datastore")
 	rootCmd.PersistentFlags().String("ancient-path", "", "path to ancient datastore")
-	rootCmd.PersistentFlags().Int("workers", 0, "number of concurrent workers to use")
+
+	rootCmd.PersistentFlags().Bool("prerun", false, "turn on prerun of toml configured ranges")
+	rootCmd.PersistentFlags().Int("service-workers", 0, "number of range requests to process concurrently")
+	rootCmd.PersistentFlags().Int("trie-workers", 0, "number of workers to use for trie traversal and processing")
+	rootCmd.PersistentFlags().Int("worker-queue-size", 0, "size of the range request queue for service workers")
 
 	rootCmd.PersistentFlags().String("database-name", "vulcanize_public", "database name")
 	rootCmd.PersistentFlags().Int("database-port", 5432, "database port")
@@ -128,12 +137,21 @@ func init() {
 	rootCmd.PersistentFlags().Bool("prom-http", false, "enable prometheus http service")
 	rootCmd.PersistentFlags().String("prom-http-addr", "127.0.0.1", "prometheus http host")
 	rootCmd.PersistentFlags().String("prom-http-port", "8080", "prometheus http port")
+	rootCmd.PersistentFlags().Bool("prom-db-stats", false, "enables prometheus db stats")
+	rootCmd.PersistentFlags().Bool("prom-metrics", false, "enable prometheus metrics")
 
-	rootCmd.PersistentFlags().Bool("metrics", false, "enable metrics")
+	rootCmd.PersistentFlags().Bool("prerun-only", false, "only process pre-configured ranges; exit afterwards")
+	rootCmd.PersistentFlags().Int("prerun-start", 0, "start height for a prerun range")
+	rootCmd.PersistentFlags().Int("prerun-stop", 0, "stop height for a prerun range")
 
+	viper.BindPFlag("server.httpPath", rootCmd.PersistentFlags().Lookup("http-path"))
+	viper.BindPFlag("server.ipcPath", rootCmd.PersistentFlags().Lookup("ipc-path"))
 	viper.BindPFlag("log.file", rootCmd.PersistentFlags().Lookup("log-file"))
 	viper.BindPFlag("log.level", rootCmd.PersistentFlags().Lookup("log-level"))
-	viper.BindPFlag("statediff.workers", rootCmd.PersistentFlags().Lookup("workers"))
+	viper.BindPFlag("statediff.prerun", rootCmd.PersistentFlags().Lookup("prerun"))
+	viper.BindPFlag("statediff.serviceWorkers", rootCmd.PersistentFlags().Lookup("service-workers"))
+	viper.BindPFlag("statediff.trieWorkers", rootCmd.PersistentFlags().Lookup("trie-workers"))
+	viper.BindPFlag("statediff.workerQueueSize", rootCmd.PersistentFlags().Lookup("worker-queue-size"))
 	viper.BindPFlag("leveldb.path", rootCmd.PersistentFlags().Lookup("leveldb-path"))
 	viper.BindPFlag("leveldb.ancient", rootCmd.PersistentFlags().Lookup("ancient-path"))
 	viper.BindPFlag("database.name", rootCmd.PersistentFlags().Lookup("database-name"))
@@ -151,7 +169,13 @@ func init() {
 	viper.BindPFlag("prom.http", rootCmd.PersistentFlags().Lookup("prom-http"))
 	viper.BindPFlag("prom.httpAddr", rootCmd.PersistentFlags().Lookup("prom-http-addr"))
 	viper.BindPFlag("prom.httpPort", rootCmd.PersistentFlags().Lookup("prom-http-port"))
-	viper.BindPFlag("prom.metrics", rootCmd.PersistentFlags().Lookup("metrics"))
+	viper.BindPFlag("prom.dbStats", rootCmd.PersistentFlags().Lookup("prom-db-stats"))
+	viper.BindPFlag("prom.metrics", rootCmd.PersistentFlags().Lookup("prom-metrics"))
+	viper.BindPFlag("prerun.only", rootCmd.PersistentFlags().Lookup("prerun-only"))
+	viper.BindPFlag("prerun.start", rootCmd.PersistentFlags().Lookup("prerun-start"))
+	viper.BindPFlag("prerun.stop", rootCmd.PersistentFlags().Lookup("prerun-stop"))
+
+	rand.Seed(time.Now().UnixNano())
 }
 
 func initConfig() {
@@ -168,13 +192,50 @@ func initConfig() {
 }
 
 func GetEthNodeInfo() node.Info {
-	return node.Info{
-		ID:           viper.GetString("ethereum.nodeID"),
-		ClientName:   viper.GetString("ethereum.clientName"),
-		GenesisBlock: viper.GetString("ethereum.genesisBlock"),
-		NetworkID:    viper.GetString("ethereum.networkID"),
-		ChainID:      viper.GetUint64("ethereum.chainID"),
+	var nodeID, genesisBlock, networkID, clientName string
+	var chainID uint64
+	if !viper.IsSet("ethereum.nodeID") {
+		nodeID = randSeq(12)
+	} else {
+		nodeID = viper.GetString("ethereum.nodeID")
 	}
+	if !viper.IsSet("ethereum.genesisBlock") {
+		genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
+	} else {
+		genesisBlock = viper.GetString("ethereum.genesisBlock")
+	}
+	if !viper.IsSet("ethereum.chainID") {
+		chainID = 1
+	} else {
+		chainID = viper.GetUint64("ethereum.chainID")
+	}
+	if !viper.IsSet("ethereum.networkID") {
+		networkID = "1"
+	} else {
+		networkID = viper.GetString("ethereum.networkID")
+	}
+	if !viper.IsSet("ethereum.clientName") {
+		clientName = "eth-statediff-service"
+	} else {
+		clientName = viper.GetString("ethereum.clientName")
+	}
+	return node.Info{
+		ID:           nodeID,
+		ClientName:   clientName,
+		GenesisBlock: genesisBlock,
+		NetworkID:    networkID,
+		ChainID:      chainID,
+	}
+}
+
+var characters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
+
+func randSeq(n int) string {
+	b := make([]rune, n)
+	for i := range b {
+		b[i] = characters[rand.Intn(len(characters))]
+	}
+	return string(b)
 }
 
 func GetDBParams() postgres.ConnectionParams {

cmd/serve.go

@@ -16,25 +16,23 @@
 package cmd
 
 import (
-	"fmt"
 	"os"
 	"os/signal"
 	"sync"
 
-	"github.com/ethereum/go-ethereum/node"
-	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
 	sd "github.com/vulcanize/eth-statediff-service/pkg"
+	srpc "github.com/vulcanize/eth-statediff-service/pkg/rpc"
 )
 
 // serveCmd represents the serve command
 var serveCmd = &cobra.Command{
 	Use:   "serve",
-	Short: "Standup a standalone statediffing RPC service on top of leveldb",
+	Short: "Stand up a standalone statediffing RPC service on top of leveldb",
 	Long: `Usage
 
 ./eth-statediff-service serve --config={path to toml config file}`,
@@ -45,6 +43,10 @@ var serveCmd = &cobra.Command{
 	},
 }
 
+func init() {
+	rootCmd.AddCommand(serveCmd)
+}
+
 func serve() {
 	logWithCommand.Info("Running eth-statediff-service serve command")
 
@@ -53,10 +55,20 @@ func serve() {
 		logWithCommand.Fatal(err)
 	}
 
+	// short circuit if we only want to perform prerun
+	if viper.GetBool("prerun.only") {
+		if err := statediffService.Run(nil); err != nil {
+			logWithCommand.Fatal("unable to perform prerun: %v", err)
+		}
+		return
+	}
+
 	// start service and servers
 	logWithCommand.Info("Starting statediff service")
 	wg := new(sync.WaitGroup)
-	go statediffService.Loop(wg)
+	if err := statediffService.Loop(wg); err != nil {
+		logWithCommand.Fatalf("unable to start statediff service: %v", err)
+	}
 	logWithCommand.Info("Starting RPC servers")
 	if err := startServers(statediffService); err != nil {
 		logWithCommand.Fatal(err)
@@ -72,19 +84,7 @@ func serve() {
 	wg.Wait()
 }
 
-func init() {
-	rootCmd.AddCommand(serveCmd)
-
-	serveCmd.PersistentFlags().String("http-path", "", "vdb server http path")
-	serveCmd.PersistentFlags().String("ipc-path", "", "vdb server ipc path")
-
-	viper.BindPFlag("server.httpPath", serveCmd.PersistentFlags().Lookup("http-path"))
-	viper.BindPFlag("server.ipcPath", serveCmd.PersistentFlags().Lookup("ipc-path"))
-}
-
-func startServers(serv sd.IService) error {
-	viper.BindEnv("server.ipcPath", "SERVER_IPC_PATH")
-	viper.BindEnv("server.httpPath", "SERVER_HTTP_PATH")
+func startServers(serv sd.StateDiffService) error {
 	ipcPath := viper.GetString("server.ipcPath")
 	httpPath := viper.GetString("server.httpPath")
 	if ipcPath == "" && httpPath == "" {
@@ -92,32 +92,20 @@ func startServers(serv sd.IService) error {
 	}
 	if ipcPath != "" {
 		logWithCommand.Info("starting up IPC server")
-		_, _, err := rpc.StartIPCEndpoint(ipcPath, serv.APIs())
+		_, _, err := srpc.StartIPCEndpoint(ipcPath, serv.APIs())
 		if err != nil {
 			return err
 		}
 	}
 	if httpPath != "" {
 		logWithCommand.Info("starting up HTTP server")
-		handler := rpc.NewServer()
-		if _, _, err := node.StartHTTPEndpoint(httpPath, rpc.HTTPTimeouts{}, handler); err != nil {
+		_, err := srpc.StartHTTPEndpoint(httpPath, serv.APIs(), []string{"statediff"}, nil, []string{"*"}, rpc.HTTPTimeouts{})
+		if err != nil {
 			return err
 		}
+	} else {
+		logWithCommand.Info("HTTP server is disabled")
 	}
+
 	return nil
 }
-
-func chainConfig(chainID uint64) (*params.ChainConfig, error) {
-	switch chainID {
-	case 1:
-		return params.MainnetChainConfig, nil
-	case 3:
-		return params.RopstenChainConfig, nil // Ropsten
-	case 4:
-		return params.RinkebyChainConfig, nil
-	case 5:
-		return params.GoerliChainConfig, nil
-	default:
-		return nil, fmt.Errorf("chain config for chainid %d not available", chainID)
-	}
-}

cmd/util.go

@@ -1,6 +1,11 @@
 package cmd
 
 import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/params"
+	gethsd "github.com/ethereum/go-ethereum/statediff"
 	ind "github.com/ethereum/go-ethereum/statediff/indexer"
 	"github.com/ethereum/go-ethereum/statediff/indexer/node"
 	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
@@ -11,7 +16,10 @@ import (
 	"github.com/vulcanize/eth-statediff-service/pkg/prom"
 )
 
-func createStateDiffService() (sd.IService, error) {
+type blockRange [2]uint64
+
+func createStateDiffService() (sd.StateDiffService, error) {
+	// load some necessary params
 	logWithCommand.Info("Loading statediff service parameters")
 	path := viper.GetString("leveldb.path")
 	ancientPath := viper.GetString("leveldb.ancient")
@@ -20,48 +28,124 @@ func createStateDiffService() (sd.IService, error) {
 	}
 
 	nodeInfo := GetEthNodeInfo()
-	config, err := chainConfig(nodeInfo.ChainID)
+	chainConf, err := chainConfig(nodeInfo.ChainID)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
 	// create leveldb reader
 	logWithCommand.Info("Creating leveldb reader")
-	conf := sd.ReaderConfig{
+	readerConf := sd.LvLDBReaderConfig{
 		TrieConfig: &trie.Config{
 			Cache:     viper.GetInt("cache.trie"),
 			Journal:   "",
 			Preimages: false,
 		},
-		ChainConfig: config,
+		ChainConfig: chainConf,
 		Path:        path,
 		AncientPath: ancientPath,
 		DBCacheSize: viper.GetInt("cache.database"),
 	}
-	lvlDBReader, err := sd.NewLvlDBReader(conf)
+	lvlDBReader, err := sd.NewLvlDBReader(readerConf)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
 	// create statediff service
-	logWithCommand.Info("Creating statediff service")
+	logWithCommand.Info("Setting up Postgres DB")
 	db, err := setupPostgres(nodeInfo)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
-	indexer, err := ind.NewStateDiffIndexer(config, db)
+	logWithCommand.Info("Creating statediff indexer")
+	indexer, err := ind.NewStateDiffIndexer(chainConf, db)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
-	return sd.NewStateDiffService(lvlDBReader, indexer, viper.GetUint("statediff.workers"))
+	logWithCommand.Info("Creating statediff service")
+	sdConf := sd.Config{
+		ServiceWorkers:  viper.GetUint("statediff.serviceWorkers"),
+		TrieWorkers:     viper.GetUint("statediff.trieWorkers"),
+		WorkerQueueSize: viper.GetUint("statediff.workerQueueSize"),
+		PreRuns:         setupPreRunRanges(),
+	}
+	return sd.NewStateDiffService(lvlDBReader, indexer, sdConf)
 }
 
 func setupPostgres(nodeInfo node.Info) (*postgres.DB, error) {
-	params := GetDBParams()
-	db, err := postgres.NewDB(postgres.DbConnectionString(params), GetDBConfig(), nodeInfo)
+	p := GetDBParams()
+	logWithCommand.Info("initializing DB connection pool")
+	db, err := postgres.NewDB(postgres.DbConnectionString(p), GetDBConfig(), nodeInfo)
 	if err != nil {
 		return nil, err
 	}
-	prom.RegisterDBCollector(params.Name, db.DB)
+	if viper.GetBool("prom.dbStats") {
+		logWithCommand.Info("registering DB collector")
+		prom.RegisterDBCollector(p.Name, db.DB)
+	}
 	return db, nil
 }
+
+func setupPreRunRanges() []sd.RangeRequest {
+	if !viper.GetBool("statediff.prerun") {
+		return nil
+	}
+	preRunParams := gethsd.Params{
+		IntermediateStateNodes:   viper.GetBool("prerun.params.intermediateStateNodes"),
+		IntermediateStorageNodes: viper.GetBool("prerun.params.intermediateStorageNodes"),
+		IncludeBlock:             viper.GetBool("prerun.params.includeBlock"),
+		IncludeReceipts:          viper.GetBool("prerun.params.includeReceipts"),
+		IncludeTD:                viper.GetBool("prerun.params.includeTD"),
+		IncludeCode:              viper.GetBool("prerun.params.includeCode"),
+	}
+	var addrStrs []string
+	viper.UnmarshalKey("prerun.params.watchedAddresses", &addrStrs)
+	addrs := make([]common.Address, len(addrStrs))
+	for i, addrStr := range addrStrs {
+		addrs[i] = common.HexToAddress(addrStr)
+	}
+	preRunParams.WatchedAddresses = addrs
+	var storageKeyStrs []string
+	viper.UnmarshalKey("prerun.params.watchedStorageKeys", &storageKeyStrs)
+	keys := make([]common.Hash, len(storageKeyStrs))
+	for i, keyStr := range storageKeyStrs {
+		keys[i] = common.HexToHash(keyStr)
+	}
+	preRunParams.WatchedStorageSlots = keys
+	var rawRanges []blockRange
+	viper.UnmarshalKey("prerun.ranges", &rawRanges)
+	blockRanges := make([]sd.RangeRequest, len(rawRanges))
+	for i, rawRange := range rawRanges {
+		blockRanges[i] = sd.RangeRequest{
+			Start:  rawRange[0],
+			Stop:   rawRange[1],
+			Params: preRunParams,
+		}
+	}
+	if viper.IsSet("prerun.start") && viper.IsSet("prerun.stop") {
+		hardStart := viper.GetInt("prerun.start")
+		hardStop := viper.GetInt("prerun.stop")
+		blockRanges = append(blockRanges, sd.RangeRequest{
+			Start:  uint64(hardStart),
+			Stop:   uint64(hardStop),
+			Params: preRunParams,
+		})
+	}
+
+	return blockRanges
+}
+
+func chainConfig(chainID uint64) (*params.ChainConfig, error) {
+	switch chainID {
+	case 1:
+		return params.MainnetChainConfig, nil
+	case 3:
+		return params.RopstenChainConfig, nil // Ropsten
+	case 4:
+		return params.RinkebyChainConfig, nil
+	case 5:
+		return params.GoerliChainConfig, nil
+	default:
+		return nil, fmt.Errorf("chain config for chainid %d not available", chainID)
+	}
+}

cmd/write.go

@@ -1,142 +0,0 @@
-// Copyright © 2019 Vulcanize, Inc
-//
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// This program 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 Affero General Public License for more details.
-//
-// You should have received a copy of the GNU Affero General Public License
-// along with this program. If not, see <http://www.gnu.org/licenses/>.
-
-package cmd
-
-import (
-	"fmt"
-	"net/http"
-	"strconv"
-	"sync"
-	"time"
-
-	gethsd "github.com/ethereum/go-ethereum/statediff"
-	"github.com/sirupsen/logrus"
-	"github.com/spf13/cobra"
-	"github.com/spf13/viper"
-)
-
-var writeCmd = &cobra.Command{
-	Use:   "write",
-	Short: "Write statediffs directly to DB for preconfigured block ranges",
-	Long: `Usage
-
-./eth-statediff-service write --config={path to toml config file}`,
-	Run: func(cmd *cobra.Command, args []string) {
-		subCommand = cmd.CalledAs()
-		logWithCommand = *logrus.WithField("SubCommand", subCommand)
-		write()
-	},
-}
-
-type blockRange [2]uint64
-
-func init() {
-	rootCmd.AddCommand(writeCmd)
-	writeCmd.PersistentFlags().String("write-api", "", "starts a server which handles write request through endpoints")
-	viper.BindPFlag("write.serve", writeCmd.PersistentFlags().Lookup("write-api"))
-}
-
-func write() {
-	logWithCommand.Info("Running eth-statediff-service write command")
-	viper.BindEnv("write.serve", WRITE_SERVER)
-	addr := viper.GetString("write.serve")
-
-	statediffService, err := createStateDiffService()
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	// Read all defined block ranges, write statediffs to database
-	var blockRanges []blockRange
-	diffParams := gethsd.Params{ // todo: configurable?
-		IntermediateStateNodes:   true,
-		IntermediateStorageNodes: true,
-		IncludeBlock:             true,
-		IncludeReceipts:          true,
-		IncludeTD:                true,
-		IncludeCode:              true,
-	}
-	viper.UnmarshalKey("write.ranges", &blockRanges)
-	viper.UnmarshalKey("write.params", &diffParams)
-
-	blockRangesCh := make(chan blockRange, 100)
-	go func() {
-		for _, r := range blockRanges {
-			blockRangesCh <- r
-		}
-		if addr == "" {
-			close(blockRangesCh)
-			return
-		}
-		startServer(addr, blockRangesCh)
-	}()
-
-	processRanges(statediffService, diffParams, blockRangesCh)
-}
-
-func startServer(addr string, blockRangesCh chan<- blockRange) {
-	handler := func(w http.ResponseWriter, req *http.Request) {
-		start, err := strconv.Atoi(req.URL.Query().Get("start"))
-		if err != nil {
-			http.Error(w, fmt.Sprintf("failed to parse start value: %v", err), http.StatusInternalServerError)
-			return
-		}
-
-		end, err := strconv.Atoi(req.URL.Query().Get("end"))
-		if err != nil {
-			http.Error(w, fmt.Sprintf("failed to parse end value: %v", err), http.StatusInternalServerError)
-			return
-		}
-
-		select {
-		case blockRangesCh <- blockRange{uint64(start), uint64(end)}:
-		case <-time.After(time.Millisecond * 200):
-			http.Error(w, "server is busy", http.StatusInternalServerError)
-			return
-		}
-
-		fmt.Fprintf(w, "added block range to the queue\n")
-	}
-
-	http.HandleFunc("/writeDiff", handler)
-	logrus.Fatal(http.ListenAndServe(addr, nil))
-}
-
-type diffService interface {
-	WriteStateDiffAt(blockNumber uint64, params gethsd.Params) error
-}
-
-func processRanges(sds diffService, param gethsd.Params, blockRangesCh chan blockRange) {
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		for rng := range blockRangesCh {
-			if rng[1] < rng[0] {
-				logWithCommand.Fatal("range ending block number needs to be greater than starting block number")
-			}
-			logrus.Infof("Writing statediffs from block %d to %d", rng[0], rng[1])
-			for height := rng[0]; height <= rng[1]; height++ {
-				err := sds.WriteStateDiffAt(height, param)
-				if err != nil {
-					logrus.Errorf("failed to write state diff for range: %v %v", rng, err)
-				}
-			}
-		}
-	}()
-
-	wg.Wait()
-}

cmd/write_test.go

@@ -1,76 +0,0 @@
-package cmd
-
-import (
-	"fmt"
-	"net/http"
-	"testing"
-	"time"
-
-	gethsd "github.com/ethereum/go-ethereum/statediff"
-	"github.com/stretchr/testify/require"
-)
-
-type mockService struct {
-	reqCount int
-}
-
-func (ms *mockService) WriteStateDiffAt(_ uint64, _ gethsd.Params) error {
-	ms.reqCount++
-	return nil
-}
-
-func TestProcessRanges(t *testing.T) {
-	blockRangesCh := make(chan blockRange)
-	srv := new(mockService)
-
-	go func() {
-		blockRangesCh <- blockRange{uint64(1), uint64(5)}
-		blockRangesCh <- blockRange{uint64(8), uint64(10)}
-		blockRangesCh <- blockRange{uint64(6), uint64(7)}
-		blockRangesCh <- blockRange{uint64(50), uint64(100)}
-		blockRangesCh <- blockRange{uint64(5), uint64(8)}
-		close(blockRangesCh)
-	}()
-
-	processRanges(srv, gethsd.Params{}, blockRangesCh)
-	require.Equal(t, 65, srv.reqCount)
-}
-
-func TestHttpEndpoint(t *testing.T) {
-	addr := ":11111"
-	queueSize := 5
-	blockRangesCh := make(chan blockRange, queueSize)
-	srv := new(mockService)
-
-	go startServer(addr, blockRangesCh)
-
-	go func() {
-		br := []blockRange{
-			{uint64(1), uint64(5)},
-			{uint64(8), uint64(10)},
-			{uint64(6), uint64(7)},
-			{uint64(50), uint64(100)},
-			{uint64(5), uint64(8)},
-			// Below request should fail since server has queue size of 5
-			{uint64(52), uint64(328)},
-			{uint64(35), uint64(428)},
-			{uint64(45), uint64(844)},
-		}
-
-		for idx, r := range br {
-			res, err := http.Get(fmt.Sprintf("http://localhost:11111/writeDiff?start=%d&end=%d", r[0], r[1]))
-			require.NoError(t, err)
-			require.NotNil(t, res)
-			if idx < queueSize {
-				require.Equal(t, res.StatusCode, 200)
-			} else {
-				require.Equal(t, res.StatusCode, 500)
-			}
-			require.NoError(t, res.Body.Close())
-		}
-		processRanges(srv, gethsd.Params{}, blockRangesCh)
-	}()
-
-	time.Sleep(1 * time.Second)
-	require.Equal(t, 65, srv.reqCount)
-}

environments/config.toml

@@ -0,0 +1,51 @@
+[leveldb]
+    path = "/app/geth-rw/chaindata"
+    ancient = "/app/geth-rw/chaindata/ancient"
+
+[server]
+    ipcPath = ""
+    httpPath = "0.0.0.0:8545"
+
+[statediff]
+    prerun = true
+    serviceWorkers = 1
+    workerQueueSize = 1024
+    trieWorkers = 16
+
+[prerun]
+    only = true
+    ranges = []
+    [prerun.params]
+        intermediateStateNodes = true
+        intermediateStorageNodes = true
+        includeBlock = true
+        includeReceipts = true
+        includeTD = true
+        includeCode = true
+        watchedAddresses = []
+        watchedStorageKeys = []
+
+[log]
+    file = ""
+    level = "info"
+
+[eth]
+    chainID = 1
+
+[database]
+    name     = ""
+    hostname = ""
+    port     = 5432
+    user     = ""
+    password = ""
+
+[cache]
+    database = 1024
+    trie = 4096
+
+[prom]
+    dbStats = false
+    metrics = true
+    http = true
+    httpAddr = "0.0.0.0"
+    httpPort = 9100

environments/example.toml

@@ -7,13 +7,25 @@
     httpPath = "127.0.0.1:8545"
 
 [statediff]
-    workers = 4
+    prerun = true
+    serviceWorkers = 1
+    workerQueueSize = 1024
+    trieWorkers = 4
 
-[write]
+[prerun]
+    only = false
     ranges = [
-        [0, 0]
+        [0, 1000]
     ]
-    serve = ":8888"
+    [prerun.params]
+        intermediateStateNodes = true
+        intermediateStorageNodes = true
+        includeBlock = true
+        includeReceipts = true
+        includeTD = true
+        includeCode = true
+        watchedAddresses = []
+        watchedStorageKeys = []
 
 [log]
     file = ""
@@ -34,7 +46,8 @@
     trie = 1024
 
 [prom]
+    dbStats = false
     metrics = true
     http = true
-    addr = "localhost"
-    port = "8889"
+    httpAddr = "localhost"
+    httpPort = "8889"

go.mod

@@ -3,14 +3,13 @@ module github.com/vulcanize/eth-statediff-service
 go 1.13
 
 require (
-	github.com/ethereum/go-ethereum v1.10.9
+	github.com/ethereum/go-ethereum v1.10.13
 	github.com/jmoiron/sqlx v1.2.0
 	github.com/prometheus/client_golang v1.0.0
 	github.com/sirupsen/logrus v1.7.0
 	github.com/spf13/cobra v1.1.1
 	github.com/spf13/viper v1.7.1
-	github.com/stretchr/testify v1.7.0
 	github.com/vulcanize/go-eth-state-node-iterator v0.0.1-alpha.0.20211014064906-d23d01ed8191
 )
 
-replace github.com/ethereum/go-ethereum v1.10.9 => github.com/vulcanize/go-ethereum v1.10.11-statediff-0.0.27
+replace github.com/ethereum/go-ethereum v1.10.13 => github.com/vulcanize/go-ethereum v1.10.13-statediff-0.0.29

go.sum

@@ -109,12 +109,16 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/deckarep/golang-set v0.0.0-20180603214616-504e848d77ea h1:j4317fAZh7X6GqbFowYdYdI0L9bwxL07jyPZIdepyZ0=
 github.com/deckarep/golang-set v0.0.0-20180603214616-504e848d77ea/go.mod h1:93vsz/8Wt4joVM7c2AVqh+YRMiUSc14yDtF28KmMOgQ=
 github.com/deepmap/oapi-codegen v1.6.0/go.mod h1:ryDa9AgbELGeB+YEXE1dR53yAjHwFvE9iAUlWl9Al3M=
+github.com/deepmap/oapi-codegen v1.8.2 h1:SegyeYGcdi0jLLrpbCMoJxnUUn8GBXHsvr4rbzjuhfU=
 github.com/deepmap/oapi-codegen v1.8.2/go.mod h1:YLgSKSDv/bZQB7N4ws6luhozi3cEdRktEqrX88CvjIw=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-bitstream v0.0.0-20180413035011-3522498ce2c8/go.mod h1:VMaSuZ+SZcx/wljOQKvp5srsbCiKDEb6K2wC4+PiBmQ=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
+github.com/dlclark/regexp2 v1.2.0/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
 github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
+github.com/docker/docker v1.4.2-0.20180625184442-8e610b2b55bf h1:sh8rkQZavChcmakYiSlqu2425CHyFXLZZnvm7PDpU8M=
 github.com/docker/docker v1.4.2-0.20180625184442-8e610b2b55bf/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/dop251/goja v0.0.0-20200721192441-a695b0cdd498/go.mod h1:Mw6PkjjMXWbTj+nnj4s3QPXq1jaT0s5pC0iFD4+BOAA=
 github.com/dop251/goja v0.0.0-20211011172007-d99e4b8cbf48/go.mod h1:R9ET47fwRVRPZnOGvHxxhuZcbrMCuiqOz3Rlrh4KSnk=
 github.com/dop251/goja_nodejs v0.0.0-20210225215109-d91c329300e7/go.mod h1:hn7BA7c8pLvoGndExHudxTDKZ84Pyvv+90pbBjbTz0Y=
 github.com/eclipse/paho.mqtt.golang v1.2.0/go.mod h1:H9keYFcgq3Qr5OUJm/JZI/i6U7joQ8SYLhZwfeOo6Ts=
@@ -122,6 +126,7 @@ github.com/edsrzf/mmap-go v1.0.0 h1:CEBF7HpRnUCSJgGUb5h1Gm7e3VkmVDrR8lvWVLtrOFw=
 github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/ethereum/go-ethereum v1.10.9/go.mod h1:CaTMQrv51WaAlD2eULQ3f03KiahDRO28fleQcKjWrrg=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fjl/memsize v0.0.0-20190710130421-bcb5799ab5e5 h1:FtmdgXiUlNeRsoNMFlKLDt+S+6hbjVMEW6RGQ7aUf7c=
 github.com/fjl/memsize v0.0.0-20190710130421-bcb5799ab5e5/go.mod h1:VvhXpOYNQvB+uIk2RvXzuaQtkQJzzIx6lSBe1xv7hi0=
@@ -148,6 +153,7 @@ github.com/go-ole/go-ole v1.2.1 h1:2lOsA72HgjxAuMlKpFiCbHTvu44PIVkZ5hqm3RSdI/E=
 github.com/go-ole/go-ole v1.2.1/go.mod h1:7FAglXiTm7HKlQRDeOQ6ZNUHidzCWXuZWq/1dTyBNF8=
 github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
+github.com/go-sourcemap/sourcemap v2.1.2+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.4.1 h1:g24URVg0OFbNUTx9qqY1IRZ9D9z3iPyi5zKhQZpNwpA=
@@ -210,6 +216,7 @@ github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORR
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/graph-gophers/graphql-go v0.0.0-20201113091052-beb923fada29 h1:sezaKhEfPFg8W0Enm61B9Gs911H8iesGY5R8NDPtd1M=
 github.com/graph-gophers/graphql-go v0.0.0-20201113091052-beb923fada29/go.mod h1:9CQHMSxwO4MprSdzoIEobiHpoLtHm77vfxsvsIN5Vuc=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
@@ -254,11 +261,14 @@ github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/influxdata/flux v0.65.1/go.mod h1:J754/zds0vvpfwuq7Gc2wRdVwEodfpCFM7mYlOw2LqY=
+github.com/influxdata/influxdb v1.8.3 h1:WEypI1BQFTT4teLM+1qkEcvUi0dAvopAI/ir0vAiBg8=
 github.com/influxdata/influxdb v1.8.3/go.mod h1:JugdFhsvvI8gadxOI6noqNeeBHvWNTbfYGtiAn+2jhI=
+github.com/influxdata/influxdb-client-go/v2 v2.4.0 h1:HGBfZYStlx3Kqvsv1h2pJixbCl/jhnFtxpKFAv9Tu5k=
 github.com/influxdata/influxdb-client-go/v2 v2.4.0/go.mod h1:vLNHdxTJkIf2mSLvGrpj8TCcISApPoXkaxP8g9uRlW8=
 github.com/influxdata/influxql v1.1.1-0.20200828144457-65d3ef77d385/go.mod h1:gHp9y86a/pxhjJ+zMjNXiQAA197Xk9wLxaz+fGG+kWk=
 github.com/influxdata/line-protocol v0.0.0-20180522152040-32c6aa80de5e/go.mod h1:4kt73NQhadE3daL3WhR5EJ/J2ocX0PZzwxQ0gXJ7oFE=
 github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
+github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097 h1:vilfsDSy7TDxedi9gyBkMvAirat/oRcL0lFdJBf6tdM=
 github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
 github.com/influxdata/promql/v2 v2.12.0/go.mod h1:fxOPu+DY0bqCTCECchSRtWfc+0X19ybifQhZoQNF5D8=
 github.com/influxdata/roaring v0.4.13-0.20180809181101-fc520f41fab6/go.mod h1:bSgUQ7q5ZLSO+bKBGqJiCBGAl+9DxyW63zLTujjUlOE=
@@ -307,9 +317,11 @@ github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/X
 github.com/jsternberg/zap-logfmt v1.0.0/go.mod h1:uvPs/4X51zdkcm5jXl5SYoN+4RK21K8mysFmDaM/h+o=
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
+github.com/julienschmidt/httprouter v1.2.0 h1:TDTW5Yz1mjftljbcKqRcrYhd4XeOoI98t+9HbQbYf7g=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/jwilder/encoding v0.0.0-20170811194829-b4e1701a28ef/go.mod h1:Ct9fl0F6iIOGgxJ5npU/IUOhOhqlVrGjyIZc8/MagT0=
+github.com/karalabe/usb v0.0.0-20190919080040-51dc0efba356/go.mod h1:Od972xHfMJowv7NGVDiWVxk2zxnWgjLlJzE+F4F7AGU=
 github.com/karalabe/usb v0.0.0-20211005121534-4c5740d64559/go.mod h1:Od972xHfMJowv7NGVDiWVxk2zxnWgjLlJzE+F4F7AGU=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
@@ -430,6 +442,7 @@ github.com/paulbellamy/ratecounter v0.2.0/go.mod h1:Hfx1hDpSGoqxkVVpBi/IlYD7kChl
 github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/peterh/liner v1.0.1-0.20180619022028-8c1271fcf47f/go.mod h1:xIteQHvHuaLYG9IFj6mSxM0fCKrs34IrEQUhOYuGPHc=
+github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM=
 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0=
 github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
@@ -531,8 +544,8 @@ github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPU
 github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
 github.com/vulcanize/go-eth-state-node-iterator v0.0.1-alpha.0.20211014064906-d23d01ed8191 h1:+xSwb0xOLCAhejoai2BQZPTJW5Z+tU9xkTaOBwGnPoU=
 github.com/vulcanize/go-eth-state-node-iterator v0.0.1-alpha.0.20211014064906-d23d01ed8191/go.mod h1:FBMQp69PTnRg3kLx9qeWGfKYfVq4fpfbr69X8lEvvfs=
-github.com/vulcanize/go-ethereum v1.10.11-statediff-0.0.27 h1:ldAVLlKll2WHHKLNu8oKbkYRUVRs9T45z12KiSrRW24=
-github.com/vulcanize/go-ethereum v1.10.11-statediff-0.0.27/go.mod h1:9L+QY31AnWnX2/2HDOySCjQoYUdWNGBRMezFJVfH73E=
+github.com/vulcanize/go-ethereum v1.10.13-statediff-0.0.29 h1:jU0msLliD4Ms4Ae2fK6ktKcqoaynmSmDkRDJg3XNqkQ=
+github.com/vulcanize/go-ethereum v1.10.13-statediff-0.0.29/go.mod h1:9L+QY31AnWnX2/2HDOySCjQoYUdWNGBRMezFJVfH73E=
 github.com/whyrusleeping/go-logging v0.0.0-20170515211332-0457bb6b88fc h1:9lDbC6Rz4bwmou+oE6Dt4Cb2BGMur5eR/GYptkKUVHo=
 github.com/whyrusleeping/go-logging v0.0.0-20170515211332-0457bb6b88fc/go.mod h1:bopw91TMyo8J3tvftk8xmU2kPmlrt4nScJQZU2hE5EM=
 github.com/willf/bitset v1.1.3/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
@@ -623,8 +636,9 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210220033124-5f55cee0dc0d/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 h1:4nGaVu0QrbjT/AK2PRLuQfQuh6DJve+pELhqTdAj3x0=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
+golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d h1:20cMwl2fHAzkJMEA+8J4JgqBQcQGzbisXo31MIeenXI=
+golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -679,6 +693,7 @@ golang.org/x/sys v0.0.0-20210316164454-77fc1eacc6aa/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210324051608-47abb6519492/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210420205809-ac73e9fd8988/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912 h1:uCLL3g5wH2xjxVREVuAbP9JM5PPKjRbXKRa6IBjkzmU=
 golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -808,6 +823,7 @@ gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

monitoring/grafana/dashboard.json

@@ -0,0 +1,449 @@
+{
+  "annotations": {
+    "list": [
+      {
+        "builtIn": 1,
+        "datasource": "-- Grafana --",
+        "enable": true,
+        "hide": true,
+        "iconColor": "rgba(0, 211, 255, 1)",
+        "name": "Annotations & Alerts",
+        "type": "dashboard"
+      }
+    ]
+  },
+  "editable": true,
+  "gnetId": null,
+  "graphTooltip": 0,
+  "id": 4,
+  "iteration": 1635774999288,
+  "links": [
+    {
+      "icon": "info",
+      "tags": [],
+      "targetBlank": true,
+      "title": "Grafana Docs",
+      "tooltip": "",
+      "type": "link",
+      "url": "https://grafana.com/docs/grafana/latest/"
+    },
+    {
+      "icon": "info",
+      "tags": [],
+      "targetBlank": true,
+      "title": "Prometheus Docs",
+      "type": "link",
+      "url": "http://prometheus.io/docs/introduction/overview/"
+    }
+  ],
+  "panels": [
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "eth-statediff-service",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {},
+          "links": [],
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 80
+              }
+            ]
+          }
+        },
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 6,
+        "w": 18,
+        "x": 0,
+        "y": 0
+      },
+      "hiddenSeries": false,
+      "id": 3,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "links": [],
+      "nullPointMode": "null",
+      "percentage": false,
+      "pluginVersion": "7.1.5",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "expr": "rate(eth_statediff_service_processed_height{instance=\"$instance\"}[1m])",
+          "interval": "",
+          "legendFormat": "",
+          "refId": "A"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Instance block processed (rate-1m)",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        },
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        }
+      ],
+      "yaxis": {
+        "align": false,
+        "alignLevel": null
+      }
+    },
+    {
+      "content": "<span style=\"font-family: 'Open Sans', 'Helvetica Neue', Helvetica; font-size: 25px;vertical-align: text-top;color: #bbbfc2;margin-left: 10px;\">Prometheus</span>\n\n<p style=\"margin-top: 10px;\">You're using Prometheus, an open-source systems monitoring and alerting toolkit originally built at SoundCloud. For more information, check out the <a href=\"https://grafana.com/\">Grafana</a> and <a href=\"http://prometheus.io/\">Prometheus</a> projects.</p>",
+      "datasource": null,
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 5,
+        "w": 6,
+        "x": 18,
+        "y": 0
+      },
+      "id": 9,
+      "links": [],
+      "mode": "html",
+      "options": {
+        "content": "<span style=\"font-family: 'Open Sans', 'Helvetica Neue', Helvetica; font-size: 25px;vertical-align: text-top;color: #bbbfc2;margin-left: 10px;\">Prometheus</span>\n\n<p style=\"margin-top: 10px;\">You're using Prometheus, an open-source systems monitoring and alerting toolkit originally built at SoundCloud. For more information, check out the <a href=\"https://grafana.com/\">Grafana</a> and <a href=\"http://prometheus.io/\">Prometheus</a> projects.</p>",
+        "mode": "html"
+      },
+      "pluginVersion": "7.1.0",
+      "style": {},
+      "title": "",
+      "transparent": true,
+      "type": "text"
+    },
+    {
+      "content": "#### Samples Ingested\nThis graph displays the count of samples ingested by the Prometheus server, as measured over the last 5 minutes, per time series in the range vector. When troubleshooting an issue on IRC or Github, this is often the first stat requested by the Prometheus team. ",
+      "datasource": null,
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 6,
+        "w": 4,
+        "x": 18,
+        "y": 5
+      },
+      "id": 8,
+      "links": [],
+      "mode": "markdown",
+      "options": {
+        "content": "#### Samples Ingested\nThis graph displays the count of samples ingested by the Prometheus server, as measured over the last 5 minutes, per time series in the range vector. When troubleshooting an issue on IRC or Github, this is often the first stat requested by the Prometheus team. ",
+        "mode": "markdown"
+      },
+      "pluginVersion": "7.1.0",
+      "style": {},
+      "title": "",
+      "transparent": true,
+      "type": "text"
+    },
+    {
+      "aliasColors": {
+        "Value": "red",
+        "aggr": "red"
+      },
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "eth-statediff-service",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {},
+          "links": [],
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 80
+              }
+            ]
+          }
+        },
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 6,
+        "w": 18,
+        "x": 0,
+        "y": 6
+      },
+      "hiddenSeries": false,
+      "id": 17,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "links": [],
+      "nullPointMode": "null",
+      "percentage": false,
+      "pluginVersion": "7.1.5",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "expr": "sum(increase(eth_statediff_service_processed_height{instance=~\".+\"}[1m]))",
+          "interval": "",
+          "legendFormat": "aggr",
+          "refId": "C"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Aggregate block processed (rate-1m)",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        },
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        }
+      ],
+      "yaxis": {
+        "align": false,
+        "alignLevel": null
+      }
+    },
+    {
+      "content": "#### Scrapes\nPrometheus scrapes metrics from instrumented jobs, either directly or via an intermediary push gateway for short-lived jobs. Target scrapes will show how frequently targets are scraped, as measured over the last 5 minutes, per time series in the range vector. Scrape Duration will show how long the scrapes are taking, with percentiles available as series. ",
+      "datasource": null,
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 7,
+        "w": 6,
+        "x": 18,
+        "y": 11
+      },
+      "id": 11,
+      "links": [],
+      "mode": "markdown",
+      "options": {
+        "content": "#### Scrapes\nPrometheus scrapes metrics from instrumented jobs, either directly or via an intermediary push gateway for short-lived jobs. Target scrapes will show how frequently targets are scraped, as measured over the last 5 minutes, per time series in the range vector. Scrape Duration will show how long the scrapes are taking, with percentiles available as series. ",
+        "mode": "markdown"
+      },
+      "pluginVersion": "7.1.0",
+      "style": {},
+      "title": "",
+      "transparent": true,
+      "type": "text"
+    },
+    {
+      "content": "#### Rule Evaluation Duration\nThis graph panel plots the duration for all evaluations to execute. The 50th percentile, 90th percentile and 99th percentile are shown as three separate series to help identify outliers that may be skewing the data.",
+      "datasource": null,
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 7,
+        "w": 6,
+        "x": 18,
+        "y": 18
+      },
+      "id": 15,
+      "links": [],
+      "mode": "markdown",
+      "options": {
+        "content": "#### Rule Evaluation Duration\nThis graph panel plots the duration for all evaluations to execute. The 50th percentile, 90th percentile and 99th percentile are shown as three separate series to help identify outliers that may be skewing the data.",
+        "mode": "markdown"
+      },
+      "pluginVersion": "7.1.0",
+      "style": {},
+      "title": "",
+      "transparent": true,
+      "type": "text"
+    }
+  ],
+  "refresh": false,
+  "revision": "1.0",
+  "schemaVersion": 26,
+  "style": "dark",
+  "tags": [],
+  "templating": {
+    "list": [
+      {
+        "allValue": null,
+        "current": {
+          "selected": true,
+          "text": "localhost:8888",
+          "value": "localhost:8888"
+        },
+        "datasource": "eth-statediff-service",
+        "definition": "label_values(instance)",
+        "hide": 0,
+        "includeAll": false,
+        "label": "SelectInstance",
+        "multi": false,
+        "name": "instance",
+        "options": [],
+        "query": "label_values(instance)",
+        "refresh": 2,
+        "regex": "",
+        "skipUrlSync": false,
+        "sort": 1,
+        "tagValuesQuery": "",
+        "tags": [],
+        "tagsQuery": "",
+        "type": "query",
+        "useTags": false
+      }
+    ]
+  },
+  "time": {
+    "from": "now-30m",
+    "to": "now"
+  },
+  "timepicker": {
+    "now": true,
+    "refresh_intervals": [
+      "5s",
+      "10s",
+      "30s",
+      "1m",
+      "5m",
+      "15m",
+      "30m",
+      "1h",
+      "2h",
+      "1d"
+    ],
+    "time_options": [
+      "5m",
+      "15m",
+      "1h",
+      "6h",
+      "12h",
+      "24h",
+      "2d",
+      "7d",
+      "30d"
+    ]
+  },
+  "timezone": "browser",
+  "title": "Prometheus Block Process Stats",
+  "uid": "q1FUJHKnz",
+  "version": 14
+}

pkg/api.go

@@ -30,11 +30,11 @@ const APIVersion = "0.0.1"
 // PublicStateDiffAPI provides an RPC interface
 // that can be used to fetch historical diffs from leveldb directly
 type PublicStateDiffAPI struct {
-	sds IService
+	sds StateDiffService
 }
 
 // NewPublicStateDiffAPI creates an rpc interface for the underlying statediff service
-func NewPublicStateDiffAPI(sds IService) *PublicStateDiffAPI {
+func NewPublicStateDiffAPI(sds StateDiffService) *PublicStateDiffAPI {
 	return &PublicStateDiffAPI{
 		sds: sds,
 	}
@@ -54,3 +54,8 @@ func (api *PublicStateDiffAPI) StateTrieAt(ctx context.Context, blockNumber uint
 func (api *PublicStateDiffAPI) WriteStateDiffAt(ctx context.Context, blockNumber uint64, params sd.Params) error {
 	return api.sds.WriteStateDiffAt(blockNumber, params)
 }
+
+// WriteStateDiffsInRange writes the state diff objects for the provided block range, with the provided params
+func (api *PublicStateDiffAPI) WriteStateDiffsInRange(ctx context.Context, start, stop uint64, params sd.Params) error {
+	return api.sds.WriteStateDiffsInRange(start, stop, params)
+}

pkg/builder.go

@@ -35,6 +35,7 @@ import (
 	sdtrie "github.com/ethereum/go-ethereum/statediff/trie"
 	sdtypes "github.com/ethereum/go-ethereum/statediff/types"
 	"github.com/ethereum/go-ethereum/trie"
+	"github.com/sirupsen/logrus"
 
 	iter "github.com/vulcanize/go-eth-state-node-iterator"
 )
@@ -206,7 +207,7 @@ func (sdb *builder) BuildStateDiffObject(args sd.Args, params sd.Params) (sd.Sta
 	}, nil
 }
 
-// Writes a statediff object to output callback
+// WriteStateDiffObject writes a statediff object to output callback
 func (sdb *builder) WriteStateDiffObject(args sd.StateRoots, params sd.Params, output sdtypes.StateNodeSink, codeOutput sdtypes.CodeSink) error {
 	if len(params.WatchedAddresses) > 0 {
 		// if we are watching only specific accounts then we are only diffing leaf nodes
@@ -234,8 +235,8 @@ func (sdb *builder) WriteStateDiffObject(args sd.StateRoots, params sd.Params, o
 	var iterPairs [][]iterPair
 	for i := uint(0); i < sdb.numWorkers; i++ {
 		iterPairs = append(iterPairs, []iterPair{
-			iterPair{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
-			iterPair{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
+			{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
+			{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
 		})
 	}
 
@@ -252,7 +253,9 @@ func (sdb *builder) WriteStateDiffObject(args sd.StateRoots, params sd.Params, o
 			wg.Add(1)
 			go func(worker uint) {
 				defer wg.Done()
-				sdb.buildStateDiff(iterPairs[worker], params, nodeSender, codeSender)
+				if err := sdb.buildStateDiff(iterPairs[worker], params, nodeSender, codeSender); err != nil {
+					logrus.Errorf("buildStateDiff error for worker %d, pparams %+v", worker, params)
+				}
 			}(w)
 		}
 		wg.Wait()

pkg/config.go

@@ -0,0 +1,9 @@
+package statediff
+
+// Config holds config params for the statediffing service
+type Config struct {
+	ServiceWorkers  uint
+	TrieWorkers     uint
+	WorkerQueueSize uint
+	PreRuns         []RangeRequest
+}

pkg/prom/db_stats_collector.go

@@ -23,7 +23,7 @@ import (
 )
 
 const (
-	namespace = "eth-statediff-service"
+	namespace = "eth_statediff_service"
 	subsystem = "connections"
 )
 

pkg/prom/middleware.go

@@ -0,0 +1,51 @@
+// VulcanizeDB
+// Copyright © 2020 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program 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 Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package prom
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+// HTTPMiddleware http connection metric reader
+func HTTPMiddleware(next http.Handler) http.Handler {
+	if !metrics {
+		return next
+	}
+
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		httpCount.Inc()
+
+		start := time.Now()
+		next.ServeHTTP(w, r)
+		duration := time.Now().Sub(start)
+		httpDuration.Observe(float64(duration.Seconds()))
+	})
+}
+
+// IPCMiddleware unix-socket connection counter
+func IPCMiddleware(server *rpc.Server, client rpc.Conn) {
+	if metrics {
+		ipcCount.Inc()
+	}
+	server.ServeCodec(rpc.NewCodec(client), 0)
+	if metrics {
+		ipcCount.Dec()
+	}
+}

pkg/prom/prom.go

@@ -24,11 +24,16 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
-const statsSubsystem = "stats"
+const (
+	statsSubsystem = "stats"
+	subsystemHTTP  = "http"
+	subsystemIPC   = "ipc"
+)
 
 var (
 	metrics bool
 
+	queuedRanges        prometheus.Gauge
 	lastLoadedHeight    prometheus.Gauge
 	lastProcessedHeight prometheus.Gauge
 
@@ -36,56 +41,108 @@ var (
 	tBlockProcessing prometheus.Histogram
 	tStateProcessing prometheus.Histogram
 	tTxCommit        prometheus.Histogram
+
+	httpCount    prometheus.Counter
+	httpDuration prometheus.Histogram
+	ipcCount     prometheus.Gauge
+)
+
+const (
+	RANGES_QUEUED        = "ranges_queued"
+	LOADED_HEIGHT        = "loaded_height"
+	PROCESSED_HEIGHT     = "processed_height"
+	T_BLOCK_LOAD         = "t_block_load"
+	T_BLOCK_PROCESSING   = "t_block_processing"
+	T_STATE_PROCESSING   = "t_state_processing"
+	T_POSTGRES_TX_COMMIT = "t_postgres_tx_commit"
 )
 
 // Init module initialization
 func Init() {
 	metrics = true
 
+	queuedRanges = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Name:      RANGES_QUEUED,
+		Help:      "Number of range requests currently queued",
+	})
 	lastLoadedHeight = promauto.NewGauge(prometheus.GaugeOpts{
 		Namespace: namespace,
-		Name:      "loaded_height",
+		Name:      LOADED_HEIGHT,
 		Help:      "The last block that was loaded for processing",
 	})
 	lastProcessedHeight = promauto.NewGauge(prometheus.GaugeOpts{
 		Namespace: namespace,
-		Name:      "processed_height",
+		Name:      PROCESSED_HEIGHT,
 		Help:      "The last block that was processed",
 	})
 
 	tBlockLoad = promauto.NewHistogram(prometheus.HistogramOpts{
 		Namespace: namespace,
 		Subsystem: statsSubsystem,
-		Name:      "t_block_load",
+		Name:      T_BLOCK_LOAD,
 		Help:      "Block loading time",
 	})
 	tBlockProcessing = promauto.NewHistogram(prometheus.HistogramOpts{
 		Namespace: namespace,
 		Subsystem: statsSubsystem,
-		Name:      "t_block_processing",
+		Name:      T_BLOCK_PROCESSING,
 		Help:      "Block (header, uncles, txs, rcts, tx trie, rct trie) processing time",
 	})
 	tStateProcessing = promauto.NewHistogram(prometheus.HistogramOpts{
 		Namespace: namespace,
 		Subsystem: statsSubsystem,
-		Name:      "t_state_processing",
+		Name:      T_STATE_PROCESSING,
 		Help:      "State (state trie, storage tries, and code) processing time",
 	})
 	tTxCommit = promauto.NewHistogram(prometheus.HistogramOpts{
 		Namespace: namespace,
 		Subsystem: statsSubsystem,
-		Name:      "t_postgres_tx_commit",
+		Name:      T_POSTGRES_TX_COMMIT,
 		Help:      "Postgres tx commit time",
 	})
+
+	httpCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: subsystemHTTP,
+		Name:      "count",
+		Help:      "http request count",
+	})
+	httpDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: subsystemHTTP,
+		Name:      "duration",
+		Help:      "http request duration",
+	})
+	ipcCount = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: subsystemIPC,
+		Name:      "count",
+		Help:      "unix socket connection count",
+	})
 }
 
-// RegisterDBCollector create metric colletor for given connection
+// RegisterDBCollector create metric collector for given connection
 func RegisterDBCollector(name string, db *sqlx.DB) {
 	if metrics {
 		prometheus.Register(NewDBStatsCollector(name, db))
 	}
 }
 
+// IncQueuedRanges increments the number of queued range requests
+func IncQueuedRanges() {
+	if metrics {
+		queuedRanges.Inc()
+	}
+}
+
+// DecQueuedRanges decrements the number of queued range requests
+func DecQueuedRanges() {
+	if metrics {
+		queuedRanges.Dec()
+	}
+}
+
 // SetLastLoadedHeight sets last loaded height
 func SetLastLoadedHeight(height int64) {
 	if metrics {
@@ -107,13 +164,13 @@ func SetTimeMetric(name string, t time.Duration) {
 	}
 	tAsF64 := t.Seconds()
 	switch name {
-	case "t_block_load":
+	case T_BLOCK_LOAD:
 		tBlockLoad.Observe(tAsF64)
-	case "t_block_processing":
+	case T_BLOCK_PROCESSING:
 		tBlockProcessing.Observe(tAsF64)
-	case "t_state_processing":
+	case T_STATE_PROCESSING:
 		tStateProcessing.Observe(tAsF64)
-	case "t_postgres_tx_commit":
+	case T_POSTGRES_TX_COMMIT:
 		tTxCommit.Observe(tAsF64)
 	}
 }

pkg/reader.go

@@ -28,23 +28,32 @@ import (
 	"github.com/ethereum/go-ethereum/trie"
 )
 
-// LvlDBReader exposes the necessary read functions on lvldb
+// Reader interface required by the statediffing service
+type Reader interface {
+	GetBlockByHash(hash common.Hash) (*types.Block, error)
+	GetBlockByNumber(number uint64) (*types.Block, error)
+	GetReceiptsByHash(hash common.Hash) (types.Receipts, error)
+	GetTdByHash(hash common.Hash) (*big.Int, error)
+	StateDB() state.Database
+}
+
+// LvlDBReader exposes the necessary Reader methods on lvldb
 type LvlDBReader struct {
 	ethDB       ethdb.Database
 	stateDB     state.Database
 	chainConfig *params.ChainConfig
 }
 
-// ReaderConfig struct for initializing a LvlDBReader
-type ReaderConfig struct {
+// LvLDBReaderConfig struct for initializing a LvlDBReader
+type LvLDBReaderConfig struct {
 	TrieConfig        *trie.Config
 	ChainConfig       *params.ChainConfig
 	Path, AncientPath string
 	DBCacheSize       int
 }
 
-// NewLvlDBReader creates a new LvlDBReader
-func NewLvlDBReader(conf ReaderConfig) (*LvlDBReader, error) {
+// NewLvlDBReader creates a new Read using LevelDB
+func NewLvlDBReader(conf LvLDBReaderConfig) (*LvlDBReader, error) {
 	edb, err := rawdb.NewLevelDBDatabaseWithFreezer(conf.Path, conf.DBCacheSize, 256, conf.AncientPath, "eth-statediff-service", true)
 	if err != nil {
 		return nil, err

pkg/rpc/check.go

@@ -0,0 +1,37 @@
+// VulcanizeDB
+// Copyright © 2020 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program 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 Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package rpc
+
+import "github.com/ethereum/go-ethereum/rpc"
+
+// checkModuleAvailability check that all names given in modules are actually
+// available API services.
+func checkModuleAvailability(modules []string, apis []rpc.API) (bad, available []string) {
+	availableSet := make(map[string]struct{})
+	for _, api := range apis {
+		if _, ok := availableSet[api.Namespace]; !ok {
+			availableSet[api.Namespace] = struct{}{}
+			available = append(available, api.Namespace)
+		}
+	}
+	for _, name := range modules {
+		if _, ok := availableSet[name]; !ok {
+			bad = append(bad, name)
+		}
+	}
+	return bad, available
+}

pkg/rpc/http.go

@@ -0,0 +1,49 @@
+// VulcanizeDB
+// Copyright © 2020 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program 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 Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package rpc
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/cmd/utils"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/rpc"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/eth-statediff-service/pkg/prom"
+)
+
+// StartHTTPEndpoint starts the HTTP RPC endpoint, configured with cors/vhosts/modules.
+func StartHTTPEndpoint(endpoint string, apis []rpc.API, modules []string, cors []string, vhosts []string, timeouts rpc.HTTPTimeouts) (*rpc.Server, error) {
+
+	srv := rpc.NewServer()
+	err := node.RegisterApis(apis, modules, srv, false)
+	if err != nil {
+		utils.Fatalf("Could not register HTTP API: %w", err)
+	}
+	handler := node.NewHTTPHandlerStack(srv, cors, vhosts)
+
+	// start http server
+	_, addr, err := node.StartHTTPEndpoint(endpoint, rpc.DefaultHTTPTimeouts, prom.HTTPMiddleware(handler))
+	if err != nil {
+		utils.Fatalf("Could not start RPC api: %v", err)
+	}
+	extapiURL := fmt.Sprintf("http://%v/", addr)
+	log.Infof("HTTP endpoint opened %s", extapiURL)
+
+	return srv, err
+}

pkg/rpc/ipc.go

@@ -0,0 +1,92 @@
+// VulcanizeDB
+// Copyright © 2020 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program 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 Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package rpc
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+
+	"github.com/ethereum/go-ethereum/p2p/netutil"
+	"github.com/ethereum/go-ethereum/rpc"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/eth-statediff-service/pkg/prom"
+)
+
+var (
+	//  On Linux, sun_path is 108 bytes in size
+	// see http://man7.org/linux/man-pages/man7/unix.7.html
+	maxPathSize = 108
+)
+
+// ipcListen will create a Unix socket on the given endpoint.
+func ipcListen(endpoint string) (net.Listener, error) {
+	if len(endpoint) > int(maxPathSize) {
+		log.Warn(fmt.Sprintf("The ipc endpoint is longer than %d characters. ", maxPathSize),
+			"endpoint", endpoint)
+	}
+
+	// Ensure the IPC path exists and remove any previous leftover
+	if err := os.MkdirAll(filepath.Dir(endpoint), 0751); err != nil {
+		return nil, err
+	}
+	os.Remove(endpoint)
+	l, err := net.Listen("unix", endpoint)
+	if err != nil {
+		return nil, err
+	}
+	os.Chmod(endpoint, 0600)
+	return l, nil
+}
+
+func ipcServe(srv *rpc.Server, listener net.Listener) {
+	for {
+		conn, err := listener.Accept()
+		if netutil.IsTemporaryError(err) {
+			log.WithError(err).Warn("rpc accept error")
+			continue
+		}
+		if err != nil {
+			log.WithError(err).Warn("unknown error")
+			continue
+		}
+		log.WithField("addr", conn.RemoteAddr()).Trace("accepted ipc connection")
+		go prom.IPCMiddleware(srv, conn)
+	}
+}
+
+// StartIPCEndpoint starts an IPC endpoint.
+func StartIPCEndpoint(ipcEndpoint string, apis []rpc.API) (net.Listener, *rpc.Server, error) {
+	// Register all the APIs exposed by the services.
+	handler := rpc.NewServer()
+	for _, api := range apis {
+		if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
+			return nil, nil, err
+		}
+		log.Debug("IPC registered", "namespace", api.Namespace)
+	}
+	// All APIs registered, start the IPC listener.
+	listener, err := ipcListen(ipcEndpoint)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	go ipcServe(handler, listener)
+	return listener, handler, nil
+}

pkg/service.go

@@ -23,7 +23,6 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
@@ -37,34 +36,31 @@ import (
 	ind "github.com/ethereum/go-ethereum/statediff/indexer"
 )
 
-// lvlDBReader are the db interfaces required by the statediffing service
-type lvlDBReader interface {
-	GetBlockByHash(hash common.Hash) (*types.Block, error)
-	GetBlockByNumber(number uint64) (*types.Block, error)
-	GetReceiptsByHash(hash common.Hash) (types.Receipts, error)
-	GetTdByHash(hash common.Hash) (*big.Int, error)
-	StateDB() state.Database
-}
+const defaultQueueSize = 1024
 
-// IService is the state-diffing service interface
-type IService interface {
-	// Start() and Stop()
+// StateDiffService is the state-diffing service interface
+type StateDiffService interface {
+	// Lifecycle Start() and Stop()
 	node.Lifecycle
-	// For node service registration
+	// APIs and Protocols() interface for node service registration
 	APIs() []rpc.API
 	Protocols() []p2p.Protocol
-	// Main event loop for processing state diffs
-	Loop(wg *sync.WaitGroup)
-	// Method to get state diff object at specific block
+	// Loop is the main event loop for processing state diffs
+	Loop(wg *sync.WaitGroup) error
+	// Run is a one-off command to run on a predefined set of ranges
+	Run(ranges []RangeRequest) error
+	// StateDiffAt method to get state diff object at specific block
 	StateDiffAt(blockNumber uint64, params sd.Params) (*sd.Payload, error)
-	// Method to get state diff object at specific block
+	// StateDiffFor method to get state diff object at specific block
 	StateDiffFor(blockHash common.Hash, params sd.Params) (*sd.Payload, error)
-	// Method to get state trie object at specific block
+	// StateTrieAt method to get state trie object at specific block
 	StateTrieAt(blockNumber uint64, params sd.Params) (*sd.Payload, error)
-	// Method to write state diff object directly to DB
+	// WriteStateDiffAt method to write state diff object directly to DB
 	WriteStateDiffAt(blockNumber uint64, params sd.Params) error
-	// Method to get state trie object at specific block
+	// WriteStateDiffFor method to get state trie object at specific block
 	WriteStateDiffFor(blockHash common.Hash, params sd.Params) error
+	// WriteStateDiffsInRange method to wrtie state diff objects within the range directly to the DB
+	WriteStateDiffsInRange(start, stop uint64, params sd.Params) error
 }
 
 // Service is the underlying struct for the state diffing service
@@ -72,24 +68,35 @@ type Service struct {
 	// Used to build the state diff objects
 	Builder Builder
 	// Used to read data from leveldb
-	lvlDBReader lvlDBReader
+	lvlDBReader Reader
 	// Used to signal shutdown of the service
-	QuitChan chan bool
+	quitChan chan struct{}
 	// Interface for publishing statediffs as PG-IPLD objects
 	indexer ind.Indexer
+	// range queue
+	queue chan RangeRequest
+	// number of ranges we can work over concurrently
+	workers uint
+	// ranges configured locally
+	preruns []RangeRequest
 }
 
 // NewStateDiffService creates a new Service
-func NewStateDiffService(lvlDBReader lvlDBReader, indexer ind.Indexer, workers uint) (*Service, error) {
-	builder, err := NewBuilder(lvlDBReader.StateDB(), workers)
+func NewStateDiffService(lvlDBReader Reader, indexer ind.Indexer, conf Config) (*Service, error) {
+	builder, err := NewBuilder(lvlDBReader.StateDB(), conf.TrieWorkers)
 	if err != nil {
 		return nil, err
 	}
+	if conf.WorkerQueueSize == 0 {
+		conf.WorkerQueueSize = defaultQueueSize
+	}
 	return &Service{
 		lvlDBReader: lvlDBReader,
 		Builder:     builder,
-		QuitChan:    make(chan bool),
 		indexer:     indexer,
+		workers:     conf.ServiceWorkers,
+		queue:       make(chan RangeRequest, conf.WorkerQueueSize),
+		preruns:     conf.PreRuns,
 	}, nil
 }
 
@@ -110,17 +117,73 @@ func (sds *Service) APIs() []rpc.API {
 	}
 }
 
-// Loop is an empty service loop for awaiting rpc requests
-func (sds *Service) Loop(wg *sync.WaitGroup) {
-	wg.Add(1)
-	for {
-		select {
-		case <-sds.QuitChan:
-			logrus.Info("closing the statediff service loop")
-			wg.Done()
-			return
+// Run does a one-off processing run on the provided RangeRequests + any pre-runs, exiting afterwards
+func (sds *Service) Run(rngs []RangeRequest) error {
+	for _, preRun := range sds.preruns {
+		logrus.Infof("processing prerun range (%d, %d)", preRun.Start, preRun.Stop)
+		for i := preRun.Start; i <= preRun.Stop; i++ {
+			if err := sds.WriteStateDiffAt(i, preRun.Params); err != nil {
+				return fmt.Errorf("error writing statediff at height %d in range (%d, %d) : %v", i, preRun.Start, preRun.Stop, err)
+			}
 		}
 	}
+	sds.preruns = nil
+	for _, rng := range rngs {
+		logrus.Infof("processing prerun range (%d, %d)", rng.Start, rng.Stop)
+		for i := rng.Start; i <= rng.Stop; i++ {
+			if err := sds.WriteStateDiffAt(i, rng.Params); err != nil {
+				return fmt.Errorf("error writing statediff at height %d in range (%d, %d) : %v", i, rng.Start, rng.Stop, err)
+			}
+		}
+	}
+	return nil
+}
+
+// Loop is an empty service loop for awaiting rpc requests
+func (sds *Service) Loop(wg *sync.WaitGroup) error {
+	if sds.quitChan != nil {
+		return fmt.Errorf("service loop is already running")
+	}
+
+	sds.quitChan = make(chan struct{})
+	for i := 0; i < int(sds.workers); i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			for {
+				select {
+				case blockRange := <-sds.queue:
+					logrus.Infof("service worker %d received range (%d, %d) off of work queue, beginning processing", id, blockRange.Start, blockRange.Stop)
+					prom.DecQueuedRanges()
+					for j := blockRange.Start; j <= blockRange.Stop; j++ {
+						if err := sds.WriteStateDiffAt(j, blockRange.Params); err != nil {
+							logrus.Errorf("service worker %d error writing statediff at height %d in range (%d, %d) : %v", id, j, blockRange.Start, blockRange.Stop, err)
+						}
+						select {
+						case <-sds.quitChan:
+							logrus.Infof("closing service worker %d\n"+
+								"working in range (%d, %d)\n"+
+								"last processed height: %d", id, blockRange.Start, blockRange.Stop, j)
+							return
+						default:
+							logrus.Infof("service worker %d finished processing statediff height %d in range (%d, %d)", id, j, blockRange.Start, blockRange.Stop)
+						}
+					}
+					logrus.Infof("service worker %d finished processing range (%d, %d)", id, blockRange.Start, blockRange.Stop)
+				case <-sds.quitChan:
+					logrus.Infof("closing the statediff service loop worker %d", id)
+					return
+				}
+			}
+		}(i)
+	}
+	for _, preRun := range sds.preruns {
+		if err := sds.WriteStateDiffsInRange(preRun.Start, preRun.Stop, preRun.Params); err != nil {
+			close(sds.quitChan)
+			return err
+		}
+	}
+	return nil
 }
 
 // StateDiffAt returns a state diff object payload at the specific blockheight
@@ -237,14 +300,13 @@ func (sds *Service) processStateTrie(block *types.Block, params sd.Params) (*sd.
 // Start is used to begin the service
 func (sds *Service) Start() error {
 	logrus.Info("starting statediff service")
-	go sds.Loop(new(sync.WaitGroup))
-	return nil
+	return sds.Loop(new(sync.WaitGroup))
 }
 
 // Stop is used to close down the service
 func (sds *Service) Stop() error {
 	logrus.Info("stopping statediff service")
-	close(sds.QuitChan)
+	close(sds.quitChan)
 	return nil
 }
 
@@ -309,7 +371,7 @@ func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, p
 	}
 	height := block.Number().Int64()
 	prom.SetLastLoadedHeight(height)
-	prom.SetTimeMetric("t_block_load", time.Now().Sub(t))
+	prom.SetTimeMetric(prom.T_BLOCK_LOAD, time.Now().Sub(t))
 	t = time.Now()
 	tx, err := sds.indexer.PushBlock(block, receipts, totalDifficulty)
 	if err != nil {
@@ -322,16 +384,32 @@ func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, p
 	codeOutput := func(c sdtypes.CodeAndCodeHash) error {
 		return sds.indexer.PushCodeAndCodeHash(tx, c)
 	}
-	prom.SetTimeMetric("t_block_processing", time.Now().Sub(t))
+	prom.SetTimeMetric(prom.T_BLOCK_PROCESSING, time.Now().Sub(t))
 	t = time.Now()
 	err = sds.Builder.WriteStateDiffObject(sd.StateRoots{
 		NewStateRoot: block.Root(),
 		OldStateRoot: parentRoot,
 	}, params, output, codeOutput)
-	prom.SetTimeMetric("t_state_processing", time.Now().Sub(t))
+	prom.SetTimeMetric(prom.T_STATE_PROCESSING, time.Now().Sub(t))
 	t = time.Now()
 	err = tx.Close(err)
 	prom.SetLastProcessedHeight(height)
-	prom.SetTimeMetric("t_postgres_tx_commit", time.Now().Sub(t))
+	prom.SetTimeMetric(prom.T_POSTGRES_TX_COMMIT, time.Now().Sub(t))
 	return err
 }
+
+// WriteStateDiffsInRange adds a RangeRequest to the work queue
+func (sds *Service) WriteStateDiffsInRange(start, stop uint64, params sd.Params) error {
+	if stop < start {
+		return fmt.Errorf("invalid block range (%d, %d): stop height must be greater or equal to start height", start, stop)
+	}
+	blocked := time.NewTimer(30 * time.Second)
+	select {
+	case sds.queue <- RangeRequest{Start: start, Stop: stop, Params: params}:
+		prom.IncQueuedRanges()
+		logrus.Infof("added range (%d, %d) to the worker queue", start, stop)
+		return nil
+	case <-blocked.C:
+		return fmt.Errorf("unable to add range (%d, %d) to the worker queue", start, stop)
+	}
+}

pkg/types.go

@@ -21,7 +21,8 @@ package statediff
 
 import (
 	"github.com/ethereum/go-ethereum/core/types"
-	sd "github.com/ethereum/go-ethereum/statediff/types"
+	sd "github.com/ethereum/go-ethereum/statediff"
+	sdTypes "github.com/ethereum/go-ethereum/statediff/types"
 )
 
 // AccountMap is a mapping of hex encoded path => account wrapper
@@ -30,8 +31,14 @@ type AccountMap map[string]accountWrapper
 // accountWrapper is used to temporary associate the unpacked node with its raw values
 type accountWrapper struct {
 	Account   *types.StateAccount
-	NodeType  sd.NodeType
+	NodeType  sdTypes.NodeType
 	Path      []byte
 	NodeValue []byte
 	LeafKey   []byte
 }
+
+// RangeRequest holds range quest work params
+type RangeRequest struct {
+	Start, Stop uint64
+	Params      sd.Params
+}