Merge pull request #78 from vulcanize/release-v3.0.0

Release v3.0.0

.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

@@ -1,4 +1,4 @@
-FROM golang:1.13-alpine as builder
+FROM golang:1.16-alpine as builder
 
 RUN apk --update --no-cache add make git g++ linux-headers
 # DEBUG
@@ -15,7 +15,7 @@ RUN GO111MODULE=on GCO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -ldfl
 FROM alpine
 
 ARG USER="vdm"
-ARG CONFIG_FILE="./environments/example.toml"
+ARG CONFIG_FILE="./environments/config.toml"
 ARG EXPOSE_PORT=8545
 
 RUN adduser -Du 5000 $USER adm

README.md

@@ -23,14 +23,13 @@ Available RPC methods are:
   * `statediff_streamCodeAndCodeHash()`
   * `statediff_stateDiffAt()`
   * `statediff_writeStateDiffAt()`
+  * `statediff_writeStateDiffsInRange()`
 
-### `write`
+e.g. `curl -X POST -H 'Content-Type: application/json' --data '{"jsonrpc":"2.0","method":"statediff_writeStateDiffsInRange","params":['"$BEGIN"', '"$END"', {"intermediateStateNodes":true,"intermediateStorageNodes":true,"includeBlock":true,"includeReceipts":true,"includeTD":true,"includeCode":true}],"id":1}' "$HOST":"$PORT"`
 
-To write state diffs directly to a database:
-
-`eth-statediff-service write --config=<config path>`
-
-This depends on the `database` settings being properly configured.
+The process can be configured locally with sets of ranges to process as a "prerun" to processing directed by the server endpoints.
+This is done by turning "prerun" on in the config (`statediff.prerun = true`) and defining ranged and params in the
+`prerun` section of the config as shown below.
 
 ## Configuration
 
@@ -42,27 +41,63 @@ An example config file:
     ancient = "/Users/user/Library/Ethereum/geth/chaindata/ancient"
 
 [server]
-    ipcPath = "~/.vulcanize/vulcanize.ipc"
+    ipcPath = ".ipc"
     httpPath = "127.0.0.1:8545"
 
-[write]
-    ranges = [[1, 2], [3, 4]]
-[write.params]
-    IntermediateStateNodes   = true
-    IntermediateStorageNodes = false
-    IncludeBlock             = true
-    IncludeReceipts          = true
-    IncludeTD                = true
-    IncludeCode              = false
-
 [statediff]
-    workers = 4
+    prerun = true
+    serviceWorkers = 1
+    workerQueueSize = 1024
+    trieWorkers = 4
+
+[prerun]
+    only = false
+    ranges = [
+        [0, 1000]
+    ]
+    [prerun.params]
+        intermediateStateNodes = true
+        intermediateStorageNodes = true
+        includeBlock = true
+        includeReceipts = true
+        includeTD = true
+        includeCode = true
+        watchedAddresses = []
+        watchedStorageKeys = []
 
 [log]
-    file = "~/.vulcanize/statediff.log"
+    file = ""
     level = "info"
 
 [eth]
     chainID = 1
 
