Write directly to postgres

README.md

@@ -1,4 +1,4 @@
-## eth-statediff-service
+# eth-statediff-service
 
 [![Go Report Card](https://goreportcard.com/badge/github.com/vulcanize/eth-statediff-service)](https://goreportcard.com/report/github.com/vulcanize/eth-statediff-service)
 
@@ -10,11 +10,31 @@ Stand up a statediffing service directly on top of a go-ethereum leveldb instanc
 This service can serve historical state data over the same rpc interface as
 [statediffing geth](https://github.com/vulcanize/go-ethereum/releases/tag/v1.9.11-statediff-0.0.5) without needing to run a full node
 
-Usage:
+## Usage
 
-`./eth-statediff-service serve --config={path to toml config file}`
+### `serve`
 
-Config:
+To serve state diffs over RPC:
+
+`eth-statediff-service serve --config=<config path>`
+
+Available RPC methods are:
+  * `statediff_stateTrieAt()`
+  * `statediff_streamCodeAndCodeHash()`
+  * `statediff_stateDiffAt()`
+  * `statediff_writeStateDiffAt()`
+
+### `write`
+
+To write state diffs directly to a database:
+
+`eth-statediff-service write --config=<config path>`
+
+This depends on the `database` settings being properly configured.
+
+## Configuration
+
+An example config file:
 
 ```toml
 [leveldb]
@@ -25,10 +45,24 @@ Config:
     ipcPath = "~/.vulcanize/vulcanize.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
+
 [log]
-    file = ""
+    file = "~/.vulcanize/statediff.log"
     level = "info"
 
 [eth]
     chainID = 1
+
 ```

cmd/env.go

@@ -0,0 +1,49 @@
+// VulcanizeDB
+// Copyright © 2019 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 (
+	"github.com/spf13/viper"
+
+	pg "github.com/ethereum/go-ethereum/statediff/indexer/postgres"
+)
+
+const (
+	ETH_NODE_ID       = "ETH_NODE_ID"
+	ETH_CLIENT_NAME   = "ETH_CLIENT_NAME"
+	ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
+	ETH_NETWORK_ID    = "ETH_NETWORK_ID"
+	ETH_CHAIN_ID      = "ETH_CHAIN_ID"
+)
+
+// Bind env vars for eth node and DB configuration
+func init() {
+	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)
+}

cmd/root.go

@@ -21,6 +21,8 @@ import (
 	"os"
 	"strings"
 
+	"github.com/ethereum/go-ethereum/statediff/indexer/node"
+	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -45,7 +47,7 @@ func Execute() {
 }
 
 func initFuncs(cmd *cobra.Command, args []string) {
-	logfile := viper.GetString("logfile")
+	logfile := viper.GetString("log.file")
 	if logfile != "" {
 		file, err := os.OpenFile(logfile,
 			os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
@@ -85,12 +87,41 @@ func init() {
 
 	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("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().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().String("eth-node-id", "", "eth node id")
+	rootCmd.PersistentFlags().String("eth-client-name", "Geth", "eth client name")
+	rootCmd.PersistentFlags().String("eth-genesis-block",
+		"0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3", "eth genesis block hash")
+	rootCmd.PersistentFlags().String("eth-network-id", "1", "eth network id")
+	rootCmd.PersistentFlags().String("eth-chain-id", "1", "eth chain id")
+
 	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("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("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"))
 }
 
 func initConfig() {
@@ -105,3 +136,31 @@ func initConfig() {
 		log.Warn("No config file passed with --config flag")
 	}
 }
+
+func GetEthNodeInfo() node.Info {
+	return node.Info{
+		ID:           viper.GetString("ethereum.nodeID"),
+		ClientName:   viper.GetString("ethereum.clientName"),
+		GenesisBlock: viper.GetString("ethereum.genesisBlock"),
+		NetworkID:    viper.GetString("ethereum.networkID"),
+		ChainID:      viper.GetUint64("ethereum.chainID"),
+	}
+}
+
+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"),
+	}
+}
+
+func GetDBConfig() postgres.ConnectionConfig {
+	return postgres.ConnectionConfig{
+		MaxIdle:     viper.GetInt("database.maxIdle"),
+		MaxOpen:     viper.GetInt("database.maxOpen"),
+		MaxLifetime: viper.GetInt("database.maxLifetime"),
+	}
+}

cmd/serve.go

@@ -21,12 +21,15 @@ import (
 	"os/signal"
 	"sync"
 
+	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
+	ind "github.com/ethereum/go-ethereum/statediff/indexer"
+	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
 	sd "github.com/vulcanize/eth-statediff-service/pkg"
 )
 
@@ -53,58 +56,57 @@ func serve() {
 	if path == "" || ancientPath == "" {
 		logWithCommand.Fatal("require a valid eth leveldb primary datastore path and ancient datastore path")
 	}
-	chainID := viper.GetUint64("eth.chainID")
-	config, err := chainConfig(chainID)
+
+	nodeInfo := GetEthNodeInfo()
+	config, err := chainConfig(nodeInfo.ChainID)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
 	// create leveldb reader
-	logWithCommand.Info("creating leveldb reader")
+	logWithCommand.Info("Creating leveldb reader")
 	lvlDBReader, err := sd.NewLvlDBReader(path, ancientPath, config)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
 	// create statediff service
-	logWithCommand.Info("creating statediff service")
-	statediffService, err := sd.NewStateDiffService(
-		lvlDBReader, sd.Config{Workers: viper.GetUint("statediff.workers")})
+	logWithCommand.Info("Creating statediff service")
+	db, err := postgres.NewDB(postgres.DbConnectionString(GetDBParams()), GetDBConfig(), nodeInfo)
+	if err != nil {
+		logWithCommand.Fatal(err)
+	}
+	indexer := ind.NewStateDiffIndexer(config, db)
+	statediffService, err := sd.NewStateDiffService(lvlDBReader, indexer, viper.GetUint("statediff.workers"))
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
 
 	// start service and servers
-	logWithCommand.Info("starting statediff service")
+	logWithCommand.Info("Starting statediff service")
 	wg := new(sync.WaitGroup)
 	go statediffService.Loop(wg)
-	logWithCommand.Info("starting rpc servers")
+	logWithCommand.Info("Starting RPC servers")
 	if err := startServers(statediffService); err != nil {
 		logWithCommand.Fatal(err)
 	}
-	logWithCommand.Info("rpc servers successfully spun up; awaiting requests")
+	logWithCommand.Info("RPC servers successfully spun up; awaiting requests")
 
 	// clean shutdown
 	shutdown := make(chan os.Signal)
 	signal.Notify(shutdown, os.Interrupt)
 	<-shutdown
-	logWithCommand.Info("received interrupt signal, shutting down")
+	logWithCommand.Info("Received interrupt signal, shutting down")
 	statediffService.Stop()
 	wg.Wait()
 }
 
 func init() {
 	rootCmd.AddCommand(serveCmd)
 
-	serveCmd.PersistentFlags().String("leveldb-path", "", "path to primary datastore")
-	serveCmd.PersistentFlags().String("ancient-path", "", "path to ancient datastore")
-	serveCmd.PersistentFlags().Uint("chain-id", 1, "ethereum chain id (mainnet = 1)")
 	serveCmd.PersistentFlags().String("http-path", "", "vdb server http path")
 	serveCmd.PersistentFlags().String("ipc-path", "", "vdb server ipc path")
 
-	viper.BindPFlag("leveldb.path", serveCmd.PersistentFlags().Lookup("leveldb-path"))
-	viper.BindPFlag("leveldb.ancient", serveCmd.PersistentFlags().Lookup("ancient-path"))
-	viper.BindPFlag("eth.chainID", serveCmd.PersistentFlags().Lookup("chain-id"))
 	viper.BindPFlag("server.httpPath", serveCmd.PersistentFlags().Lookup("http-path"))
 	viper.BindPFlag("server.ipcPath", serveCmd.PersistentFlags().Lookup("ipc-path"))
 }
@@ -126,7 +128,8 @@ func startServers(serv sd.IService) error {
 	}
 	if httpPath != "" {
 		logWithCommand.Info("starting up HTTP server")
-		if _, _, err := rpc.StartHTTPEndpoint(httpPath, serv.APIs(), []string{sd.APIName}, nil, nil, rpc.HTTPTimeouts{}); err != nil {
+		handler := rpc.NewServer()
+		if _, _, err := node.StartHTTPEndpoint(httpPath, rpc.HTTPTimeouts{}, handler); err != nil {
 			return err
 		}
 	}
@@ -138,7 +141,7 @@ func chainConfig(chainID uint64) (*params.ChainConfig, error) {
 	case 1:
 		return params.MainnetChainConfig, nil
 	case 3:
-		return params.TestnetChainConfig, nil // Ropsten
+		return params.RopstenChainConfig, nil // Ropsten
 	case 4:
 		return params.RinkebyChainConfig, nil
 	case 5: