Upgrade geth and add params for CSV file mode (#96)

* Upgrade geth and add params for CSV file mode

* Add options for file mode flag

* index intermediate nodes when watching addresses

README.md

@@ -161,6 +161,8 @@ An example config file:
     type = "postgres"
     driver = "sqlx"
     dumpDestination = ""
+    fileMode = "csv"
+    fileCsvDir = ""
     filePath = ""
 
 [cache]

cmd/env.go

@@ -68,10 +68,12 @@ const (
 	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_TYPE         = "DATABASE_TYPE"
+	DATABASE_DRIVER_TYPE  = "DATABASE_DRIVER_TYPE"
+	DATABASE_DUMP_DST     = "DATABASE_DUMP_DST"
+	DATABASE_FILE_PATH    = "DATABASE_FILE_PATH"
+	DATABASE_FILE_MODE    = "DATABASE_FILE_MODE"
+	DATABASE_FILE_CSV_DIR = "DATABASE_FILE_CSV_DIR"
 
 	DATABASE_MAX_IDLE_CONNECTIONS = "DATABASE_MAX_IDLE_CONNECTIONS"
 	DATABASE_MAX_OPEN_CONNECTIONS = "DATABASE_MAX_OPEN_CONNECTIONS"
@@ -109,7 +111,9 @@ func init() {
 	viper.BindEnv("database.type", DATABASE_TYPE)
 	viper.BindEnv("database.driver", DATABASE_DRIVER_TYPE)
 	viper.BindEnv("database.dumpDestination", DATABASE_DUMP_DST)
+	viper.BindEnv("database.fileMode", DATABASE_FILE_MODE)
 	viper.BindEnv("database.filePath", DATABASE_FILE_PATH)
+	viper.BindEnv("database.fileCsvDir", DATABASE_FILE_CSV_DIR)
 
 	viper.BindEnv("cache.database", DB_CACHE_SIZE_MB)
 	viper.BindEnv("cache.trie", TRIE_CACHE_SIZE_MB)

cmd/root.go

@@ -140,7 +140,9 @@ func init() {
 	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("database-file-mode", "csv", "mode for writing file (for database-type=file; options: csv, sql)")
+	rootCmd.PersistentFlags().String("database-file-csv-dir", "", "full directory path (for database-file-mode=csv)")
+	rootCmd.PersistentFlags().String("database-file-path", "", "full file path (for database-file-mode=sql)")
 
 	rootCmd.PersistentFlags().String("eth-node-id", "", "eth node id")
 	rootCmd.PersistentFlags().String("eth-client-name", "eth-statediff-service", "eth client name")
@@ -198,6 +200,8 @@ func init() {
 	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.fileMode", rootCmd.PersistentFlags().Lookup("database-file-mode"))
+	viper.BindPFlag("database.fileCsvDir", rootCmd.PersistentFlags().Lookup("database-file-csv-dir"))
 	viper.BindPFlag("database.filePath", rootCmd.PersistentFlags().Lookup("database-file-path"))
 
 	viper.BindPFlag("ethereum.nodeID", rootCmd.PersistentFlags().Lookup("eth-node-id"))
@@ -302,11 +306,28 @@ func getConfig(nodeInfo node.Info) (interfaces.Config, error) {
 	switch dbType {
 	case shared.FILE:
 		logWithCommand.Info("starting in sql file writing mode")
+
+		fileModeStr := viper.GetString("database.fileMode")
+		fileMode, err := file.ResolveFileMode(fileModeStr)
+		if err != nil {
+			utils.Fatalf("%v", err)
+		}
+
 		filePathStr := viper.GetString("database.filePath")
-		if filePathStr == "" {
+		if fileMode == file.SQL && filePathStr == "" {
 			logWithCommand.Fatal("when operating in sql file writing mode a file path must be provided")
 		}
-		indexerConfig = file.Config{FilePath: filePathStr}
+
+		fileCsvDirStr := viper.GetString("database.fileCsvDir")
+		if fileMode == file.CSV && fileCsvDirStr == "" {
+			logWithCommand.Fatal("when operating in csv file writing mode a directory path must be provided")
+		}
+
+		indexerConfig = file.Config{
+			Mode:      fileMode,
+			OutputDir: fileCsvDirStr,
+			FilePath:  filePathStr,
+		}
 	case shared.DUMP:
 		logWithCommand.Info("starting in data dump mode")
 		dumpDstStr := viper.GetString("database.dumpDestination")

environments/config.toml

@@ -39,6 +39,8 @@
     type = "postgres"
     driver = "sqlx"
     dumpDestination = ""
+    fileMode = "csv"
+    fileCsvDir = ""
     filePath = ""
 
 [cache]

go.mod

@@ -128,4 +128,4 @@ require (
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 )
 
-replace github.com/ethereum/go-ethereum v1.10.19 => github.com/vulcanize/go-ethereum v1.10.19-statediff-4.0.3-alpha
+replace github.com/ethereum/go-ethereum v1.10.19 => github.com/vulcanize/go-ethereum v1.10.19-statediff-4.0.4-alpha

go.sum

@@ -727,8 +727,8 @@ github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPU
 github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
 github.com/vulcanize/go-eth-state-node-iterator v1.1.1 h1:/9zKsUzF1zlQDriAljdspmp1ixcIpOcbetiyHCPhbU0=
 github.com/vulcanize/go-eth-state-node-iterator v1.1.1/go.mod h1:pL9jDztI5Yv9OOCKpCyo08/v+27othmbQlggoqH84Zo=
-github.com/vulcanize/go-ethereum v1.10.19-statediff-4.0.3-alpha h1:Pr6mSkWM7bobRJ/GzeztH+EOQR5IVLScO57G+N/q20g=
-github.com/vulcanize/go-ethereum v1.10.19-statediff-4.0.3-alpha/go.mod h1:5tMN+CDbK/qI2UlfN307HJykDmVIOCB1FM5RcHK9Kp8=
+github.com/vulcanize/go-ethereum v1.10.19-statediff-4.0.4-alpha h1:fcUztJhvVHytSMwFzjoVR1RIPDkuAoZG8uqQUb7wBMg=
+github.com/vulcanize/go-ethereum v1.10.19-statediff-4.0.4-alpha/go.mod h1:5tMN+CDbK/qI2UlfN307HJykDmVIOCB1FM5RcHK9Kp8=
 github.com/vulcanize/leveldb-ethdb-rpc v0.1.3 h1:R0+2GNMQKPXsKVWjTc6h5khD/aZzSarxxVsPEtBHiHU=
 github.com/vulcanize/leveldb-ethdb-rpc v0.1.3/go.mod h1:A3aI/tDglAHGpB9/vmtwS/mWOvfGl1UFJ1jCbhIPMAM=
 github.com/whyrusleeping/go-logging v0.0.0-20170515211332-0457bb6b88fc h1:9lDbC6Rz4bwmou+oE6Dt4Cb2BGMur5eR/GYptkKUVHo=

pkg/builder.go

@@ -114,7 +114,7 @@ func (sdb *builder) WriteStateDiffObject(args sdtypes.StateRoots, params sd.Para
 			go func(worker uint) {
 				defer wg.Done()
 				var err error
-				if !params.IntermediateStateNodes || len(params.WatchedAddresses) > 0 {
+				if !params.IntermediateStateNodes {
 					// if we are watching only specific accounts then we are only diffing leaf nodes
 					err = sdb.BuildStateDiffWithoutIntermediateStateNodes(iterPairs[worker], params, nodeSender, codeSender)
 				} else {

pkg/builder_test.go

@@ -1001,7 +1001,7 @@ func TestBuilderWithWatchedAddressList(t *testing.T) {
 	params := sd.Params{
 		WatchedAddresses: []common.Address{test_helpers.Account1Addr, test_helpers.ContractAddr},
 	}
-	params.ComputeWatchedAddressesLeafKeys()
+	params.ComputeWatchedAddressesLeafPaths()
 
 	var tests = []struct {
 		name              string
@@ -1613,7 +1613,7 @@ func TestBuilderWithRemovedNonWatchedAccount(t *testing.T) {
 	params := sd.Params{
 		WatchedAddresses: []common.Address{test_helpers.Account1Addr, test_helpers.Account2Addr},
 	}
-	params.ComputeWatchedAddressesLeafKeys()
+	params.ComputeWatchedAddressesLeafPaths()
 
 	var tests = []struct {
 		name              string
@@ -1731,7 +1731,7 @@ func TestBuilderWithRemovedWatchedAccount(t *testing.T) {
 	params := sd.Params{
 		WatchedAddresses: []common.Address{test_helpers.Account1Addr, test_helpers.ContractAddr},
 	}
-	params.ComputeWatchedAddressesLeafKeys()
+	params.ComputeWatchedAddressesLeafPaths()
 
 	var tests = []struct {
 		name              string

pkg/service.go

@@ -196,7 +196,7 @@ func (sds *Service) StateDiffAt(blockNumber uint64, params sd.Params) (*sd.Paylo
 	logrus.Infof("sending state diff at block %d", blockNumber)
 
 	// compute leaf keys of watched addresses in the params
-	params.ComputeWatchedAddressesLeafKeys()
+	params.ComputeWatchedAddressesLeafPaths()
 
 	if blockNumber == 0 {
 		return sds.processStateDiff(currentBlock, common.Hash{}, params)
@@ -218,7 +218,7 @@ func (sds *Service) StateDiffFor(blockHash common.Hash, params sd.Params) (*sd.P
 	logrus.Infof("sending state diff at block %s", blockHash.Hex())
 
 	// compute leaf keys of watched addresses in the params
-	params.ComputeWatchedAddressesLeafKeys()
+	params.ComputeWatchedAddressesLeafPaths()
 
 	if currentBlock.NumberU64() == 0 {
 		return sds.processStateDiff(currentBlock, common.Hash{}, params)
@@ -291,7 +291,7 @@ func (sds *Service) StateTrieAt(blockNumber uint64, params sd.Params) (*sd.Paylo
 	logrus.Infof("sending state trie at block %d", blockNumber)
 
 	// compute leaf keys of watched addresses in the params
-	params.ComputeWatchedAddressesLeafKeys()
+	params.ComputeWatchedAddressesLeafPaths()
 
 	return sds.processStateTrie(currentBlock, params)
 }
@@ -334,7 +334,7 @@ func (sds *Service) WriteStateDiffAt(blockNumber uint64, params sd.Params) error
 	}
 
 	// compute leaf keys of watched addresses in the params
-	params.ComputeWatchedAddressesLeafKeys()
+	params.ComputeWatchedAddressesLeafPaths()
 
 	parentRoot := common.Hash{}
 	if blockNumber != 0 {
@@ -359,7 +359,7 @@ func (sds *Service) WriteStateDiffFor(blockHash common.Hash, params sd.Params) e
 	}
 
 	// compute leaf keys of watched addresses in the params
-	params.ComputeWatchedAddressesLeafKeys()
+	params.ComputeWatchedAddressesLeafPaths()
 
 	parentRoot := common.Hash{}
 	if currentBlock.NumberU64() != 0 {