+[database]
+    name     = "vulcanize_test"
+    hostname = "localhost"
+    port     = 5432
+    user     = "vulcanize"
+    password = "..."
+    type = "postgres"
+    driver = "sqlx"
+    dumpDestination = ""
+    filePath = ""
+
+[cache]
+    database = 1024
+    trie = 1024
+
+[prom]
+    dbStats = false
+    metrics = true
+    http = true
+    httpAddr = "localhost"
+    httpPort = "8889"
+
+[ethereum]
+    nodeID = ""
+    clientName = "eth-statediff-service"
+    genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
+    networkID = 1
+    chainID = 1
 ```

cmd/env.go

@@ -18,8 +18,6 @@ package cmd
 
 import (
 	"github.com/spf13/viper"
-
-	pg "github.com/ethereum/go-ethereum/statediff/indexer/postgres"
 )
 
 const (
@@ -33,31 +31,81 @@ const (
 	TRIE_CACHE_SIZE_MB = "TRIE_CACHE_SIZE_MB"
 	LVLDB_PATH         = "LVLDB_PATH"
 	LVLDB_ANCIENT      = "LVLDB_ANCIENT"
-	STATEDIFF_WORKERS  = "STATEDIFF_WORKERS"
-	WRITE_SERVER       = "WRITE_SERVER"
+
+	STATEDIFF_PRERUN            = "STATEDIFF_PRERUN"
+	STATEDIFF_TRIE_WORKERS      = "STATEDIFF_TRIE_WORKERS"
+	STATEDIFF_SERVICE_WORKERS   = "STATEDIFF_SERVICE_WORKERS"
+	STATEDIFF_WORKER_QUEUE_SIZE = "STATEDIFF_WORKER_QUEUE_SIZE"
+
+	SERVICE_IPC_PATH  = "SERVICE_IPC_PATH"
+	SERVICE_HTTP_PATH = "SERVICE_HTTP_PATH"
 
 	PROM_METRICS   = "PROM_METRICS"
 	PROM_HTTP      = "PROM_HTTP"
 	PROM_HTTP_ADDR = "PROM_HTTP_ADDR"
 	PROM_HTTP_PORT = "PROM_HTTP_PORT"
+	PROM_DB_STATS  = "PROM_DB_STATS"
+
+	PRERUN_ONLY        = "PRERUN_ONLY"
+	PRERUN_RANGE_START = "PRERUN_RANGE_START"
+	PRERUN_RANGE_STOP  = "PRERUN_RANGE_STOP"
+	PRERUN_INTERMEDIATE_STATE_NODES = "PRERUN_INTERMEDIATE_STATE_NODES"
+	PRERUN_INTERMEDIATE_STORAGE_NODES = "PRERUN_INTERMEDIATE_STORAGE_NODES"
+	PRERUN_INCLUDE_BLOCK = "PRERUN_INCLUDE_BLOCK"
+	PRERUN_INCLUDE_RECEIPTS = "PRERUN_INCLUDE_RECEIPTS"
+	PRERUN_INCLUDE_TD = "PRERUN_INCLUDE_TD"
+	PRERUN_INCLUDE_CODE = "PRERUN_INCLUDE_CODE"
+
+	LOG_LEVEL     = "LOG_LEVEL"
+	LOG_FILE_PATH = "LOG_FILE_PATH"
+
+	DATABASE_NAME     = "DATABASE_NAME"
+	DATABASE_HOSTNAME = "DATABASE_HOSTNAME"
+	DATABASE_PORT     = "DATABASE_PORT"
+	DATABASE_USER     = "DATABASE_USER"
+	DATABASE_PASSWORD = "DATABASE_PASSWORD"
+
+	DATABASE_TYPE        = "DATABASE_TYPE"
+	DATABASE_DRIVER_TYPE = "DATABASE_DRIVER_TYPE"
+	DATABASE_DUMP_DST    = "DATABASE_DUMP_DST"
+	DATABASE_FILE_PATH   = "DATABASE_FILE_PATH"
+
+	DATABASE_MAX_IDLE_CONNECTIONS = "DATABASE_MAX_IDLE_CONNECTIONS"
+	DATABASE_MAX_OPEN_CONNECTIONS = "DATABASE_MAX_OPEN_CONNECTIONS"
+	DATABASE_MIN_OPEN_CONNS       = "DATABASE_MIN_OPEN_CONNS"
+	DATABASE_MAX_CONN_LIFETIME    = "DATABASE_MAX_CONN_LIFETIME"
+	DATABASE_CONN_TIMEOUT         = "DATABSE_CONN_TIMEOUT"
+	DATABASE_MAX_CONN_IDLE_TIME   = "DATABASE_MAX_CONN_IDLE_TIME"
 )
 
 // 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)
 	viper.BindEnv("ethereum.networkID", ETH_NETWORK_ID)
 	viper.BindEnv("ethereum.chainID", ETH_CHAIN_ID)
 
-	viper.BindEnv("database.name", pg.DATABASE_NAME)
-	viper.BindEnv("database.hostname", pg.DATABASE_HOSTNAME)
-	viper.BindEnv("database.port", pg.DATABASE_PORT)
-	viper.BindEnv("database.user", pg.DATABASE_USER)
-	viper.BindEnv("database.password", pg.DATABASE_PASSWORD)
-	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("database.name", DATABASE_NAME)
+	viper.BindEnv("database.hostname", DATABASE_HOSTNAME)
+	viper.BindEnv("database.port", DATABASE_PORT)
+	viper.BindEnv("database.user", DATABASE_USER)
+	viper.BindEnv("database.password", DATABASE_PASSWORD)
+
+	viper.BindEnv("database.maxIdle", DATABASE_MAX_IDLE_CONNECTIONS)
+	viper.BindEnv("database.maxOpen", DATABASE_MAX_OPEN_CONNECTIONS)
+	viper.BindEnv("database.minOpen", DATABASE_MIN_OPEN_CONNS)
+	viper.BindEnv("database.maxConnLifetime", DATABASE_MAX_CONN_LIFETIME)
+	viper.BindEnv("database.connTimeout", DATABASE_CONN_TIMEOUT)
+	viper.BindEnv("database.maxIdleTime", DATABASE_MAX_CONN_IDLE_TIME)
+
+	viper.BindEnv("database.type", DATABASE_TYPE)
+	viper.BindEnv("database.driver", DATABASE_DRIVER_TYPE)
+	viper.BindEnv("database.dumpDestination", DATABASE_DUMP_DST)
+	viper.BindEnv("database.filePath", DATABASE_FILE_PATH)
 
 	viper.BindEnv("cache.database", DB_CACHE_SIZE_MB)
 	viper.BindEnv("cache.trie", TRIE_CACHE_SIZE_MB)
@@ -69,6 +117,23 @@ func init() {
 	viper.BindEnv("prom.http", PROM_HTTP)
 	viper.BindEnv("prom.httpAddr", PROM_HTTP_ADDR)
 	viper.BindEnv("prom.httpPort", PROM_HTTP_PORT)
+	viper.BindEnv("prom.dbStats", PROM_DB_STATS)
 
-	viper.BindEnv("statediff.workers", STATEDIFF_WORKERS)
+	viper.BindEnv("statediff.serviceWorkers", STATEDIFF_SERVICE_WORKERS)
+	viper.BindEnv("statediff.trieWorkers", STATEDIFF_TRIE_WORKERS)
+	viper.BindEnv("statediff.workerQueueSize", STATEDIFF_WORKER_QUEUE_SIZE)
+
+	viper.BindEnv("statediff.prerun", STATEDIFF_PRERUN)
+	viper.BindEnv("prerun.only", PRERUN_ONLY)
+	viper.BindEnv("prerun.start", PRERUN_RANGE_START)
+	viper.BindEnv("prerun.stop", PRERUN_RANGE_STOP)
+	viper.BindEnv("prerun.params.intermediateStateNodes", PRERUN_INTERMEDIATE_STATE_NODES)
+	viper.BindEnv("prerun.params.intermediateStorageNodes", PRERUN_INTERMEDIATE_STORAGE_NODES)
+	viper.BindEnv("prerun.params.includeBlock", PRERUN_INCLUDE_BLOCK)
+	viper.BindEnv("prerun.params.includeReceipts", PRERUN_INCLUDE_RECEIPTS)
+	viper.BindEnv("prerun.params.includeTD", PRERUN_INCLUDE_TD)
+	viper.BindEnv("prerun.params.includeCode", PRERUN_INCLUDE_CODE)
+
+	viper.BindEnv("log.level", LOG_LEVEL)
+	viper.BindEnv("log.file", LOG_FILE_PATH)
 }

cmd/root.go

@@ -18,11 +18,18 @@ package cmd
 
 import (
 	"fmt"
+	"math/rand"
 	"os"
 	"strings"
+	"time"
 
+	"github.com/ethereum/go-ethereum/cmd/utils"
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/dump"
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/file"
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
+	"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
 	"github.com/ethereum/go-ethereum/statediff/indexer/node"
-	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
+	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -68,6 +75,7 @@ func initFuncs(cmd *cobra.Command, args []string) {
 	}
 
 	if viper.GetBool("prom.metrics") {
+		log.Info("initializing prometheus metrics")
 		prom.Init()
 	}
 
@@ -77,12 +85,12 @@ func initFuncs(cmd *cobra.Command, args []string) {
 			viper.GetString("prom.httpAddr"),
 			viper.GetString("prom.httpPort"),
 		)
+		log.Info("starting prometheus server")
 		prom.Listen(addr)
 	}
 }
 
 func logLevel() error {
-	viper.BindEnv("log.level", "LOGRUS_LEVEL")
 	lvl, err := log.ParseLevel(viper.GetString("log.level"))
 	if err != nil {
 		return err
@@ -100,23 +108,40 @@ 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(),
 		"log level (trace, debug, info, warn, error, fatal, panic")
 
 	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")
 	rootCmd.PersistentFlags().String("database-hostname", "localhost", "database hostname")
 	rootCmd.PersistentFlags().String("database-user", "", "database user")
 	rootCmd.PersistentFlags().String("database-password", "", "database password")
+	rootCmd.PersistentFlags().Int("database-max-idle", 0, "database max number of idle connections")
+	rootCmd.PersistentFlags().Int("database-max-open", 0, "database max number of open connections")
+	rootCmd.PersistentFlags().Int("database-min-open", 0, "database min number of open connections")
+	rootCmd.PersistentFlags().Duration("database-max-conn-lifetime", 0, "database max connection lifetime")
+	rootCmd.PersistentFlags().Duration("database-conn-timeout", 0, "database connection timeout")
+	rootCmd.PersistentFlags().Duration("database-max-idle-time", 0, "database max connection idle time")
+	rootCmd.PersistentFlags().String("database-type", "postgres", "database type (currently supported: postgres, dump)")
+	rootCmd.PersistentFlags().String("database-driver", "sqlx", "database driver type (currently supported: sqlx, pgx)")
+	rootCmd.PersistentFlags().String("database-dump-dst", "stdout", "dump destination (for database-type=dump; options: stdout, stderr, discard)")
+	rootCmd.PersistentFlags().String("database-file-path", "", "full file path (for database-type=file)")
 
 	rootCmd.PersistentFlags().String("eth-node-id", "", "eth node id")
-	rootCmd.PersistentFlags().String("eth-client-name", "Geth", "eth client name")
+	rootCmd.PersistentFlags().String("eth-client-name", "eth-statediff-service", "eth client name")
 	rootCmd.PersistentFlags().String("eth-genesis-block",
 		"0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3", "eth genesis block hash")
 	rootCmd.PersistentFlags().String("eth-network-id", "1", "eth network id")
@@ -128,30 +153,76 @@ func init() {
 	rootCmd.PersistentFlags().Bool("prom-http", false, "enable prometheus http service")
 	rootCmd.PersistentFlags().String("prom-http-addr", "127.0.0.1", "prometheus http host")
 	rootCmd.PersistentFlags().String("prom-http-port", "8080", "prometheus http port")
+	rootCmd.PersistentFlags().Bool("prom-db-stats", false, "enables prometheus db stats")
+	rootCmd.PersistentFlags().Bool("prom-metrics", false, "enable prometheus metrics")
 
-	rootCmd.PersistentFlags().Bool("metrics", false, "enable metrics")
+	rootCmd.PersistentFlags().Bool("prerun-only", false, "only process pre-configured ranges; exit afterwards")
+	rootCmd.PersistentFlags().Int("prerun-start", 0, "start height for a prerun range")
+	rootCmd.PersistentFlags().Int("prerun-stop", 0, "stop height for a prerun range")
+	rootCmd.PersistentFlags().Bool("prerun-intermediate-state-nodes", true, "include intermediate state nodes in state diff")
+	rootCmd.PersistentFlags().Bool("prerun-intermediate-storage-nodes", true, "include intermediate storage nodes in state diff")
+	rootCmd.PersistentFlags().Bool("prerun-include-block", true, "include block data in the statediff payload")
+	rootCmd.PersistentFlags().Bool("prerun-include-receipts", true, "include receipts in the statediff payload")
+	rootCmd.PersistentFlags().Bool("prerun-include-td", true, "include td in the statediff payload")
+	rootCmd.PersistentFlags().Bool("prerun-include-code", true, "include code and codehash mappings in statediff payload")
+
+
+	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"))
 	viper.BindPFlag("database.port", rootCmd.PersistentFlags().Lookup("database-port"))
 	viper.BindPFlag("database.hostname", rootCmd.PersistentFlags().Lookup("database-hostname"))
 	viper.BindPFlag("database.user", rootCmd.PersistentFlags().Lookup("database-user"))
 	viper.BindPFlag("database.password", rootCmd.PersistentFlags().Lookup("database-password"))
+	viper.BindPFlag("database.maxIdle", rootCmd.PersistentFlags().Lookup("database-max-idle"))
+	viper.BindPFlag("database.maxOpen", rootCmd.PersistentFlags().Lookup("database-max-open"))
+	viper.BindPFlag("database.minOpen", rootCmd.PersistentFlags().Lookup("database-min-open"))
+	viper.BindPFlag("database.maxConnLifetime", rootCmd.PersistentFlags().Lookup("database-max-conn-lifetime"))
+	viper.BindPFlag("database.connTimeout", rootCmd.PersistentFlags().Lookup("database-conn-timeout"))
+	viper.BindPFlag("database.maxIdleTime", rootCmd.PersistentFlags().Lookup("database-max-idle-time"))
+	viper.BindPFlag("database.type", rootCmd.PersistentFlags().Lookup("database-type"))
+	viper.BindPFlag("database.driver", rootCmd.PersistentFlags().Lookup("database-driver"))
+	viper.BindPFlag("database.dumpDestination", rootCmd.PersistentFlags().Lookup("database-dump-dst"))
+	viper.BindPFlag("database.filePath", rootCmd.PersistentFlags().Lookup("database-file-path"))
+
 	viper.BindPFlag("ethereum.nodeID", rootCmd.PersistentFlags().Lookup("eth-node-id"))
 	viper.BindPFlag("ethereum.clientName", rootCmd.PersistentFlags().Lookup("eth-client-name"))
 	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.metrics", rootCmd.PersistentFlags().Lookup("metrics"))
+	viper.BindPFlag("prom.dbStats", rootCmd.PersistentFlags().Lookup("prom-db-stats"))
+	viper.BindPFlag("prom.metrics", rootCmd.PersistentFlags().Lookup("prom-metrics"))
+
+	viper.BindPFlag("prerun.only", rootCmd.PersistentFlags().Lookup("prerun-only"))
+	viper.BindPFlag("prerun.start", rootCmd.PersistentFlags().Lookup("prerun-start"))
+	viper.BindPFlag("prerun.stop", rootCmd.PersistentFlags().Lookup("prerun-stop"))
+	viper.BindPFlag("prerun.params.intermediateStateNodes", rootCmd.PersistentFlags().Lookup("prerun-intermediate-state-nodes"))
+	viper.BindPFlag("prerun.params.intermediateStorageNodes", rootCmd.PersistentFlags().Lookup("prerun-intermediate-storage-nodes"))
+	viper.BindPFlag("prerun.params.includeBlock", rootCmd.PersistentFlags().Lookup("prerun-include-block"))
+	viper.BindPFlag("prerun.params.includeReceipts", rootCmd.PersistentFlags().Lookup("prerun-include-receipts"))
+	viper.BindPFlag("prerun.params.includeTD", rootCmd.PersistentFlags().Lookup("prerun-include-td"))
+	viper.BindPFlag("prerun.params.includeCode", rootCmd.PersistentFlags().Lookup("prerun-include-code"))
+
+	rand.Seed(time.Now().UnixNano())
 }
 
 func initConfig() {
@@ -167,30 +238,126 @@ func initConfig() {
 	}
 }
 
-func GetEthNodeInfo() node.Info {
+func getEthNodeInfo() node.Info {
+	var nodeID, genesisBlock, networkID, clientName string
+	var chainID uint64
+	if !viper.IsSet("ethereum.nodeID") {
+		nodeID = randSeq(12)
+	} else {
+		nodeID = viper.GetString("ethereum.nodeID")
+	}
+	genesisBlock = viper.GetString("ethereum.genesisBlock")
+	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:           viper.GetString("ethereum.nodeID"),
-		ClientName:   viper.GetString("ethereum.clientName"),
-		GenesisBlock: viper.GetString("ethereum.genesisBlock"),
-		NetworkID:    viper.GetString("ethereum.networkID"),
-		ChainID:      viper.GetUint64("ethereum.chainID"),
+		ID:           nodeID,
+		ClientName:   clientName,
+		GenesisBlock: genesisBlock,
+		NetworkID:    networkID,
+		ChainID:      chainID,
 	}
 }
 
-func GetDBParams() postgres.ConnectionParams {
-	return postgres.ConnectionParams{
-		Name:     viper.GetString("database.name"),
-		Hostname: viper.GetString("database.hostname"),
-		Port:     viper.GetInt("database.port"),
-		User:     viper.GetString("database.user"),
-		Password: viper.GetString("database.password"),
+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 GetDBConfig() postgres.ConnectionConfig {
-	return postgres.ConnectionConfig{
-		MaxIdle:     viper.GetInt("database.maxIdle"),
-		MaxOpen:     viper.GetInt("database.maxOpen"),
-		MaxLifetime: viper.GetInt("database.maxLifetime"),
+// getConfig constructs and returns the appropriate config from viper params
+func getConfig(nodeInfo node.Info) (interfaces.Config, error) {
+	dbTypeStr := viper.GetString("database.type")
+	dbType, err := shared.ResolveDBType(dbTypeStr)
+	if err != nil {
+		return nil, err
 	}
+	logWithCommand.Infof("configuring service for database type: %s", dbType)
+	var indexerConfig interfaces.Config
+	switch dbType {
+	case shared.FILE:
+		logWithCommand.Info("starting in sql file writing mode")
+		filePathStr := viper.GetString("database.filePath")
+		if filePathStr == "" {
+			logWithCommand.Fatal("when operating in sql file writing mode a file path must be provided")
+		}
+		indexerConfig = file.Config{FilePath: filePathStr}
+	case shared.DUMP:
+		logWithCommand.Info("starting in data dump mode")
+		dumpDstStr := viper.GetString("database.dumpDestination")
+		dumpDst, err := dump.ResolveDumpType(dumpDstStr)
+		if err != nil {
+			return nil, err
+		}
+		switch dumpDst {
+		case dump.STDERR:
+			indexerConfig = dump.Config{Dump: os.Stdout}
+		case dump.STDOUT:
+			indexerConfig = dump.Config{Dump: os.Stderr}
+		case dump.DISCARD:
+			indexerConfig = dump.Config{Dump: dump.NewDiscardWriterCloser()}
+		default:
+			return nil, fmt.Errorf("unrecognized dump destination: %s", dumpDst)
+		}
+	case shared.POSTGRES:
+		logWithCommand.Info("starting in postgres mode")
+		driverTypeStr := viper.GetString("database.driver")
+		driverType, err := postgres.ResolveDriverType(driverTypeStr)
+		if err != nil {
+			utils.Fatalf("%v", err)
+		}
+		pgConfig := postgres.Config{
+			Hostname:     viper.GetString("database.hostname"),
+			Port:         viper.GetInt("database.port"),
+			DatabaseName: viper.GetString("database.name"),
+			Username:     viper.GetString("database.user"),
+			Password:     viper.GetString("database.password"),
+			ID:           nodeInfo.ID,
+			ClientName:   nodeInfo.ClientName,
+			Driver:       driverType,
+		}
+		if viper.IsSet("database.maxIdle") {
+			pgConfig.MaxIdle = viper.GetInt("database.maxIdle")
+		}
+		if viper.IsSet("database.maxOpen") {
+			pgConfig.MaxConns = viper.GetInt("database.maxOpen")
+		}
+		if viper.IsSet("database.minOpen") {
+			pgConfig.MinConns = viper.GetInt("database.minOpen")
+		}
+		if viper.IsSet("database.maxConnLifetime") {
+			pgConfig.MaxConnLifetime = viper.GetDuration("database.maxConnLifetime")
+		}
+		if viper.IsSet("database.connTimeout") {
+			pgConfig.ConnTimeout = viper.GetDuration("database.connTimeout")
+		}
+		if viper.IsSet("database.maxIdleTime") {
+			pgConfig.MaxConnIdleTime = viper.GetDuration("database.maxIdleTime")
+		}
+		indexerConfig = pgConfig
+	default:
+		return nil, fmt.Errorf("unrecognized db type: %s", dbType)
+	}
+	return indexerConfig, nil
 }

cmd/serve.go

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

cmd/util.go

@@ -1,17 +1,23 @@
 package cmd
 
 import (
+	"context"
+	"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"
 )
 
-func createStateDiffService() (sd.IService, error) {
+type blockRange [2]uint64
+
+func createStateDiffService() (sd.StateDiffService, error) {
+	// load some necessary params
 	logWithCommand.Info("Loading statediff service parameters")
 	path := viper.GetString("leveldb.path")
 	ancientPath := viper.GetString("leveldb.ancient")
@@ -19,49 +25,111 @@ func createStateDiffService() (sd.IService, error) {
 		logWithCommand.Fatal("require a valid eth leveldb primary datastore path and ancient datastore path")
 	}
 
-	nodeInfo := GetEthNodeInfo()
-	config, err := chainConfig(nodeInfo.ChainID)
+	nodeInfo := getEthNodeInfo()
+	chainConf, err := chainConfig(nodeInfo.ChainID)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
 	// create leveldb reader
 	logWithCommand.Info("Creating leveldb reader")
-	conf := sd.ReaderConfig{
+	readerConf := sd.LvLDBReaderConfig{
 		TrieConfig: &trie.Config{
 			Cache:     viper.GetInt("cache.trie"),
 			Journal:   "",
 			Preimages: false,
 		},
-		ChainConfig: config,
+		ChainConfig: chainConf,
 		Path:        path,
 		AncientPath: ancientPath,
 		DBCacheSize: viper.GetInt("cache.database"),
 	}
-	lvlDBReader, err := sd.NewLvlDBReader(conf)
+	lvlDBReader, err := sd.NewLvlDBReader(readerConf)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
 	// create statediff service
+	logWithCommand.Info("Setting up database")
+	conf, err := getConfig(nodeInfo)
+	if err != nil {
+		logWithCommand.Fatal(err)
+	}
+	logWithCommand.Info("Creating statediff indexer")
+	indexer, err := ind.NewStateDiffIndexer(context.Background(), chainConf, nodeInfo, conf)
+	if err != nil {
+		logWithCommand.Fatal(err)
+	}
 	logWithCommand.Info("Creating statediff service")
-	db, err := setupPostgres(nodeInfo)
-	if err != nil {
-		logWithCommand.Fatal(err)
+	sdConf := sd.Config{
+		ServiceWorkers:  viper.GetUint("statediff.serviceWorkers"),
+		TrieWorkers:     viper.GetUint("statediff.trieWorkers"),
+		WorkerQueueSize: viper.GetUint("statediff.workerQueueSize"),
+		PreRuns:         setupPreRunRanges(),
 	}
-	indexer, err := ind.NewStateDiffIndexer(config, db)
-	if err != nil {
-		logWithCommand.Fatal(err)
-	}
-	return sd.NewStateDiffService(lvlDBReader, indexer, viper.GetUint("statediff.workers"))
+	return sd.NewStateDiffService(lvlDBReader, indexer, sdConf)
 }
 
-func setupPostgres(nodeInfo node.Info) (*postgres.DB, error) {
-	params := GetDBParams()
-	db, err := postgres.NewDB(postgres.DbConnectionString(params), GetDBConfig(), nodeInfo)
-	if err != nil {
-		return nil, err
+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)
 	}
-	prom.RegisterDBCollector(params.Name, db.DB)
-	return db, nil
 }

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

cmd/write_test.go

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

docker-compose.yml

@@ -6,7 +6,7 @@ services:
       context: ./
       cache_from:
       - alpine:latest
-      - golang:1.13
+      - golang:1.16
       dockerfile: ./Dockerfile
       args:
         USER: "vdbm"

environments/config.toml

@@ -0,0 +1,59 @@
+[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"
+
+[database]
+    name     = ""
+    hostname = ""
+    port     = 5432
+    user     = ""
+    password = ""
+    type = "postgres"
+    driver = "sqlx"
+    dumpDestination = ""
+    filePath = ""
+
+[cache]
+    database = 1024
+    trie = 4096
+
+[prom]
+    dbStats = false
+    metrics = true
+    http = true
+    httpAddr = "0.0.0.0"
+    httpPort = 9100
+
+[ethereum]
+    nodeID = ""
+    clientName = "eth-statediff-service"
+    genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
+    networkID = 1
+    chainID = 1

environments/example.toml

@@ -7,34 +7,55 @@
     httpPath = "127.0.0.1:8545"
 
 [statediff]
-    workers = 4
+    prerun = true
+    serviceWorkers = 1
+    workerQueueSize = 1024
+    trieWorkers = 4
 
-[write]
+[prerun]
+    only = false
     ranges = [
-        [0, 0]
+        [0, 1000]
     ]
-    serve = ":8888"
+    [prerun.params]
+        intermediateStateNodes = true
+        intermediateStorageNodes = true
+        includeBlock = true
+        includeReceipts = true
+        includeTD = true
+        includeCode = true
+        watchedAddresses = []
+        watchedStorageKeys = []
 
 [log]
     file = ""
     level = "info"
 
-[eth]
-    chainID = 1
-
 [database]
     name     = "vulcanize_test"
     hostname = "localhost"
     port     = 5432
     user     = "vulcanize"
     password = "..."
+    type = "postgres"
+    driver = "sqlx"
+    dumpDestination = ""
+    filePath = ""
 
 [cache]
     database = 1024
     trie = 1024
 
 [prom]
+    dbStats = false
     metrics = true
     http = true
-    addr = "localhost"
-    port = "8889"
+    httpAddr = "localhost"
+    httpPort = "8889"
+
+[ethereum]
+    nodeID = ""
+    clientName = "eth-statediff-service"
+    genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
+    networkID = 1
+    chainID = 1

go.mod

@@ -1,16 +1,15 @@
 module github.com/vulcanize/eth-statediff-service
 
-go 1.13
+go 1.16
 
 require (
-	github.com/ethereum/go-ethereum v1.10.9
+	github.com/ethereum/go-ethereum v1.10.15
 	github.com/jmoiron/sqlx v1.2.0
-	github.com/prometheus/client_golang v1.0.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/stretchr/testify v1.7.0
-	github.com/vulcanize/go-eth-state-node-iterator v0.0.1-alpha.0.20211014064906-d23d01ed8191
+	github.com/spf13/cobra v1.3.0
+	github.com/spf13/viper v1.10.0
+	github.com/vulcanize/go-eth-state-node-iterator v1.0.0
 )
 
-replace github.com/ethereum/go-ethereum v1.10.9 => github.com/vulcanize/go-ethereum v1.10.11-statediff-0.0.27
+replace github.com/ethereum/go-ethereum v1.10.15 => github.com/vulcanize/go-ethereum v1.10.15-statediff-3.0.1

monitoring/grafana/dashboard.json

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

pkg/api.go

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

pkg/builder.go

@@ -32,9 +32,10 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rlp"
 	sd "github.com/ethereum/go-ethereum/statediff"
-	sdtrie "github.com/ethereum/go-ethereum/statediff/trie"
+	sdtrie "github.com/ethereum/go-ethereum/statediff/trie_helpers"
 	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"
 )
@@ -48,9 +49,9 @@ var (
 
 // Builder interface exposes the method for building a state diff between two blocks
 type Builder interface {
-	BuildStateDiffObject(args sd.Args, params sd.Params) (sd.StateObject, error)
-	BuildStateTrieObject(current *types.Block) (sd.StateObject, error)
-	WriteStateDiffObject(args sd.StateRoots, params sd.Params, output sdtypes.StateNodeSink, codeOutput sdtypes.CodeSink) error
+	BuildStateDiffObject(args sd.Args, params sd.Params) (sdtypes.StateObject, error)
+	BuildStateTrieObject(current *types.Block) (sdtypes.StateObject, error)
+	WriteStateDiffObject(args sdtypes.StateRoots, params sd.Params, output sdtypes.StateNodeSink, codeOutput sdtypes.CodeSink) error
 }
 
 type builder struct {
@@ -119,17 +120,17 @@ func NewBuilder(stateCache state.Database, workers uint) (Builder, error) {
 }
 
 // BuildStateTrieObject builds a state trie object from the provided block
-func (sdb *builder) BuildStateTrieObject(current *types.Block) (sd.StateObject, error) {
+func (sdb *builder) BuildStateTrieObject(current *types.Block) (sdtypes.StateObject, error) {
 	currentTrie, err := sdb.stateCache.OpenTrie(current.Root())
 	if err != nil {
-		return sd.StateObject{}, fmt.Errorf("error creating trie for block %d: %v", current.Number(), err)
+		return sdtypes.StateObject{}, fmt.Errorf("error creating trie for block %d: %v", current.Number(), err)
 	}
 	it := currentTrie.NodeIterator([]byte{})
 	stateNodes, codeAndCodeHashes, err := sdb.buildStateTrie(it)
 	if err != nil {
-		return sd.StateObject{}, fmt.Errorf("error collecting state nodes for block %d: %v", current.Number(), err)
+		return sdtypes.StateObject{}, fmt.Errorf("error collecting state nodes for block %d: %v", current.Number(), err)
 	}
-	return sd.StateObject{
+	return sdtypes.StateObject{
 		BlockNumber:       current.Number(),
 		BlockHash:         current.Hash(),
 		Nodes:             stateNodes,
@@ -189,16 +190,16 @@ func (sdb *builder) buildStateTrie(it trie.NodeIterator) ([]sdtypes.StateNode, [
 }
 
 // BuildStateDiffObject builds a statediff object from two blocks and the provided parameters
-func (sdb *builder) BuildStateDiffObject(args sd.Args, params sd.Params) (sd.StateObject, error) {
+func (sdb *builder) BuildStateDiffObject(args sd.Args, params sd.Params) (sdtypes.StateObject, error) {
 	var stateNodes []sdtypes.StateNode
 	var codeAndCodeHashes []sdtypes.CodeAndCodeHash
 	err := sdb.WriteStateDiffObject(
-		sd.StateRoots{OldStateRoot: args.OldStateRoot, NewStateRoot: args.NewStateRoot},
+		sdtypes.StateRoots{OldStateRoot: args.OldStateRoot, NewStateRoot: args.NewStateRoot},
 		params, stateNodeAppender(&stateNodes), codeMappingAppender(&codeAndCodeHashes))
 	if err != nil {
-		return sd.StateObject{}, err
+		return sdtypes.StateObject{}, err
 	}
-	return sd.StateObject{
+	return sdtypes.StateObject{
 		BlockHash:         args.BlockHash,
 		BlockNumber:       args.BlockNumber,
 		Nodes:             stateNodes,
@@ -206,8 +207,8 @@ func (sdb *builder) BuildStateDiffObject(args sd.Args, params sd.Params) (sd.Sta
 	}, nil
 }
 
-// Writes a statediff object to output callback
-func (sdb *builder) WriteStateDiffObject(args sd.StateRoots, params sd.Params, output sdtypes.StateNodeSink, codeOutput sdtypes.CodeSink) error {
+// WriteStateDiffObject writes a statediff object to output callback
+func (sdb *builder) WriteStateDiffObject(args sdtypes.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
 		log.Info("Ignoring intermediate state nodes because WatchedAddresses was passed")
@@ -234,8 +235,8 @@ func (sdb *builder) WriteStateDiffObject(args sd.StateRoots, params sd.Params, o
 	var iterPairs [][]iterPair
 	for i := uint(0); i < sdb.numWorkers; i++ {
 		iterPairs = append(iterPairs, []iterPair{
-			iterPair{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
-			iterPair{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
+			{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
+			{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
 		})
 	}
 
@@ -252,7 +253,9 @@ func (sdb *builder) WriteStateDiffObject(args sd.StateRoots, params sd.Params, o
 			wg.Add(1)
 			go func(worker uint) {
 				defer wg.Done()
-				sdb.buildStateDiff(iterPairs[worker], params, nodeSender, codeSender)
+				if err := sdb.buildStateDiff(iterPairs[worker], params, nodeSender, codeSender); err != nil {
+					logrus.Errorf("buildStateDiff error for worker %d, pparams %+v", worker, params)
+				}
 			}(w)
 		}
 		wg.Wait()

pkg/builder_test.go

@@ -479,7 +479,7 @@ func TestBuilder(t *testing.T) {
 	var tests = []struct {
 		name              string
 		startingArguments statediff.Args
-		expected          *statediff.StateObject
+		expected          *sdtypes.StateObject
 	}{
 		{
 			"testEmptyDiff",
@@ -489,7 +489,7 @@ func TestBuilder(t *testing.T) {
 				BlockNumber:  block0.Number(),
 				BlockHash:    block0.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block0.Number(),
 				BlockHash:   block0.Hash(),
 				Nodes:       emptyDiffs,
@@ -504,7 +504,7 @@ func TestBuilder(t *testing.T) {
 				BlockNumber:  block0.Number(),
 				BlockHash:    block0.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block0.Number(),
 				BlockHash:   block0.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -527,7 +527,7 @@ func TestBuilder(t *testing.T) {
 				BlockNumber:  block1.Number(),
 				BlockHash:    block1.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block1.Number(),
 				BlockHash:   block1.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -566,7 +566,7 @@ func TestBuilder(t *testing.T) {
 				BlockNumber:  block2.Number(),
 				BlockHash:    block2.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block2.Number(),
 				BlockHash:   block2.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -637,7 +637,7 @@ func TestBuilder(t *testing.T) {
 				BlockNumber:  block3.Number(),
 				BlockHash:    block3.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block3.Number(),
 				BlockHash:   block3.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -716,7 +716,7 @@ func TestBuilderWithIntermediateNodes(t *testing.T) {
 	var tests = []struct {
 		name              string
 		startingArguments statediff.Args
-		expected          *statediff.StateObject
+		expected          *sdtypes.StateObject
 	}{
 		{
 			"testEmptyDiff",
@@ -726,7 +726,7 @@ func TestBuilderWithIntermediateNodes(t *testing.T) {
 				BlockNumber:  block0.Number(),
 				BlockHash:    block0.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block0.Number(),
 				BlockHash:   block0.Hash(),
 				Nodes:       emptyDiffs,
@@ -741,7 +741,7 @@ func TestBuilderWithIntermediateNodes(t *testing.T) {
 				BlockNumber:  block0.Number(),
 				BlockHash:    block0.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block0.Number(),
 				BlockHash:   block0.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -764,7 +764,7 @@ func TestBuilderWithIntermediateNodes(t *testing.T) {
 				BlockNumber:  block1.Number(),
 				BlockHash:    block1.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block1.Number(),
 				BlockHash:   block1.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -809,7 +809,7 @@ func TestBuilderWithIntermediateNodes(t *testing.T) {
 				BlockNumber:  block2.Number(),
 				BlockHash:    block2.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block2.Number(),
 				BlockHash:   block2.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -891,7 +891,7 @@ func TestBuilderWithIntermediateNodes(t *testing.T) {
 				BlockNumber:  block3.Number(),
 				BlockHash:    block3.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block3.Number(),
 				BlockHash:   block3.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -993,7 +993,7 @@ func TestBuilderWithWatchedAddressList(t *testing.T) {
 	var tests = []struct {
 		name              string
 		startingArguments statediff.Args
-		expected          *statediff.StateObject
+		expected          *sdtypes.StateObject
 	}{
 		{
 			"testEmptyDiff",
@@ -1003,7 +1003,7 @@ func TestBuilderWithWatchedAddressList(t *testing.T) {
 				BlockNumber:  block0.Number(),
 				BlockHash:    block0.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block0.Number(),
 				BlockHash:   block0.Hash(),
 				Nodes:       emptyDiffs,
@@ -1018,7 +1018,7 @@ func TestBuilderWithWatchedAddressList(t *testing.T) {
 				BlockNumber:  block0.Number(),
 				BlockHash:    block0.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block0.Number(),
 				BlockHash:   block0.Hash(),
 				Nodes:       emptyDiffs,
@@ -1033,7 +1033,7 @@ func TestBuilderWithWatchedAddressList(t *testing.T) {
 				BlockNumber:  block1.Number(),
 				BlockHash:    block1.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block1.Number(),
 				BlockHash:   block1.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1057,7 +1057,7 @@ func TestBuilderWithWatchedAddressList(t *testing.T) {
 				BlockNumber:  block2.Number(),
 				BlockHash:    block2.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block2.Number(),
 				BlockHash:   block2.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1107,7 +1107,7 @@ func TestBuilderWithWatchedAddressList(t *testing.T) {
 				BlockNumber:  block3.Number(),
 				BlockHash:    block3.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block3.Number(),
 				BlockHash:   block3.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1171,7 +1171,7 @@ func TestBuilderWithWatchedAddressAndStorageKeyList(t *testing.T) {
 	var tests = []struct {
 		name              string
 		startingArguments statediff.Args
-		expected          *statediff.StateObject
+		expected          *sdtypes.StateObject
 	}{
 		{
 			"testEmptyDiff",
@@ -1181,7 +1181,7 @@ func TestBuilderWithWatchedAddressAndStorageKeyList(t *testing.T) {
 				BlockNumber:  block0.Number(),
 				BlockHash:    block0.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block0.Number(),
 				BlockHash:   block0.Hash(),
 				Nodes:       emptyDiffs,
@@ -1196,7 +1196,7 @@ func TestBuilderWithWatchedAddressAndStorageKeyList(t *testing.T) {
 				BlockNumber:  block0.Number(),
 				BlockHash:    block0.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block0.Number(),
 				BlockHash:   block0.Hash(),
 				Nodes:       emptyDiffs,
@@ -1211,7 +1211,7 @@ func TestBuilderWithWatchedAddressAndStorageKeyList(t *testing.T) {
 				BlockNumber:  block1.Number(),
 				BlockHash:    block1.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block1.Number(),
 				BlockHash:   block1.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1235,7 +1235,7 @@ func TestBuilderWithWatchedAddressAndStorageKeyList(t *testing.T) {
 				BlockNumber:  block2.Number(),
 				BlockHash:    block2.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block2.Number(),
 				BlockHash:   block2.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1279,7 +1279,7 @@ func TestBuilderWithWatchedAddressAndStorageKeyList(t *testing.T) {
 				BlockNumber:  block3.Number(),
 				BlockHash:    block3.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block3.Number(),
 				BlockHash:   block3.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1336,7 +1336,7 @@ func TestBuilderWithRemovedAccountAndStorage(t *testing.T) {
 	var tests = []struct {
 		name              string
 		startingArguments statediff.Args
-		expected          *statediff.StateObject
+		expected          *sdtypes.StateObject
 	}{
 		// blocks 0-3 are the same as in TestBuilderWithIntermediateNodes
 		{
@@ -1347,7 +1347,7 @@ func TestBuilderWithRemovedAccountAndStorage(t *testing.T) {
 				BlockNumber:  block4.Number(),
 				BlockHash:    block4.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block4.Number(),
 				BlockHash:   block4.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1413,7 +1413,7 @@ func TestBuilderWithRemovedAccountAndStorage(t *testing.T) {
 				BlockNumber:  block5.Number(),
 				BlockHash:    block5.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block5.Number(),
 				BlockHash:   block5.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1474,7 +1474,7 @@ func TestBuilderWithRemovedAccountAndStorage(t *testing.T) {
 				BlockNumber:  block6.Number(),
 				BlockHash:    block6.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block6.Number(),
 				BlockHash:   block6.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1550,7 +1550,7 @@ func TestBuilderWithRemovedAccountAndStorageWithoutIntermediateNodes(t *testing.
 	var tests = []struct {
 		name              string
 		startingArguments statediff.Args
-		expected          *statediff.StateObject
+		expected          *sdtypes.StateObject
 	}{
 		// blocks 0-3 are the same as in TestBuilderWithIntermediateNodes
 		{
@@ -1561,7 +1561,7 @@ func TestBuilderWithRemovedAccountAndStorageWithoutIntermediateNodes(t *testing.
 				BlockNumber:  block4.Number(),
 				BlockHash:    block4.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block4.Number(),
 				BlockHash:   block4.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1616,7 +1616,7 @@ func TestBuilderWithRemovedAccountAndStorageWithoutIntermediateNodes(t *testing.
 				BlockNumber:  block5.Number(),
 				BlockHash:    block5.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block5.Number(),
 				BlockHash:   block5.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1671,7 +1671,7 @@ func TestBuilderWithRemovedAccountAndStorageWithoutIntermediateNodes(t *testing.
 				BlockNumber:  block6.Number(),
 				BlockHash:    block6.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block6.Number(),
 				BlockHash:   block6.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1821,7 +1821,7 @@ func TestBuilderWithMovedAccount(t *testing.T) {
 	var tests = []struct {
 		name              string
 		startingArguments statediff.Args
-		expected          *statediff.StateObject
+		expected          *sdtypes.StateObject
 	}{
 		{
 			"testBlock1",
@@ -1831,7 +1831,7 @@ func TestBuilderWithMovedAccount(t *testing.T) {
 				BlockNumber:  block1.Number(),
 				BlockHash:    block1.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block1.Number(),
 				BlockHash:   block1.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1890,7 +1890,7 @@ func TestBuilderWithMovedAccount(t *testing.T) {
 				BlockNumber:  block2.Number(),
 				BlockHash:    block2.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block2.Number(),
 				BlockHash:   block2.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -1958,7 +1958,7 @@ func TestBuilderWithMovedAccountOnlyLeafs(t *testing.T) {
 	var tests = []struct {
 		name              string
 		startingArguments statediff.Args
-		expected          *statediff.StateObject
+		expected          *sdtypes.StateObject
 	}{
 		{
 			"testBlock1",
@@ -1968,7 +1968,7 @@ func TestBuilderWithMovedAccountOnlyLeafs(t *testing.T) {
 				BlockNumber:  block1.Number(),
 				BlockHash:    block1.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block1.Number(),
 				BlockHash:   block1.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -2016,7 +2016,7 @@ func TestBuilderWithMovedAccountOnlyLeafs(t *testing.T) {
 				BlockNumber:  block2.Number(),
 				BlockHash:    block2.Hash(),
 			},
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block2.Number(),
 				BlockHash:   block2.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -2080,12 +2080,12 @@ func TestBuildStateTrie(t *testing.T) {
 	var tests = []struct {
 		name     string
 		block    *types.Block
-		expected *statediff.StateObject
+		expected *sdtypes.StateObject
 	}{
 		{
 			"testBlock1",
 			block1,
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block1.Number(),
 				BlockHash:   block1.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -2122,7 +2122,7 @@ func TestBuildStateTrie(t *testing.T) {
 		{
 			"testBlock2",
 			block2,
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block2.Number(),
 				BlockHash:   block2.Hash(),
 				Nodes: []sdtypes.StateNode{
@@ -2197,7 +2197,7 @@ func TestBuildStateTrie(t *testing.T) {
 		{
 			"testBlock3",
 			block3,
-			&statediff.StateObject{
+			&sdtypes.StateObject{
 				BlockNumber: block3.Number(),
 				BlockHash:   block3.Hash(),
 				Nodes: []sdtypes.StateNode{

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/middleware.go

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

pkg/prom/prom.go

@@ -24,11 +24,16 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
-const statsSubsystem = "stats"
+const (
+	statsSubsystem = "stats"
+	subsystemHTTP  = "http"
+	subsystemIPC   = "ipc"
+)
 
 var (
 	metrics bool
 
+	queuedRanges        prometheus.Gauge
 	lastLoadedHeight    prometheus.Gauge
 	lastProcessedHeight prometheus.Gauge
 
@@ -36,9 +41,14 @@ var (
 	tBlockProcessing prometheus.Histogram
 	tStateProcessing prometheus.Histogram
 	tTxCommit        prometheus.Histogram
+
+	httpCount    prometheus.Counter
+	httpDuration prometheus.Histogram
+	ipcCount     prometheus.Gauge
 )
 
 const (
+	RANGES_QUEUED        = "ranges_queued"
 	LOADED_HEIGHT        = "loaded_height"
 	PROCESSED_HEIGHT     = "processed_height"
 	T_BLOCK_LOAD         = "t_block_load"
@@ -51,6 +61,11 @@ const (
 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,
@@ -86,15 +101,48 @@ func Init() {
 		Name:      T_POSTGRES_TX_COMMIT,
 		Help:      "Postgres tx commit time",
 	})
+
+	httpCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: subsystemHTTP,
+		Name:      "count",
+		Help:      "http request count",
+	})
+	httpDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: subsystemHTTP,
+		Name:      "duration",
+		Help:      "http request duration",
+	})
+	ipcCount = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: subsystemIPC,
+		Name:      "count",
+		Help:      "unix socket connection count",
+	})
 }
 
-// RegisterDBCollector create metric colletor for given connection
+// RegisterDBCollector create metric collector for given connection
 func RegisterDBCollector(name string, db *sqlx.DB) {
 	if metrics {
 		prometheus.Register(NewDBStatsCollector(name, db))
 	}
 }
 
+// IncQueuedRanges increments the number of queued range requests
+func IncQueuedRanges() {
+	if metrics {
+		queuedRanges.Inc()
+	}
+}
+
+// DecQueuedRanges decrements the number of queued range requests
+func DecQueuedRanges() {
+	if metrics {
+		queuedRanges.Dec()
+	}
+}
+
 // SetLastLoadedHeight sets last loaded height
 func SetLastLoadedHeight(height int64) {
 	if metrics {

pkg/reader.go

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

pkg/rpc/check.go

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

pkg/rpc/http.go

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

pkg/rpc/ipc.go

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

pkg/service.go

@@ -23,48 +23,44 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/rpc"
 	sd "github.com/ethereum/go-ethereum/statediff"
+	"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
 	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"
+	"github.com/vulcanize/eth-statediff-service/pkg/prom"
 )
 
-// lvlDBReader are the db interfaces required by the statediffing service
-type lvlDBReader interface {
-	GetBlockByHash(hash common.Hash) (*types.Block, error)
-	GetBlockByNumber(number uint64) (*types.Block, error)
-	GetReceiptsByHash(hash common.Hash) (types.Receipts, error)
-	GetTdByHash(hash common.Hash) (*big.Int, error)
-	StateDB() state.Database
-}
+const defaultQueueSize = 1024
 
-// IService is the state-diffing service interface
-type IService interface {
-	// Start() and Stop()
+// StateDiffService is the state-diffing service interface
+type StateDiffService interface {
+	// Lifecycle Start() and Stop()
 	node.Lifecycle
-	// For node service registration
+	// APIs and Protocols() interface for node service registration
 	APIs() []rpc.API
 	Protocols() []p2p.Protocol
-	// Main event loop for processing state diffs
-	Loop(wg *sync.WaitGroup)
-	// Method to get state diff object at specific block
+	// Loop is the main event loop for processing state diffs
+	Loop(wg *sync.WaitGroup) error
+	// Run is a one-off command to run on a predefined set of ranges
+	Run(ranges []RangeRequest) error
+	// StateDiffAt method to get state diff object at specific block
 	StateDiffAt(blockNumber uint64, params sd.Params) (*sd.Payload, error)
-	// Method to get state diff object at specific block
+	// StateDiffFor method to get state diff object at specific block
 	StateDiffFor(blockHash common.Hash, params sd.Params) (*sd.Payload, error)
-	// Method to get state trie object at specific block
+	// StateTrieAt method to get state trie object at specific block
 	StateTrieAt(blockNumber uint64, params sd.Params) (*sd.Payload, error)
-	// Method to write state diff object directly to DB
+	// WriteStateDiffAt method to write state diff object directly to DB
 	WriteStateDiffAt(blockNumber uint64, params sd.Params) error
-	// Method to get state trie object at specific block
+	// WriteStateDiffFor method to get state trie object at specific block
 	WriteStateDiffFor(blockHash common.Hash, params sd.Params) error
+	// WriteStateDiffsInRange method to wrtie state diff objects within the range directly to the DB
+	WriteStateDiffsInRange(start, stop uint64, params sd.Params) error
 }
 
 // Service is the underlying struct for the state diffing service
@@ -72,24 +68,35 @@ type Service struct {
 	// Used to build the state diff objects
 	Builder Builder
 	// Used to read data from leveldb
-	lvlDBReader lvlDBReader
+	lvlDBReader Reader
 	// Used to signal shutdown of the service
-	QuitChan chan bool
+	quitChan chan struct{}
 	// Interface for publishing statediffs as PG-IPLD objects
-	indexer ind.Indexer
+	indexer interfaces.StateDiffIndexer
+	// 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 interfaces.StateDiffIndexer, conf Config) (*Service, error) {
+	b, 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),
+		Builder:     b,
 		indexer:     indexer,
+		workers:     conf.ServiceWorkers,
+		queue:       make(chan RangeRequest, conf.WorkerQueueSize),
+		preruns:     conf.PreRuns,
 	}, nil
 }
 
@@ -110,17 +117,73 @@ func (sds *Service) APIs() []rpc.API {
 	}
 }
 
-// Loop is an empty service loop for awaiting rpc requests
-func (sds *Service) Loop(wg *sync.WaitGroup) {
-	wg.Add(1)
-	for {
-		select {
-		case <-sds.QuitChan:
-			logrus.Info("closing the statediff service loop")
-			wg.Done()
-			return
+// Run does a one-off processing run on the provided RangeRequests + any pre-runs, exiting afterwards
+func (sds *Service) Run(rngs []RangeRequest) error {
+	for _, preRun := range sds.preruns {
+		logrus.Infof("processing prerun range (%d, %d)", preRun.Start, preRun.Stop)
+		for i := preRun.Start; i <= preRun.Stop; i++ {
+			if err := sds.WriteStateDiffAt(i, preRun.Params); err != nil {
+				return fmt.Errorf("error writing statediff at height %d in range (%d, %d) : %v", i, preRun.Start, preRun.Stop, err)
+			}
 		}
 	}
+	sds.preruns = nil
+	for _, rng := range rngs {
+		logrus.Infof("processing prerun range (%d, %d)", rng.Start, rng.Stop)
+		for i := rng.Start; i <= rng.Stop; i++ {
+			if err := sds.WriteStateDiffAt(i, rng.Params); err != nil {
+				return fmt.Errorf("error writing statediff at height %d in range (%d, %d) : %v", i, rng.Start, rng.Stop, err)
+			}
+		}
+	}
+	return nil
+}
+
+// Loop is an empty service loop for awaiting rpc requests
+func (sds *Service) Loop(wg *sync.WaitGroup) error {
+	if sds.quitChan != nil {
+		return fmt.Errorf("service loop is already running")
+	}
+
+	sds.quitChan = make(chan struct{})
+	for i := 0; i < int(sds.workers); i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			for {
+				select {
+				case blockRange := <-sds.queue:
+					logrus.Infof("service worker %d received range (%d, %d) off of work queue, beginning processing", id, blockRange.Start, blockRange.Stop)
+					prom.DecQueuedRanges()
+					for j := blockRange.Start; j <= blockRange.Stop; j++ {
+						if err := sds.WriteStateDiffAt(j, blockRange.Params); err != nil {
+							logrus.Errorf("service worker %d error writing statediff at height %d in range (%d, %d) : %v", id, j, blockRange.Start, blockRange.Stop, err)
+						}
+						select {
+						case <-sds.quitChan:
+							logrus.Infof("closing service worker %d\n"+
+								"working in range (%d, %d)\n"+
+								"last processed height: %d", id, blockRange.Start, blockRange.Stop, j)
+							return
+						default:
+							logrus.Infof("service worker %d finished processing statediff height %d in range (%d, %d)", id, j, blockRange.Start, blockRange.Stop)
+						}
+					}
+					logrus.Infof("service worker %d finished processing range (%d, %d)", id, blockRange.Start, blockRange.Stop)
+				case <-sds.quitChan:
+					logrus.Infof("closing the statediff service loop worker %d", id)
+					return
+				}
+			}
+		}(i)
+	}
+	for _, preRun := range sds.preruns {
+		if err := sds.WriteStateDiffsInRange(preRun.Start, preRun.Stop, preRun.Params); err != nil {
+			close(sds.quitChan)
+			return err
+		}
+	}
+	return nil
 }
 
 // StateDiffAt returns a state diff object payload at the specific blockheight
@@ -130,7 +193,7 @@ func (sds *Service) StateDiffAt(blockNumber uint64, params sd.Params) (*sd.Paylo
 	if err != nil {
 		return nil, err
 	}
-	logrus.Info(fmt.Sprintf("sending state diff at block %d", blockNumber))
+	logrus.Infof("sending state diff at block %d", blockNumber)
 	if blockNumber == 0 {
 		return sds.processStateDiff(currentBlock, common.Hash{}, params)
 	}
@@ -148,7 +211,7 @@ func (sds *Service) StateDiffFor(blockHash common.Hash, params sd.Params) (*sd.P
 	if err != nil {
 		return nil, err
 	}
-	logrus.Info(fmt.Sprintf("sending state diff at block %s", blockHash.Hex()))
+	logrus.Infof("sending state diff at block %s", blockHash.Hex())
 	if currentBlock.NumberU64() == 0 {
 		return sds.processStateDiff(currentBlock, common.Hash{}, params)
 	}
@@ -217,7 +280,7 @@ func (sds *Service) StateTrieAt(blockNumber uint64, params sd.Params) (*sd.Paylo
 	if err != nil {
 		return nil, err
 	}
-	logrus.Info(fmt.Sprintf("sending state trie at block %d", blockNumber))
+	logrus.Infof("sending state trie at block %d", blockNumber)
 	return sds.processStateTrie(currentBlock, params)
 }
 
@@ -237,14 +300,13 @@ func (sds *Service) processStateTrie(block *types.Block, params sd.Params) (*sd.
 // Start is used to begin the service
 func (sds *Service) Start() error {
 	logrus.Info("starting statediff service")
-	go sds.Loop(new(sync.WaitGroup))
-	return nil
+	return sds.Loop(new(sync.WaitGroup))
 }
 
 // Stop is used to close down the service
 func (sds *Service) Stop() error {
 	logrus.Info("stopping statediff service")
-	close(sds.QuitChan)
+	close(sds.quitChan)
 	return nil
 }
 
@@ -252,7 +314,7 @@ func (sds *Service) Stop() error {
 // This operation cannot be performed back past the point of db pruning; it requires an archival node
 // for historical data
 func (sds *Service) WriteStateDiffAt(blockNumber uint64, params sd.Params) error {
-	logrus.Info(fmt.Sprintf("Writing state diff at block %d", blockNumber))
+	logrus.Infof("Writing state diff at block %d", blockNumber)
 	t := time.Now()
 	currentBlock, err := sds.lvlDBReader.GetBlockByNumber(blockNumber)
 	if err != nil {
@@ -273,7 +335,7 @@ func (sds *Service) WriteStateDiffAt(blockNumber uint64, params sd.Params) error
 // This operation cannot be performed back past the point of db pruning; it requires an archival node
 // 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()))
+	logrus.Infof("Writing state diff for block %s", blockHash.Hex())
 	t := time.Now()
 	currentBlock, err := sds.lvlDBReader.GetBlockByHash(blockHash)
 	if err != nil {
@@ -317,21 +379,37 @@ func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, p
 	}
 	// defer handling of commit/rollback for any return case
 	output := func(node sdtypes.StateNode) error {
-		return sds.indexer.PushStateNode(tx, node)
+		return sds.indexer.PushStateNode(tx, node, block.Hash().String())
 	}
 	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{
+	err = sds.Builder.WriteStateDiffObject(sdtypes.StateRoots{
 		NewStateRoot: block.Root(),
 		OldStateRoot: parentRoot,
 	}, params, output, codeOutput)
 	prom.SetTimeMetric(prom.T_STATE_PROCESSING, time.Now().Sub(t))
 	t = time.Now()
-	err = tx.Close(err)
+	err = tx.Submit(err)
 	prom.SetLastProcessedHeight(height)
 	prom.SetTimeMetric(prom.T_POSTGRES_TX_COMMIT, time.Now().Sub(t))
 	return err
 }
+
+// WriteStateDiffsInRange adds a RangeRequest to the work queue
+func (sds *Service) WriteStateDiffsInRange(start, stop uint64, params sd.Params) error {
+	if stop < start {
+		return fmt.Errorf("invalid block range (%d, %d): stop height must be greater or equal to start height", start, stop)
+	}
+	blocked := time.NewTimer(30 * time.Second)
+	select {
+	case sds.queue <- RangeRequest{Start: start, Stop: stop, Params: params}:
+		prom.IncQueuedRanges()
+		logrus.Infof("added range (%d, %d) to the worker queue", start, stop)
+		return nil
+	case <-blocked.C:
+		return fmt.Errorf("unable to add range (%d, %d) to the worker queue", start, stop)
+	}
+}

pkg/types.go

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

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 = 3 // 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)
+}()