Merge pull request #77 from vulcanize/release-v2.0.0

update to use v1.10.15-statediff-2.0.0 geth

.github/workflows/publish.yaml

@@ -7,19 +7,23 @@ 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
-        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}}
+          echo ::set-output name=sha::$(echo ${GITHUB_SHA})
+        id: vars
       - 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}}
+        uses: docker/login-action@v1
+        with:
+          username: vulcanizejenkins
+          password: ${{ secrets.VULCANIZEJENKINS_PAT }}
+      - 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: |
+            GIT_COMMIT=${{ steps.vars.outputs.sha }}
 

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 .
 
@@ -13,9 +15,12 @@ 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
+
+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

@@ -28,10 +28,39 @@ const (
 	ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
 	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_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)
@@ -46,4 +75,28 @@ func init() {
 	viper.BindEnv("database.maxIdle", pg.DATABASE_MAX_IDLE_CONNECTIONS)
 	viper.BindEnv("database.maxOpen", pg.DATABASE_MAX_OPEN_CONNECTIONS)
 	viper.BindEnv("database.maxLifetime", pg.DATABASE_MAX_CONN_LIFETIME)
+
+	viper.BindEnv("cache.database", DB_CACHE_SIZE_MB)
+	viper.BindEnv("cache.trie", TRIE_CACHE_SIZE_MB)
+
+	viper.BindEnv("leveldb.path", LVLDB_PATH)
+	viper.BindEnv("leveldb.ancient", LVLDB_ANCIENT)
+
+	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

@@ -18,14 +18,18 @@ 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"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+
+	"github.com/vulcanize/eth-statediff-service/pkg/prom"
 )
 
 var (
@@ -64,10 +68,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 +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(),
@@ -92,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")
@@ -107,9 +131,27 @@ func init() {
 	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"))
@@ -122,6 +164,18 @@ func init() {
 	viper.BindPFlag("ethereum.genesisBlock", rootCmd.PersistentFlags().Lookup("eth-genesis-block"))
 	viper.BindPFlag("ethereum.networkID", rootCmd.PersistentFlags().Lookup("eth-network-id"))
 	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"))
+
+	rand.Seed(time.Now().UnixNano())
 }
 
 func initConfig() {
@@ -138,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,27 +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"
 
-	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}`,
@@ -47,45 +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")
-	lvlDBReader, err := sd.NewLvlDBReader(path, ancientPath, config)
-	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 := ind.NewStateDiffIndexer(config, db)
-	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)
@@ -101,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 == "" {
@@ -121,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/version.go

@@ -0,0 +1,42 @@
+// 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 cmd
+
+import (
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+
+	v "github.com/vulcanize/eth-statediff-service/version"
+)
+
+// versionCmd represents the version command
+var versionCmd = &cobra.Command{
+	Use:   "version",
+	Short: "Prints the version of eth-statediff-service",
+	Long: `Use this command to fetch the version of eth-statediff-service
+
+Usage: ./eth-statediff-service  version`,
+	Run: func(cmd *cobra.Command, args []string) {
+		subCommand = cmd.CalledAs()
+		logWithCommand = *log.WithField("SubCommand", subCommand)
+		logWithCommand.Infof("eth-statediff-service version: %s", v.Version)
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(versionCmd)
+}

cmd/write.go

@@ -1,103 +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 (
-	"github.com/sirupsen/logrus"
-	"github.com/spf13/cobra"
-	"github.com/spf13/viper"
-
-	gethsd "github.com/ethereum/go-ethereum/statediff"
-	ind "github.com/ethereum/go-ethereum/statediff/indexer"
-	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
-	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()
-	},
-}
-
-func init() {
-	rootCmd.AddCommand(writeCmd)
-}
-
-func write() {
-	logWithCommand.Info("Starting statediff writer")
-
-	// 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)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-
-	// create leveldb reader
-	logWithCommand.Info("Creating leveldb reader")
-	lvlDBReader, err := sd.NewLvlDBReader(path, ancientPath, config)
-	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 := ind.NewStateDiffIndexer(config, db)
-	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 [][2]uint64
-	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)
-
-	for _, rng := range blockRanges {
-		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++ {
-			statediffService.WriteStateDiffAt(height, diffParams)
-		}
-	}
-}

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,12 +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]
     ]
+    [prerun.params]
+        intermediateStateNodes = true
+        intermediateStorageNodes = true
+        includeBlock = true
+        includeReceipts = true
+        includeTD = true
+        includeCode = true
+        watchedAddresses = []
+        watchedStorageKeys = []
 
 [log]
     file = ""
@@ -27,3 +40,14 @@
     port     = 5432
     user     = "vulcanize"
     password = "..."
+
+[cache]
+    database = 1024
+    trie = 1024
+
+[prom]
+    dbStats = false
+    metrics = true
+    http = true
+    httpAddr = "localhost"
+    httpPort = "8889"

go.mod

@@ -3,15 +3,13 @@ module github.com/vulcanize/eth-statediff-service
 go 1.13
 
 require (
-	github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6 // indirect
-	github.com/ethereum/go-ethereum v1.10.1
-	github.com/konsorten/go-windows-terminal-sequences v1.0.3 // indirect
+	github.com/ethereum/go-ethereum v1.10.15
+	github.com/jmoiron/sqlx v1.2.0
+	github.com/prometheus/client_golang v1.4.0
 	github.com/sirupsen/logrus v1.7.0
-	github.com/spf13/cobra v1.1.1
-	github.com/spf13/viper v1.7.1
-	github.com/ugorji/go v1.1.4 // indirect
-	github.com/vulcanize/go-eth-state-node-iterator v0.0.1-alpha
-	github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77 // indirect
+	github.com/spf13/cobra v1.3.0
+	github.com/spf13/viper v1.10.0
+	github.com/vulcanize/go-eth-state-node-iterator v0.0.1-alpha.0.20211014064906-d23d01ed8191
 )
 
-replace github.com/ethereum/go-ethereum v1.10.1 => github.com/vulcanize/go-ethereum v1.10.1-statediff-0.0.16
+replace github.com/ethereum/go-ethereum v1.10.15 => github.com/vulcanize/go-ethereum v1.10.15-statediff-2.0.0

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
+}

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

@@ -31,10 +31,12 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rlp"
-	"github.com/ethereum/go-ethereum/trie"
-
 	sd "github.com/ethereum/go-ethereum/statediff"
+	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"
 )
 
@@ -72,7 +74,7 @@ func resolveNode(it trie.NodeIterator, trieDB *trie.Database) (sdtypes.StateNode
 	if err := rlp.DecodeBytes(node, &nodeElements); err != nil {
 		return sdtypes.StateNode{}, nil, err
 	}
-	ty, err := sd.CheckKeyType(nodeElements)
+	ty, err := sdtrie.CheckKeyType(nodeElements)
 	if err != nil {
 		return sdtypes.StateNode{}, nil, err
 	}
@@ -150,7 +152,7 @@ func (sdb *builder) buildStateTrie(it trie.NodeIterator) ([]sdtypes.StateNode, [
 		}
 		switch node.NodeType {
 		case sdtypes.Leaf:
-			var account state.Account
+			var account types.StateAccount
 			if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
 				return nil, nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
 			}
@@ -205,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
@@ -233,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)},
 		})
 	}
 
@@ -251,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()
@@ -351,7 +355,7 @@ func (sdb *builder) createdAndUpdatedState(iters iterPair, watchedAddresses []co
 		case sdtypes.Leaf:
 			// created vs updated is important for leaf nodes since we need to diff their storage
 			// so we need to map all changed accounts at B to their leafkey, since account can change pathes but not leafkey
-			var account state.Account
+			var account types.StateAccount
 			if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
 				return nil, nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
 			}
@@ -395,7 +399,7 @@ func (sdb *builder) createdAndUpdatedStateWithIntermediateNodes(iters iterPair,
 		case sdtypes.Leaf:
 			// created vs updated is important for leaf nodes since we need to diff their storage
 			// so we need to map all changed accounts at B to their leafkey, since account can change paths but not leafkey
-			var account state.Account
+			var account types.StateAccount
 			if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
 				return nil, nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
 			}
@@ -445,7 +449,7 @@ func (sdb *builder) deletedOrUpdatedState(iters iterPair, diffPathsAtB map[strin
 		switch node.NodeType {
 		case sdtypes.Leaf:
 			// map all different accounts at A to their leafkey
-			var account state.Account
+			var account types.StateAccount
 			if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
 				return nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
 			}

pkg/builder_test.go

@@ -22,13 +22,11 @@ import (
 	"sort"
 	"testing"
 
-	"github.com/ethereum/go-ethereum/statediff"
-
 	"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/crypto"
 	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/statediff"
 	sdtypes "github.com/ethereum/go-ethereum/statediff/types"
 
 	pkg "github.com/vulcanize/eth-statediff-service/pkg"
@@ -89,7 +87,7 @@ var (
 		slot0StorageValue,
 	})
 
-	contractAccountAtBlock2, _ = rlp.EncodeToBytes(state.Account{
+	contractAccountAtBlock2, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    1,
 		Balance:  big.NewInt(0),
 		CodeHash: common.HexToHash("0xaaea5efba4fd7b45d7ec03918ac5d8b31aa93b48986af0e6b591f0f087c80127").Bytes(),
@@ -99,7 +97,7 @@ var (
 		common.Hex2Bytes("3114658a74d9cc9f7acf2c5cd696c3494d7c344d78bfec3add0d91ec4e8d1c45"),
 		contractAccountAtBlock2,
 	})
-	contractAccountAtBlock3, _ = rlp.EncodeToBytes(state.Account{
+	contractAccountAtBlock3, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    1,
 		Balance:  big.NewInt(0),
 		CodeHash: common.HexToHash("0xaaea5efba4fd7b45d7ec03918ac5d8b31aa93b48986af0e6b591f0f087c80127").Bytes(),
@@ -109,7 +107,7 @@ var (
 		common.Hex2Bytes("3114658a74d9cc9f7acf2c5cd696c3494d7c344d78bfec3add0d91ec4e8d1c45"),
 		contractAccountAtBlock3,
 	})
-	contractAccountAtBlock4, _ = rlp.EncodeToBytes(state.Account{
+	contractAccountAtBlock4, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    1,
 		Balance:  big.NewInt(0),
 		CodeHash: common.HexToHash("0xaaea5efba4fd7b45d7ec03918ac5d8b31aa93b48986af0e6b591f0f087c80127").Bytes(),
@@ -119,7 +117,7 @@ var (
 		common.Hex2Bytes("3114658a74d9cc9f7acf2c5cd696c3494d7c344d78bfec3add0d91ec4e8d1c45"),
 		contractAccountAtBlock4,
 	})
-	contractAccountAtBlock5, _ = rlp.EncodeToBytes(state.Account{
+	contractAccountAtBlock5, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    1,
 		Balance:  big.NewInt(0),
 		CodeHash: common.HexToHash("0xaaea5efba4fd7b45d7ec03918ac5d8b31aa93b48986af0e6b591f0f087c80127").Bytes(),
@@ -130,7 +128,7 @@ var (
 		contractAccountAtBlock5,
 	})
 
-	minerAccountAtBlock1, _ = rlp.EncodeToBytes(state.Account{
+	minerAccountAtBlock1, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    0,
 		Balance:  big.NewInt(miningReward),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -140,7 +138,7 @@ var (
 		common.Hex2Bytes("3380c7b7ae81a58eb98d9c78de4a1fd7fd9535fc953ed2be602daaa41767312a"),
 		minerAccountAtBlock1,
 	})
-	minerAccountAtBlock2, _ = rlp.EncodeToBytes(state.Account{
+	minerAccountAtBlock2, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    0,
 		Balance:  big.NewInt(miningReward + miningReward),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -151,7 +149,7 @@ var (
 		minerAccountAtBlock2,
 	})
 
-	account1AtBlock1, _ = rlp.EncodeToBytes(state.Account{
+	account1AtBlock1, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    0,
 		Balance:  big.NewInt(balanceChange10000),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -161,7 +159,7 @@ var (
 		common.Hex2Bytes("3926db69aaced518e9b9f0f434a473e7174109c943548bb8f23be41ca76d9ad2"),
 		account1AtBlock1,
 	})
-	account1AtBlock2, _ = rlp.EncodeToBytes(state.Account{
+	account1AtBlock2, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    2,
 		Balance:  big.NewInt(block1Account1Balance - balanceChange1000 + balanceChange1000),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -171,7 +169,7 @@ var (
 		common.Hex2Bytes("3926db69aaced518e9b9f0f434a473e7174109c943548bb8f23be41ca76d9ad2"),
 		account1AtBlock2,
 	})
-	account1AtBlock5, _ = rlp.EncodeToBytes(state.Account{
+	account1AtBlock5, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    2,
 		Balance:  big.NewInt(block1Account1Balance - balanceChange1000 + balanceChange1000 + miningReward),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -181,7 +179,7 @@ var (
 		common.Hex2Bytes("3926db69aaced518e9b9f0f434a473e7174109c943548bb8f23be41ca76d9ad2"),
 		account1AtBlock5,
 	})
-	account1AtBlock6, _ = rlp.EncodeToBytes(state.Account{
+	account1AtBlock6, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    3,
 		Balance:  big.NewInt(block1Account1Balance - balanceChange1000 + balanceChange1000 + miningReward),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -192,7 +190,7 @@ var (
 		account1AtBlock6,
 	})
 
-	account2AtBlock2, _ = rlp.EncodeToBytes(state.Account{
+	account2AtBlock2, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    0,
 		Balance:  big.NewInt(balanceChange1000),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -202,7 +200,7 @@ var (
 		common.Hex2Bytes("3957f3e2f04a0764c3a0491b175f69926da61efbcc8f61fa1455fd2d2b4cdd45"),
 		account2AtBlock2,
 	})
-	account2AtBlock3, _ = rlp.EncodeToBytes(state.Account{
+	account2AtBlock3, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    0,
 		Balance:  big.NewInt(block2Account2Balance + miningReward),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -212,7 +210,7 @@ var (
 		common.Hex2Bytes("3957f3e2f04a0764c3a0491b175f69926da61efbcc8f61fa1455fd2d2b4cdd45"),
 		account2AtBlock3,
 	})
-	account2AtBlock4, _ = rlp.EncodeToBytes(state.Account{
+	account2AtBlock4, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    0,
 		Balance:  big.NewInt(block2Account2Balance + miningReward*2),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -222,7 +220,7 @@ var (
 		common.Hex2Bytes("3957f3e2f04a0764c3a0491b175f69926da61efbcc8f61fa1455fd2d2b4cdd45"),
 		account2AtBlock4,
 	})
-	account2AtBlock6, _ = rlp.EncodeToBytes(state.Account{
+	account2AtBlock6, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    0,
 		Balance:  big.NewInt(block2Account2Balance + miningReward*3),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -233,7 +231,7 @@ var (
 		account2AtBlock6,
 	})
 
-	bankAccountAtBlock0, _ = rlp.EncodeToBytes(state.Account{
+	bankAccountAtBlock0, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    0,
 		Balance:  big.NewInt(testhelpers.TestBankFunds.Int64()),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -243,7 +241,7 @@ var (
 		common.Hex2Bytes("2000bf49f440a1cd0527e4d06e2765654c0f56452257516d793a9b8d604dcfdf2a"),
 		bankAccountAtBlock0,
 	})
-	bankAccountAtBlock1, _ = rlp.EncodeToBytes(state.Account{
+	bankAccountAtBlock1, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    1,
 		Balance:  big.NewInt(testhelpers.TestBankFunds.Int64() - balanceChange10000),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -253,7 +251,7 @@ var (
 		common.Hex2Bytes("30bf49f440a1cd0527e4d06e2765654c0f56452257516d793a9b8d604dcfdf2a"),
 		bankAccountAtBlock1,
 	})
-	bankAccountAtBlock2, _ = rlp.EncodeToBytes(state.Account{
+	bankAccountAtBlock2, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    2,
 		Balance:  big.NewInt(block1BankBalance - balanceChange1000),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -263,7 +261,7 @@ var (
 		common.Hex2Bytes("30bf49f440a1cd0527e4d06e2765654c0f56452257516d793a9b8d604dcfdf2a"),
 		bankAccountAtBlock2,
 	})
-	bankAccountAtBlock3, _ = rlp.EncodeToBytes(state.Account{
+	bankAccountAtBlock3, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    3,
 		Balance:  big.NewInt(99989000),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -273,7 +271,7 @@ var (
 		common.Hex2Bytes("30bf49f440a1cd0527e4d06e2765654c0f56452257516d793a9b8d604dcfdf2a"),
 		bankAccountAtBlock3,
 	})
-	bankAccountAtBlock4, _ = rlp.EncodeToBytes(state.Account{
+	bankAccountAtBlock4, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    6,
 		Balance:  big.NewInt(99989000),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -283,7 +281,7 @@ var (
 		common.Hex2Bytes("30bf49f440a1cd0527e4d06e2765654c0f56452257516d793a9b8d604dcfdf2a"),
 		bankAccountAtBlock4,
 	})
-	bankAccountAtBlock5, _ = rlp.EncodeToBytes(state.Account{
+	bankAccountAtBlock5, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    7,
 		Balance:  big.NewInt(99989000),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -1735,7 +1733,7 @@ var (
 		slot00StorageValue,
 	})
 
-	contractAccountAtBlock01, _ = rlp.EncodeToBytes(state.Account{
+	contractAccountAtBlock01, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    1,
 		Balance:  big.NewInt(0),
 		CodeHash: common.HexToHash("0xaaea5efba4fd7b45d7ec03918ac5d8b31aa93b48986af0e6b591f0f087c80127").Bytes(),
@@ -1746,7 +1744,7 @@ var (
 		contractAccountAtBlock01,
 	})
 
-	bankAccountAtBlock01, _ = rlp.EncodeToBytes(state.Account{
+	bankAccountAtBlock01, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    1,
 		Balance:  big.NewInt(testhelpers.TestBankFunds.Int64() + miningReward),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),
@@ -1756,7 +1754,7 @@ var (
 		common.Hex2Bytes("30bf49f440a1cd0527e4d06e2765654c0f56452257516d793a9b8d604dcfdf2a"),
 		bankAccountAtBlock01,
 	})
-	bankAccountAtBlock02, _ = rlp.EncodeToBytes(state.Account{
+	bankAccountAtBlock02, _ = rlp.EncodeToBytes(types.StateAccount{
 		Nonce:    2,
 		Balance:  big.NewInt(testhelpers.TestBankFunds.Int64() + miningReward*2),
 		CodeHash: testhelpers.NullCodeHash.Bytes(),

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

@@ -25,25 +25,43 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/params"
+	"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
 }
 
-// NewLvlDBReader creates a new LvlDBReader
-func NewLvlDBReader(path, ancient string, chainConfig *params.ChainConfig) (*LvlDBReader, error) {
-	edb, err := rawdb.NewLevelDBDatabaseWithFreezer(path, 1024, 256, ancient, "eth-statediff-service")
+// LvLDBReaderConfig struct for initializing a LvlDBReader
+type LvLDBReaderConfig struct {
+	TrieConfig        *trie.Config
+	ChainConfig       *params.ChainConfig
+	Path, AncientPath string
+	DBCacheSize       int
+}
+
+// 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
 	}
 	return &LvlDBReader{
 		ethDB:       edb,
-		stateDB:     state.NewDatabase(edb),
-		chainConfig: chainConfig,
+		stateDB:     state.NewDatabaseWithConfig(edb, conf.TrieConfig),
+		chainConfig: conf.ChainConfig,
 	}, nil
 }
 

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,22 +369,47 @@ 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
 	}
 	// defer handling of commit/rollback for any return case
-	defer tx.Close()
 	output := func(node sdtypes.StateNode) error {
 		return sds.indexer.PushStateNode(tx, node)
 	}
 	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 nil
+	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/testhelpers/helpers.go

@@ -32,6 +32,7 @@ import (
 // the returned hash chain is ordered head->parent.
 func MakeChain(n int, parent *types.Block, chainGen func(int, *core.BlockGen)) ([]*types.Block, *core.BlockChain) {
 	config := params.TestChainConfig
+	config.LondonBlock.Set(big.NewInt(10))
 	blocks, _ := core.GenerateChain(config, parent, ethash.NewFaker(), Testdb, n, chainGen)
 	chain, _ := core.NewBlockChain(Testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, nil)
 	return blocks, chain

pkg/testhelpers/mocks/blockchain.go

@@ -1,128 +0,0 @@
-// Copyright 2019 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library 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 Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
-package mocks
-
-import (
-	"errors"
-	"math/big"
-	"time"
-
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core"
-	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/event"
-)
-
-// BlockChain is a mock blockchain for testing
-type BlockChain struct {
-	HashesLookedUp         []common.Hash
-	blocksToReturnByHash   map[common.Hash]*types.Block
-	blocksToReturnByNumber map[uint64]*types.Block
-	callCount              int
-	ChainEvents            []core.ChainEvent
-	Receipts               map[common.Hash]types.Receipts
-	TDByHash               map[common.Hash]*big.Int
-}
-
-// SetBlocksForHashes mock method
-func (blockChain *BlockChain) SetBlocksForHashes(blocks map[common.Hash]*types.Block) {
-	if blockChain.blocksToReturnByHash == nil {
-		blockChain.blocksToReturnByHash = make(map[common.Hash]*types.Block)
-	}
-	blockChain.blocksToReturnByHash = blocks
-}
-
-// GetBlockByHash mock method
-func (blockChain *BlockChain) GetBlockByHash(hash common.Hash) *types.Block {
-	blockChain.HashesLookedUp = append(blockChain.HashesLookedUp, hash)
-
-	var block *types.Block
-	if len(blockChain.blocksToReturnByHash) > 0 {
-		block = blockChain.blocksToReturnByHash[hash]
-	}
-
-	return block
-}
-
-// SetChainEvents mock method
-func (blockChain *BlockChain) SetChainEvents(chainEvents []core.ChainEvent) {
-	blockChain.ChainEvents = chainEvents
-}
-
-// SubscribeChainEvent mock method
-func (blockChain *BlockChain) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
-	subErr := errors.New("subscription error")
-
-	var eventCounter int
-	subscription := event.NewSubscription(func(quit <-chan struct{}) error {
-		for _, chainEvent := range blockChain.ChainEvents {
-			if eventCounter > 1 {
-				time.Sleep(250 * time.Millisecond)
-				return subErr
-			}
-			select {
-			case ch <- chainEvent:
-			case <-quit:
-				return nil
-			}
-			eventCounter++
-		}
-		return nil
-	})
-
-	return subscription
-}
-
-// SetReceiptsForHash test method
-func (blockChain *BlockChain) SetReceiptsForHash(hash common.Hash, receipts types.Receipts) {
-	if blockChain.Receipts == nil {
-		blockChain.Receipts = make(map[common.Hash]types.Receipts)
-	}
-	blockChain.Receipts[hash] = receipts
-}
-
-// GetReceiptsByHash mock method
-func (blockChain *BlockChain) GetReceiptsByHash(hash common.Hash) types.Receipts {
-	return blockChain.Receipts[hash]
-}
-
-// SetBlockForNumber test method
-func (blockChain *BlockChain) SetBlockForNumber(block *types.Block, number uint64) {
-	if blockChain.blocksToReturnByNumber == nil {
-		blockChain.blocksToReturnByNumber = make(map[uint64]*types.Block)
-	}
-	blockChain.blocksToReturnByNumber[number] = block
-}
-
-// GetBlockByNumber mock method
-func (blockChain *BlockChain) GetBlockByNumber(number uint64) *types.Block {
-	return blockChain.blocksToReturnByNumber[number]
-}
-
-// GetTdByHash mock method
-func (blockChain *BlockChain) GetTdByHash(hash common.Hash) *big.Int {
-	return blockChain.TDByHash[hash]
-}
-
-func (blockChain *BlockChain) SetTdByHash(hash common.Hash, td *big.Int) {
-	if blockChain.TDByHash == nil {
-		blockChain.TDByHash = make(map[common.Hash]*big.Int)
-	}
-	blockChain.TDByHash[hash] = td
-}
-
-func (blockChain *BlockChain) UnlockTrie(root common.Hash) {}

pkg/testhelpers/mocks/builder.go

@@ -1,65 +0,0 @@
-// Copyright 2019 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library 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 Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
-package mocks
-
-import (
-	"github.com/ethereum/go-ethereum/core/types"
-	sd "github.com/ethereum/go-ethereum/statediff/types"
-)
-
-// Builder is a mock state diff builder
-type Builder struct {
-	Args         sd.Args
-	Params       sd.Params
-	stateDiff    sd.StateObject
-	block        *types.Block
-	stateTrie    sd.StateObject
-	builderError error
-}
-
-// BuildStateDiffObject mock method
-func (builder *Builder) BuildStateDiffObject(args sd.Args, params sd.Params) (sd.StateObject, error) {
-	builder.Args = args
-	builder.Params = params
-
-	return builder.stateDiff, builder.builderError
-}
-
-// BuildStateDiffObject mock method
-func (builder *Builder) WriteStateDiffObject(args sd.StateRoots, params sd.Params, output sd.StateNodeSink, codeOutput sd.CodeSink) error {
-	builder.StateRoots = args
-	builder.Params = params
-
-	return builder.builderError
-}
-
-// BuildStateTrieObject mock method
-func (builder *Builder) BuildStateTrieObject(block *types.Block) (sd.StateObject, error) {
-	builder.block = block
-
-	return builder.stateTrie, builder.builderError
-}
-
-// SetStateDiffToBuild mock method
-func (builder *Builder) SetStateDiffToBuild(stateDiff sd.StateObject) {
-	builder.stateDiff = stateDiff
-}
-
-// SetBuilderError mock method
-func (builder *Builder) SetBuilderError(err error) {
-	builder.builderError = err
-}

pkg/testhelpers/test_data.go

@@ -18,7 +18,6 @@ package testhelpers
 
 import (
 	"math/big"
-	"math/rand"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
@@ -42,12 +41,7 @@ func AddressToEncodedPath(address common.Address) []byte {
 // Test variables
 var (
 	EvenLeafFlag = []byte{byte(2) << 4}
-	BlockNumber  = big.NewInt(rand.Int63())
-	BlockHash    = "0xfa40fbe2d98d98b3363a778d52f2bcd29d6790b9b3f3cab2b167fd12d3550f73"
 	NullCodeHash = crypto.Keccak256Hash([]byte{})
-	StoragePath  = common.HexToHash("0xc5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470").Bytes()
-	StorageKey   = common.HexToHash("0000000000000000000000000000000000000000000000000000000000000001").Bytes()
-	StorageValue = common.Hex2Bytes("0x03")
 	NullHash     = common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000")
 
 	Testdb          = rawdb.NewMemoryDatabase()

pkg/types.go

@@ -20,8 +20,9 @@
 package statediff
 
 import (
-	"github.com/ethereum/go-ethereum/core/state"
-	sd "github.com/ethereum/go-ethereum/statediff/types"
+	"github.com/ethereum/go-ethereum/core/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
@@ -29,9 +30,15 @@ type AccountMap map[string]accountWrapper
 
 // accountWrapper is used to temporary associate the unpacked node with its raw values
 type accountWrapper struct {
-	Account   *state.Account
-	NodeType  sd.NodeType
+	Account   *types.StateAccount
+	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

version/version.go

@@ -0,0 +1,30 @@
+// 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 version
+
+import "fmt"
+
+const (
+	major = 2 // major version component of the current release
+	minor = 0 // minor version component of the current release
+	patch = 0 // patch version component of the current release
+)
+
+// Version holds the textual version string.
+var Version = func() string {
+	return fmt.Sprintf("%d.%d.%d", major, minor, patch)
+}()