Merge pull request #47 from vulcanize/prerun

cleanup

Dockerfile

@@ -6,6 +6,8 @@ RUN apk add busybox-extras
 
 # Get and build ipfs-blockchain-watcher
 ADD . /go/src/github.com/vulcanize/eth-statediff-service
+#RUN git clone https://github.com/vulcanize/eth-statediff-service.git /go/src/github.com/vulcanize/eth-statediff-service
+
 WORKDIR /go/src/github.com/vulcanize/eth-statediff-service
 RUN GO111MODULE=on GCO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -ldflags '-extldflags "-static"' -o eth-statediff-service .
 
@@ -15,7 +17,10 @@ FROM alpine
 ARG USER="vdm"
 ARG CONFIG_FILE="./environments/example.toml"
 ARG EXPOSE_PORT=8545
-RUN adduser -Du 5000 $USER
+
+RUN adduser -Du 5000 $USER adm
+RUN adduser $USER adm; apk --no-cache add sudo; echo '%adm ALL=(ALL) NOPASSWD:ALL' >> /etc/sudoers
+
 WORKDIR /app
 RUN chown $USER /app
 USER $USER

cmd/env.go

@@ -29,16 +29,38 @@ const (
 	ETH_NETWORK_ID    = "ETH_NETWORK_ID"
 	ETH_CHAIN_ID      = "ETH_CHAIN_ID"
 
-	DB_CACHE_SIZE_MB   = "DB_CACHE_SIZE_MB"
-	TRIE_CACHE_SIZE_MB = "TRIE_CACHE_SIZE_MB"
-	LVLDB_PATH         = "LVLDB_PATH"
-	LVLDB_ANCIENT      = "LVLDB_ANCIENT"
-	STATEDIFF_WORKERS  = "STATEDIFF_WORKERS"
-	WRITE_SERVER       = "WRITE_SERVER"
+	DB_CACHE_SIZE_MB            = "DB_CACHE_SIZE_MB"
+	TRIE_CACHE_SIZE_MB          = "TRIE_CACHE_SIZE_MB"
+	LVLDB_PATH                  = "LVLDB_PATH"
+	LVLDB_ANCIENT               = "LVLDB_ANCIENT"
+
+	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)
@@ -60,5 +82,21 @@ func init() {
 	viper.BindEnv("leveldb.path", LVLDB_PATH)
 	viper.BindEnv("leveldb.ancient", LVLDB_ANCIENT)
 
-	viper.BindEnv("statediff.workers", STATEDIFF_WORKERS)
+	viper.BindEnv("prom.metrics", PROM_METRICS)
+	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.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

@@ -26,6 +26,8 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+
+	"github.com/vulcanize/eth-statediff-service/pkg/prom"
 )
 
 var (
@@ -64,10 +66,24 @@ func initFuncs(cmd *cobra.Command, args []string) {
 	if err := logLevel(); err != nil {
 		log.Fatal("Could not set log level: ", err)
 	}
+
+	if viper.GetBool("prom.metrics") {
+		log.Info("initializing prometheus metrics")
+		prom.Init()
+	}
+
+	if viper.GetBool("prom.http") {
+		addr := fmt.Sprintf(
+			"%s:%s",
+			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
@@ -85,6 +101,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(),
@@ -92,7 +110,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")
@@ -106,12 +128,28 @@ func init() {
 		"0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3", "eth genesis block hash")
 	rootCmd.PersistentFlags().String("eth-network-id", "1", "eth network id")
 	rootCmd.PersistentFlags().String("eth-chain-id", "1", "eth chain id")
+
 	rootCmd.PersistentFlags().Int("cache-db", 1024, "megabytes of memory allocated to database cache")
 	rootCmd.PersistentFlags().Int("cache-trie", 1024, "Megabytes of memory allocated to trie cache")
 
+	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("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"))
@@ -126,6 +164,14 @@ func init() {
 	viper.BindPFlag("ethereum.chainID", rootCmd.PersistentFlags().Lookup("eth-chain-id"))
 	viper.BindPFlag("cache.database", rootCmd.PersistentFlags().Lookup("cache-db"))
 	viper.BindPFlag("cache.trie", rootCmd.PersistentFlags().Lookup("cache-trie"))
+	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.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"))
 }
 
 func initConfig() {

cmd/serve.go

@@ -16,29 +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/ethereum/go-ethereum/trie"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
-	ind "github.com/ethereum/go-ethereum/statediff/indexer"
-	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
-
 	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}`,
@@ -49,61 +43,32 @@ var serveCmd = &cobra.Command{
 	},
 }
 
+func init() {
+	rootCmd.AddCommand(serveCmd)
+}
+
 func serve() {
-	logWithCommand.Info("starting statediff RPC service")
+	logWithCommand.Info("Running eth-statediff-service serve command")
 
-	// load params
-	path := viper.GetString("leveldb.path")
-	ancientPath := viper.GetString("leveldb.ancient")
-	if path == "" || ancientPath == "" {
-		logWithCommand.Fatal("require a valid eth leveldb primary datastore path and ancient datastore path")
-	}
-
-	nodeInfo := GetEthNodeInfo()
-	config, err := chainConfig(nodeInfo.ChainID)
+	statediffService, err := createStateDiffService()
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
-	// create leveldb reader
-	logWithCommand.Info("Creating leveldb reader")
-	conf := sd.ReaderConfig{
-		TrieConfig: &trie.Config{
-			Cache:     viper.GetInt("cache.trie"),
-			Journal:   "",
-			Preimages: false,
-		},
-		ChainConfig: config,
-		Path:        path,
-		AncientPath: ancientPath,
-		DBCacheSize: viper.GetInt("cache.database"),
-	}
-	lvlDBReader, err := sd.NewLvlDBReader(conf)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	// create statediff service
-	logWithCommand.Info("Creating statediff service")
-	db, err := postgres.NewDB(postgres.DbConnectionString(GetDBParams()), GetDBConfig(), nodeInfo)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	indexer, err := ind.NewStateDiffIndexer(config, db)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	statediffService, err := sd.NewStateDiffService(lvlDBReader, indexer, viper.GetUint("statediff.workers"))
-	if err != nil {
-		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)
@@ -119,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 == "" {
@@ -139,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

@@ -0,0 +1,151 @@
+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"
+	"github.com/ethereum/go-ethereum/trie"
+	"github.com/spf13/viper"
+
+	sd "github.com/vulcanize/eth-statediff-service/pkg"
+	"github.com/vulcanize/eth-statediff-service/pkg/prom"
+)
+
+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")
+	if path == "" || ancientPath == "" {
+		logWithCommand.Fatal("require a valid eth leveldb primary datastore path and ancient datastore path")
+	}
+
+	nodeInfo := GetEthNodeInfo()
+	chainConf, err := chainConfig(nodeInfo.ChainID)
+	if err != nil {
+		logWithCommand.Fatal(err)
+	}
+
+	// create leveldb reader
+	logWithCommand.Info("Creating leveldb reader")
+	readerConf := sd.LvLDBReaderConfig{
+		TrieConfig: &trie.Config{
+			Cache:     viper.GetInt("cache.trie"),
+			Journal:   "",
+			Preimages: false,
+		},
+		ChainConfig: chainConf,
+		Path:        path,
+		AncientPath: ancientPath,
+		DBCacheSize: viper.GetInt("cache.database"),
+	}
+	lvlDBReader, err := sd.NewLvlDBReader(readerConf)
+	if err != nil {
+		logWithCommand.Fatal(err)
+	}
+
+	// create statediff service
+	logWithCommand.Info("Setting up Postgres DB")
+	db, err := setupPostgres(nodeInfo)
+	if err != nil {
+		logWithCommand.Fatal(err)
+	}
+	logWithCommand.Info("Creating statediff indexer")
+	indexer, err := ind.NewStateDiffIndexer(chainConf, db)
+	if err != nil {
+		logWithCommand.Fatal(err)
+	}
+	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) {
+	p := GetDBParams()
+	logWithCommand.Info("initializing DB connection pool")
+	db, err := postgres.NewDB(postgres.DbConnectionString(p), GetDBConfig(), nodeInfo)
+	if err != nil {
+		return nil, err
+	}
+	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,189 +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"
-	ind "github.com/ethereum/go-ethereum/statediff/indexer"
-	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
-	"github.com/ethereum/go-ethereum/trie"
-	"github.com/sirupsen/logrus"
-	"github.com/spf13/cobra"
-	"github.com/spf13/viper"
-
-	sd "github.com/vulcanize/eth-statediff-service/pkg"
-)
-
-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("Starting statediff writer")
-	// load params
-	viper.BindEnv("write.serve", WRITE_SERVER)
-	addr := viper.GetString("write.serve")
-	path := viper.GetString("leveldb.path")
-	ancientPath := viper.GetString("leveldb.ancient")
-	if path == "" || ancientPath == "" {
-		logWithCommand.Fatal("require a valid eth leveldb primary datastore path and ancient datastore path")
-	}
-
-	nodeInfo := GetEthNodeInfo()
-	config, err := chainConfig(nodeInfo.ChainID)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	// create leveldb reader
-	logWithCommand.Info("Creating leveldb reader")
-	conf := sd.ReaderConfig{
-		TrieConfig: &trie.Config{
-			Cache:     viper.GetInt("cache.trie"),
-			Journal:   "",
-			Preimages: false,
-		},
-		ChainConfig: config,
-		Path:        path,
-		AncientPath: ancientPath,
-		DBCacheSize: viper.GetInt("cache.database"),
-	}
-	lvlDBReader, err := sd.NewLvlDBReader(conf)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	// create statediff service
-	logWithCommand.Info("Creating statediff service")
-	db, err := postgres.NewDB(postgres.DbConnectionString(GetDBParams()), GetDBConfig(), nodeInfo)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	indexer, err := ind.NewStateDiffIndexer(config, db)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	statediffService, err := sd.NewStateDiffService(lvlDBReader, indexer, viper.GetUint("statediff.workers"))
-	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 = ""
@@ -32,3 +44,10 @@
 [cache]
     database = 1024
     trie = 1024
+
+[prom]
+    dbStats = false
+    metrics = true
+    http = true
+    httpAddr = "localhost"
+    httpPort = "8889"

go.mod

@@ -4,11 +4,12 @@ go 1.13
 
 require (
 	github.com/ethereum/go-ethereum v1.10.9
+	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.10-statediff-0.0.27
+replace github.com/ethereum/go-ethereum v1.10.9 => github.com/vulcanize/go-ethereum v1.10.11-statediff-0.0.27

go.sum

@@ -62,6 +62,7 @@ github.com/aws/aws-sdk-go-v2/service/sso v1.1.1/go.mod h1:SuZJxklHxLAXgLTc1iFXbE
 github.com/aws/aws-sdk-go-v2/service/sts v1.1.1/go.mod h1:Wi0EBZwiz/K44YliU0EKxqTCJGUfYTWXrrBwkq736bM=
 github.com/aws/smithy-go v1.1.0/go.mod h1:EzMw8dbp/YJL4A5/sbhGddag+NPT7q084agLbB9LgIw=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
+github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
 github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84=
@@ -108,11 +109,13 @@ 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.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-20211011172007-d99e4b8cbf48/go.mod h1:R9ET47fwRVRPZnOGvHxxhuZcbrMCuiqOz3Rlrh4KSnk=
 github.com/dop251/goja_nodejs v0.0.0-20210225215109-d91c329300e7/go.mod h1:hn7BA7c8pLvoGndExHudxTDKZ84Pyvv+90pbBjbTz0Y=
@@ -209,6 +212,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=
@@ -253,11 +257,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=
@@ -306,6 +313,7 @@ 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=
@@ -364,6 +372,7 @@ github.com/mattn/go-sqlite3 v1.11.0/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsO
 github.com/mattn/go-sqlite3 v1.14.7 h1:fxWBnXkxfM6sRiuH3bqJ4CfzZojMOLVc0UTsTglEghA=
 github.com/mattn/go-sqlite3 v1.14.7/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
 github.com/mattn/go-tty v0.0.0-20180907095812-13ff1204f104/go.mod h1:XPvLUNfbS4fJH25nqRHfWLMa1ONC8Amw+mIA639KxkE=
+github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1 h1:lYpkrQH5ajf0OXOcUbGjvZxxijuBwbbmlSxLiuofa+g=
@@ -428,6 +437,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=
@@ -441,16 +451,20 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
+github.com/prometheus/client_golang v1.0.0 h1:vrDKnkGzuGvhNAL56c7DBz29ZL+KxnoR0x7enabFceM=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4 h1:gQz4mCbXsO+nc9n1hCxHcGA3Zx3Eo+UHZoInFGUIXNM=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
 github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
+github.com/prometheus/common v0.6.0 h1:kRhiuYSXR3+uv2IbVbZhUxK5zVD/2pp3Gd2PpvPkpEo=
 github.com/prometheus/common v0.6.0/go.mod h1:eBmuwkDJBwy6iBfxCBob6t6dR6ENT/y+J+Zk0j9GMYc=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
+github.com/prometheus/procfs v0.0.2 h1:6LJUbpNm42llc4HRCuvApCSWB/WfhuNo9K98Q9sNGfs=
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/tsdb v0.7.1 h1:YZcsG11NqnK4czYLrWd9mpEuAJIHVQLwdrleYfszMAA=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
@@ -525,8 +539,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.10-statediff-0.0.27 h1:Oe6iSB8Qt1SkcECz7hVoRXByM1XGNnWUQw7KvA1+n+g=
-github.com/vulcanize/go-ethereum v1.10.10-statediff-0.0.27/go.mod h1:9L+QY31AnWnX2/2HDOySCjQoYUdWNGBRMezFJVfH73E=
+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/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=
@@ -785,6 +799,7 @@ gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce h1:+JknDZhAj8YMt7GC73Ei8pv4MzjDUNPHgQWJdtMAaDU=
 gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce/go.mod h1:5AcXVHNjg+BDxry382+8OKon8SEWiKktQR07RKPsv1c=
+gopkg.in/olebedev/go-duktape.v3 v3.0.0-20200619000410-60c24ae608a6 h1:a6cXbcDDUkSBlpnkWV1bJ+vv3mOgQEltEJ2rPxroVu0=
 gopkg.in/olebedev/go-duktape.v3 v3.0.0-20200619000410-60c24ae608a6/go.mod h1:uAJfkITjFhyEEuUfm7bsmCZRbW5WRq8s9EY8HZ6hCns=
 gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
@@ -802,6 +817,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/prometheus.yml

@@ -0,0 +1,7 @@
+global:
+  scrape_interval: 10s
+
+scrape_configs:
+  - job_name: 'eth-statediff-service'
+    static_configs:
+      - targets: ['localhost:8100']

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

@@ -0,0 +1,162 @@
+// VulcanizeDB
+// Copyright © 2021 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 (
+	"database/sql"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	namespace = "eth_statediff_service"
+	subsystem = "connections"
+)
+
+// DBStatsGetter is an interface that gets sql.DBStats.
+type DBStatsGetter interface {
+	Stats() sql.DBStats
+}
+
+// DBStatsCollector implements the prometheus.Collector interface.
+type DBStatsCollector struct {
+	sg DBStatsGetter
+
+	// descriptions of exported metrics
+	maxOpenDesc           *prometheus.Desc
+	openDesc              *prometheus.Desc
+	inUseDesc             *prometheus.Desc
+	idleDesc              *prometheus.Desc
+	waitedForDesc         *prometheus.Desc
+	blockedSecondsDesc    *prometheus.Desc
+	closedMaxIdleDesc     *prometheus.Desc
+	closedMaxLifetimeDesc *prometheus.Desc
+}
+
+// NewDBStatsCollector creates a new DBStatsCollector.
+func NewDBStatsCollector(dbName string, sg DBStatsGetter) *DBStatsCollector {
+	labels := prometheus.Labels{"db_name": dbName}
+	return &DBStatsCollector{
+		sg: sg,
+		maxOpenDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, subsystem, "max_open"),
+			"Maximum number of open connections to the database.",
+			nil,
+			labels,
+		),
+		openDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, subsystem, "open"),
+			"The number of established connections both in use and idle.",
+			nil,
+			labels,
+		),
+		inUseDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, subsystem, "in_use"),
+			"The number of connections currently in use.",
+			nil,
+			labels,
+		),
+		idleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, subsystem, "idle"),
+			"The number of idle connections.",
+			nil,
+			labels,
+		),
+		waitedForDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, subsystem, "waited_for"),
+			"The total number of connections waited for.",
+			nil,
+			labels,
+		),
+		blockedSecondsDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, subsystem, "blocked_seconds"),
+			"The total time blocked waiting for a new connection.",
+			nil,
+			labels,
+		),
+		closedMaxIdleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, subsystem, "closed_max_idle"),
+			"The total number of connections closed due to SetMaxIdleConns.",
+			nil,
+			labels,
+		),
+		closedMaxLifetimeDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, subsystem, "closed_max_lifetime"),
+			"The total number of connections closed due to SetConnMaxLifetime.",
+			nil,
+			labels,
+		),
+	}
+}
+
+// Describe implements the prometheus.Collector interface.
+func (c DBStatsCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.maxOpenDesc
+	ch <- c.openDesc
+	ch <- c.inUseDesc
+	ch <- c.idleDesc
+	ch <- c.waitedForDesc
+	ch <- c.blockedSecondsDesc
+	ch <- c.closedMaxIdleDesc
+	ch <- c.closedMaxLifetimeDesc
+}
+
+// Collect implements the prometheus.Collector interface.
+func (c DBStatsCollector) Collect(ch chan<- prometheus.Metric) {
+	stats := c.sg.Stats()
+
+	ch <- prometheus.MustNewConstMetric(
+		c.maxOpenDesc,
+		prometheus.GaugeValue,
+		float64(stats.MaxOpenConnections),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.openDesc,
+		prometheus.GaugeValue,
+		float64(stats.OpenConnections),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.inUseDesc,
+		prometheus.GaugeValue,
+		float64(stats.InUse),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.idleDesc,
+		prometheus.GaugeValue,
+		float64(stats.Idle),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.waitedForDesc,
+		prometheus.CounterValue,
+		float64(stats.WaitCount),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.blockedSecondsDesc,
+		prometheus.CounterValue,
+		stats.WaitDuration.Seconds(),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.closedMaxIdleDesc,
+		prometheus.CounterValue,
+		float64(stats.MaxIdleClosed),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.closedMaxLifetimeDesc,
+		prometheus.CounterValue,
+		float64(stats.MaxLifetimeClosed),
+	)
+}

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

@@ -0,0 +1,176 @@
+// VulcanizeDB
+// Copyright © 2021 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 (
+	"time"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const (
+	statsSubsystem = "stats"
+	subsystemHTTP  = "http"
+	subsystemIPC   = "ipc"
+)
+
+var (
+	metrics bool
+
+	queuedRanges        prometheus.Gauge
+	lastLoadedHeight    prometheus.Gauge
+	lastProcessedHeight prometheus.Gauge
+
+	tBlockLoad       prometheus.Histogram
+	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,
+		Help:      "The last block that was loaded for processing",
+	})
+	lastProcessedHeight = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Name:      PROCESSED_HEIGHT,
+		Help:      "The last block that was processed",
+	})
+
+	tBlockLoad = promauto.NewHistogram(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		Name:      T_BLOCK_LOAD,
+		Help:      "Block loading time",
+	})
+	tBlockProcessing = promauto.NewHistogram(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		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,
+		Help:      "State (state trie, storage tries, and code) processing time",
+	})
+	tTxCommit = promauto.NewHistogram(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		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 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 {
+		lastLoadedHeight.Set(float64(height))
+	}
+}
+
+// SetLastProcessedHeight sets last processed height
+func SetLastProcessedHeight(height int64) {
+	if metrics {
+		lastProcessedHeight.Set(float64(height))
+	}
+}
+
+// SetTimeMetric time metric observation
+func SetTimeMetric(name string, t time.Duration) {
+	if !metrics {
+		return
+	}
+	tAsF64 := t.Seconds()
+	switch name {
+	case T_BLOCK_LOAD:
+		tBlockLoad.Observe(tAsF64)
+	case T_BLOCK_PROCESSING:
+		tBlockProcessing.Observe(tAsF64)
+	case T_STATE_PROCESSING:
+		tStateProcessing.Observe(tAsF64)
+	case T_POSTGRES_TX_COMMIT:
+		tTxCommit.Observe(tAsF64)
+	}
+}

pkg/prom/web.go

@@ -0,0 +1,40 @@
+// VulcanizeDB
+// Copyright © 2021 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"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/sirupsen/logrus"
+)
+
+// Listen start listening http
+func Listen(addr string) *http.Server {
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+	srv := http.Server{
+		Addr:    addr,
+		Handler: mux,
+	}
+	go func() {
+		if err := srv.ListenAndServe(); err != nil {
+			logrus.Fatal(err)
+		}
+	}()
+	return &srv
+}

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

@@ -20,9 +20,9 @@ import (
 	"fmt"
 	"math/big"
 	"sync"
+	"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"
@@ -31,38 +31,36 @@ import (
 	sd "github.com/ethereum/go-ethereum/statediff"
 	sdtypes "github.com/ethereum/go-ethereum/statediff/types"
 	"github.com/sirupsen/logrus"
+	"github.com/vulcanize/eth-statediff-service/pkg/prom"
 
 	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
@@ -70,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
 }
 
@@ -108,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
@@ -235,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
 }
 
@@ -251,6 +315,7 @@ func (sds *Service) Stop() error {
 // for historical data
 func (sds *Service) WriteStateDiffAt(blockNumber uint64, params sd.Params) error {
 	logrus.Info(fmt.Sprintf("Writing state diff at block %d", blockNumber))
+	t := time.Now()
 	currentBlock, err := sds.lvlDBReader.GetBlockByNumber(blockNumber)
 	if err != nil {
 		return err
@@ -263,7 +328,7 @@ func (sds *Service) WriteStateDiffAt(blockNumber uint64, params sd.Params) error
 		}
 		parentRoot = parentBlock.Root()
 	}
-	return sds.writeStateDiff(currentBlock, parentRoot, params)
+	return sds.writeStateDiff(currentBlock, parentRoot, params, t)
 }
 
 // WriteStateDiffFor writes a state diff for the specific blockHash directly to the database
@@ -271,6 +336,7 @@ func (sds *Service) WriteStateDiffAt(blockNumber uint64, params sd.Params) error
 // for historical data
 func (sds *Service) WriteStateDiffFor(blockHash common.Hash, params sd.Params) error {
 	logrus.Info(fmt.Sprintf("Writing state diff for block %s", blockHash.Hex()))
+	t := time.Now()
 	currentBlock, err := sds.lvlDBReader.GetBlockByHash(blockHash)
 	if err != nil {
 		return err
@@ -283,11 +349,11 @@ func (sds *Service) WriteStateDiffFor(blockHash common.Hash, params sd.Params) e
 		}
 		parentRoot = parentBlock.Root()
 	}
-	return sds.writeStateDiff(currentBlock, parentRoot, params)
+	return sds.writeStateDiff(currentBlock, parentRoot, params, t)
 }
 
 // Writes a state diff from the current block, parent state root, and provided params
-func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, params sd.Params) error {
+func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, params sd.Params, t time.Time) error {
 	var totalDifficulty *big.Int
 	var receipts types.Receipts
 	var err error
@@ -303,7 +369,10 @@ func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, p
 	if err != nil {
 		return err
 	}
-
+	height := block.Number().Int64()
+	prom.SetLastLoadedHeight(height)
+	prom.SetTimeMetric(prom.T_BLOCK_LOAD, time.Now().Sub(t))
+	t = time.Now()
 	tx, err := sds.indexer.PushBlock(block, receipts, totalDifficulty)
 	if err != nil {
 		return err
@@ -315,9 +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(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)
-	return tx.Close(err)
+	prom.SetTimeMetric(prom.T_STATE_PROCESSING, time.Now().Sub(t))
+	t = time.Now()
+	err = tx.Close(err)
+	prom.SetLastProcessedHeight(height)
+	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
+}

startup_script.sh

@@ -4,8 +4,17 @@ set -e
 set +x
 
 # Check the database variables are set
-test $VDB_COMMAND
+test "$VDB_COMMAND"
 set +e
 
-echo "Running the statediff service"
-./eth-statediff-service ${VDB_COMMAND} --config=config.toml
+# docker must be run in privilaged mode for mounts to work
+echo "Setting up /app/geth-rw overlayed /app/geth-ro"
+mkdir -p /tmp/overlay && \
+sudo mount -t tmpfs tmpfs /tmp/overlay && \
+mkdir -p /tmp/overlay/upper && \
+mkdir -p /tmp/overlay/work && \
+mkdir -p /app/geth-rw && \
+sudo mount -t overlay overlay -o lowerdir=/app/geth-ro,upperdir=/tmp/overlay/upper,workdir=/tmp/overlay/work /app/geth-rw && \
+
+echo "Running the statediff service" && \
+sudo ./eth-statediff-service "$VDB_COMMAND" --config=config.toml