Upgrade geth (#51 )

* Fix Makefile for installing mockgen tool * Upgrade geth * Add sleep in the failing recovery test Co-authored-by: nabarun <nabarun@deepstacksoft.com>
Add command to take an in-place snapshot (#42 )
2022-07-13 11:39:36 +05:30 · 2022-06-15 12:51:26 +05:30 · 2022-06-10 14:24:33 +05:30 · 2022-06-10 14:12:34 +05:30 · 2022-06-10 14:03:43 +05:30 · 2022-06-10 14:02:13 +05:30
29 changed files with 1686 additions and 482 deletions
@@ -0,0 +1,30 @@
+name: Docker Build
+
+on: [pull_request]
+
+jobs:
+  test:
+    name: Run unit tests
+    runs-on: ubuntu-latest
+    env:
+      GOPATH: /tmp/go
+      GO111MODULE: on
+    steps:
+      - name: Create GOPATH
+        run: mkdir -p /tmp/go
+
+      - uses: actions/setup-go@v3
+        with:
+          go-version: ">=1.18.0"
+          check-latest: true
+
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      - name: Run database
+        run: docker-compose up -d
+
+      - name: Run unit tests
+        run: |
+          sleep 45
+          make dbtest
@@ -1,2 +1,5 @@
 .idea/
+.vscode/
 ipld-eth-state-snapshot
+mocks/
+.vscode
@@ -1,12 +1,28 @@
+BIN = $(GOPATH)/bin
+
+## Mockgen tool
+MOCKGEN = $(BIN)/mockgen
+$(BIN)/mockgen:
+	go install github.com/golang/mock/mockgen@v1.6.0
+
 MOCKS_DIR = $(CURDIR)/mocks
-mockgen_cmd=mockgen

-.PHONY: mocks
+.PHONY: mocks test

-mocks: mocks/snapshot/publisher.go
+mocks: $(MOCKGEN) mocks/snapshot/publisher.go

 mocks/snapshot/publisher.go: pkg/types/publisher.go
-	$(mockgen_cmd) -package snapshot_mock -destination $@ -source $< Publisher Tx
+	$(MOCKGEN) -package snapshot_mock -destination $@ -source $< Publisher Tx

 clean:
 	rm -f mocks/snapshot/publisher.go
+
+build:
+	go fmt ./...
+	go build
+
+test: mocks
+	go clean -testcache && go test -p 1 -v ./...
+
+dbtest: mocks
+	go clean -testcache && TEST_WITH_DB=true go test -p 1 -v ./...
@@ -4,9 +4,17 @@

 [![Go Report Card](https://goreportcard.com/badge/github.com/vulcanize/ipld-eth-state-snapshot)](https://goreportcard.com/report/github.com/vulcanize/ipld-eth-state-snapshot)

-## Usage 
+## Usage

+For state snapshot from LevelDB
+```bash
 ./ipld-eth-state-snapshot stateSnapshot --config={path to toml config file}
+```
+
+For in-place snapshot in database
+```bash
+./ipld-eth-state-snapshot inPlaceStateSnapshot --config={path to toml config file}
+```

 ### Config

@@ -32,4 +40,28 @@ Config format:

 [file]
    outputDir = "output_dir/" # when operating in 'file' output mode, this is the directory the files are written to
+
+[log]
+    level = "info" # log level (trace, debug, info, warn, error, fatal, panic) (default: info)
+    file = "log_file" # file path for logging
+
+[prom]
+    metrics = true # enable prometheus metrics (default: false)
+    http = true # enable prometheus http service (default: false)
+    httpAddr = "0.0.0.0" # prometheus http host (default: 127.0.0.1)
+    httpPort = 9101 # prometheus http port (default: 8086)
+    dbStats = true # enable prometheus db stats (default: false)
+
+# node info
+[ethereum]
+    clientName = "Geth" # $ETH_CLIENT_NAME
+    nodeID = "arch1" # $ETH_NODE_ID
+    networkID = "1" # $ETH_NETWORK_ID
+    chainID = "1" # $ETH_CHAIN_ID
+    genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3" # $ETH_GENESIS_BLOCK
 ```
+
+## Tests
+
+* Install [mockgen](https://github.com/golang/mock#installation)
+* `make test`
@@ -0,0 +1,63 @@
+// VulcanizeDB
+// Copyright © 2022 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/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot"
+)
+
+// inPlaceStateSnapshotCmd represents the inPlaceStateSnapshot command
+var inPlaceStateSnapshotCmd = &cobra.Command{
+	Use:   "inPlaceStateSnapshot",
+	Short: "Take an in-place state snapshot in the database",
+	Long: `Usage:
+
+	./ipld-eth-state-snapshot inPlaceStateSnapshot --config={path to toml config file}`,
+	Run: func(cmd *cobra.Command, args []string) {
+		subCommand = cmd.CalledAs()
+		logWithCommand = *logrus.WithField("SubCommand", subCommand)
+		inPlaceStateSnapshot()
+	},
+}
+
+func inPlaceStateSnapshot() {
+	config := snapshot.NewInPlaceSnapshotConfig()
+
+	startHeight := viper.GetUint64(snapshot.SNAPSHOT_START_HEIGHT_TOML)
+	endHeight := viper.GetUint64(snapshot.SNAPSHOT_END_HEIGHT_TOML)
+
+	params := snapshot.InPlaceSnapshotParams{StartHeight: uint64(startHeight), EndHeight: uint64(endHeight)}
+	if err := snapshot.CreateInPlaceSnapshot(config, params); err != nil {
+		logWithCommand.Fatal(err)
+	}
+
+	logWithCommand.Infof("snapshot taken at height %d starting from height %d", endHeight, startHeight)
+}
+
+func init() {
+	rootCmd.AddCommand(inPlaceStateSnapshotCmd)
+
+	inPlaceStateSnapshotCmd.PersistentFlags().String(snapshot.SNAPSHOT_START_HEIGHT_CLI, "", "start block height for in-place snapshot")
+	inPlaceStateSnapshotCmd.PersistentFlags().String(snapshot.SNAPSHOT_END_HEIGHT_CLI, "", "end block height for in-place snapshot")
+
+	viper.BindPFlag(snapshot.SNAPSHOT_START_HEIGHT_TOML, inPlaceStateSnapshotCmd.PersistentFlags().Lookup(snapshot.SNAPSHOT_START_HEIGHT_CLI))
+	viper.BindPFlag(snapshot.SNAPSHOT_END_HEIGHT_TOML, inPlaceStateSnapshotCmd.PersistentFlags().Lookup(snapshot.SNAPSHOT_END_HEIGHT_CLI))
+}
@@ -24,6 +24,9 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/prom"
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot"
 )

 var (
@@ -46,7 +49,7 @@ func Execute() {
 }

 func initFuncs(cmd *cobra.Command, args []string) {
-	logfile := viper.GetString("logfile")
+	logfile := viper.GetString(snapshot.LOGRUS_FILE_TOML)
 	if logfile != "" {
 		file, err := os.OpenFile(logfile,
 			os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
@@ -63,10 +66,25 @@ func initFuncs(cmd *cobra.Command, args []string) {
 	if err := logLevel(); err != nil {
 		log.Fatal("Could not set log level: ", err)
 	}
+
+	if viper.GetBool(snapshot.PROM_METRICS_TOML) {
+		log.Info("initializing prometheus metrics")
+		prom.Init()
+	}
+
+	if viper.GetBool(snapshot.PROM_HTTP_TOML) {
+		addr := fmt.Sprintf(
+			"%s:%s",
+			viper.GetString(snapshot.PROM_HTTP_ADDR_TOML),
+			viper.GetString(snapshot.PROM_HTTP_PORT_TOML),
+		)
+		log.Info("starting prometheus server")
+		prom.Serve(addr)
+	}
 }

 func logLevel() error {
-	lvl, err := log.ParseLevel(viper.GetString("log.level"))
+	lvl, err := log.ParseLevel(viper.GetString(snapshot.LOGRUS_LEVEL_TOML))
 	if err != nil {
 		return err
 	}
@@ -75,6 +93,7 @@ func logLevel() error {
 		log.SetReportCaller(true)
 	}
 	log.Info("Log level set to ", lvl.String())
+
 	return nil
 }

@@ -84,21 +103,33 @@ func init() {
 	viper.AutomaticEnv()

 	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file location")
-	rootCmd.PersistentFlags().String("logfile", "", "file path for logging")
-	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("log-level", log.InfoLevel.String(), "Log level (trace, debug, info, warn, error, fatal, panic")
+	rootCmd.PersistentFlags().String(snapshot.LOGRUS_FILE_CLI, "", "file path for logging")
+	rootCmd.PersistentFlags().String(snapshot.DATABASE_NAME_CLI, "vulcanize_public", "database name")
+	rootCmd.PersistentFlags().Int(snapshot.DATABASE_PORT_CLI, 5432, "database port")
+	rootCmd.PersistentFlags().String(snapshot.DATABASE_HOSTNAME_CLI, "localhost", "database hostname")
+	rootCmd.PersistentFlags().String(snapshot.DATABASE_USER_CLI, "", "database user")
+	rootCmd.PersistentFlags().String(snapshot.DATABASE_PASSWORD_CLI, "", "database password")
+	rootCmd.PersistentFlags().String(snapshot.LOGRUS_LEVEL_CLI, log.InfoLevel.String(), "log level (trace, debug, info, warn, error, fatal, panic)")

-	viper.BindPFlag("logfile", rootCmd.PersistentFlags().Lookup("logfile"))
-	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("log.level", rootCmd.PersistentFlags().Lookup("log-level"))
+	rootCmd.PersistentFlags().Bool(snapshot.PROM_METRICS_CLI, false, "enable prometheus metrics")
+	rootCmd.PersistentFlags().Bool(snapshot.PROM_HTTP_CLI, false, "enable prometheus http service")
+	rootCmd.PersistentFlags().String(snapshot.PROM_HTTP_ADDR_CLI, "127.0.0.1", "prometheus http host")
+	rootCmd.PersistentFlags().String(snapshot.PROM_HTTP_PORT_CLI, "8086", "prometheus http port")
+	rootCmd.PersistentFlags().Bool(snapshot.PROM_DB_STATS_CLI, false, "enables prometheus db stats")
+
+	viper.BindPFlag(snapshot.LOGRUS_FILE_TOML, rootCmd.PersistentFlags().Lookup(snapshot.LOGRUS_FILE_CLI))
+	viper.BindPFlag(snapshot.DATABASE_NAME_TOML, rootCmd.PersistentFlags().Lookup(snapshot.DATABASE_NAME_CLI))
+	viper.BindPFlag(snapshot.DATABASE_PORT_TOML, rootCmd.PersistentFlags().Lookup(snapshot.DATABASE_PORT_CLI))
+	viper.BindPFlag(snapshot.DATABASE_HOSTNAME_TOML, rootCmd.PersistentFlags().Lookup(snapshot.DATABASE_HOSTNAME_CLI))
+	viper.BindPFlag(snapshot.DATABASE_USER_TOML, rootCmd.PersistentFlags().Lookup(snapshot.DATABASE_USER_CLI))
+	viper.BindPFlag(snapshot.DATABASE_PASSWORD_TOML, rootCmd.PersistentFlags().Lookup(snapshot.DATABASE_PASSWORD_CLI))
+	viper.BindPFlag(snapshot.LOGRUS_LEVEL_TOML, rootCmd.PersistentFlags().Lookup(snapshot.LOGRUS_LEVEL_CLI))
+
+	viper.BindPFlag(snapshot.PROM_METRICS_TOML, rootCmd.PersistentFlags().Lookup(snapshot.PROM_METRICS_CLI))
+	viper.BindPFlag(snapshot.PROM_HTTP_TOML, rootCmd.PersistentFlags().Lookup(snapshot.PROM_HTTP_CLI))
+	viper.BindPFlag(snapshot.PROM_HTTP_ADDR_TOML, rootCmd.PersistentFlags().Lookup(snapshot.PROM_HTTP_ADDR_CLI))
+	viper.BindPFlag(snapshot.PROM_HTTP_PORT_TOML, rootCmd.PersistentFlags().Lookup(snapshot.PROM_HTTP_PORT_CLI))
+	viper.BindPFlag(snapshot.PROM_DB_STATS_TOML, rootCmd.PersistentFlags().Lookup(snapshot.PROM_DB_STATS_CLI))
 }

 func initConfig() {
@@ -40,11 +40,11 @@ var stateSnapshotCmd = &cobra.Command{
 }

 func stateSnapshot() {
-	modeStr := viper.GetString("snapshot.mode")
+	modeStr := viper.GetString(snapshot.SNAPSHOT_MODE_TOML)
 	mode := snapshot.SnapshotMode(modeStr)
 	config, err := snapshot.NewConfig(mode)
 	if err != nil {
-		logWithCommand.Fatal("unable to initialize config: %v", err)
+		logWithCommand.Fatalf("unable to initialize config: %v", err)
 	}
 	logWithCommand.Infof("opening levelDB and ancient data at %s and %s",
 		config.Eth.LevelDBPath, config.Eth.AncientDBPath)
@@ -52,11 +52,11 @@ func stateSnapshot() {
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
-	height := viper.GetInt64("snapshot.blockHeight")
-	recoveryFile := viper.GetString("snapshot.recoveryFile")
+	height := viper.GetInt64(snapshot.SNAPSHOT_BLOCK_HEIGHT_TOML)
+	recoveryFile := viper.GetString(snapshot.SNAPSHOT_RECOVERY_FILE_TOML)
 	if recoveryFile == "" {
 		recoveryFile = fmt.Sprintf("./%d_snapshot_recovery", height)
-		logWithCommand.Infof("no recovery file set, creating default: %s", recoveryFile)
+		logWithCommand.Infof("no recovery file set, using default: %s", recoveryFile)
 	}

 	pub, err := snapshot.NewPublisher(mode, config)
@@ -68,7 +68,7 @@ func stateSnapshot() {
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
-	workers := viper.GetUint("snapshot.workers")
+	workers := viper.GetUint(snapshot.SNAPSHOT_WORKERS_TOML)

 	if height < 0 {
 		if err := snapshotService.CreateLatestSnapshot(workers); err != nil {
@@ -86,19 +86,19 @@ func stateSnapshot() {
 func init() {
 	rootCmd.AddCommand(stateSnapshotCmd)

-	stateSnapshotCmd.PersistentFlags().String("leveldb-path", "", "path to primary datastore")
-	stateSnapshotCmd.PersistentFlags().String("ancient-path", "", "path to ancient datastore")
-	stateSnapshotCmd.PersistentFlags().String("block-height", "", "blockheight to extract state at")
-	stateSnapshotCmd.PersistentFlags().Int("workers", 1, "number of concurrent workers to use")
-	stateSnapshotCmd.PersistentFlags().String("recovery-file", "", "file to recover from a previous iteration")
-	stateSnapshotCmd.PersistentFlags().String("snapshot-mode", "postgres", "output mode for snapshot ('file' or 'postgres')")
-	stateSnapshotCmd.PersistentFlags().String("output-dir", "", "directory for writing ouput to while operating in 'file' mode")
+	stateSnapshotCmd.PersistentFlags().String(snapshot.LVL_DB_PATH_CLI, "", "path to primary datastore")
+	stateSnapshotCmd.PersistentFlags().String(snapshot.ANCIENT_DB_PATH_CLI, "", "path to ancient datastore")
+	stateSnapshotCmd.PersistentFlags().String(snapshot.SNAPSHOT_BLOCK_HEIGHT_CLI, "", "block height to extract state at")
+	stateSnapshotCmd.PersistentFlags().Int(snapshot.SNAPSHOT_WORKERS_CLI, 1, "number of concurrent workers to use")
+	stateSnapshotCmd.PersistentFlags().String(snapshot.SNAPSHOT_RECOVERY_FILE_CLI, "", "file to recover from a previous iteration")
+	stateSnapshotCmd.PersistentFlags().String(snapshot.SNAPSHOT_MODE_CLI, "postgres", "output mode for snapshot ('file' or 'postgres')")
+	stateSnapshotCmd.PersistentFlags().String(snapshot.FILE_OUTPUT_DIR_CLI, "", "directory for writing ouput to while operating in 'file' mode")

-	viper.BindPFlag("leveldb.path", stateSnapshotCmd.PersistentFlags().Lookup("leveldb-path"))
-	viper.BindPFlag("leveldb.ancient", stateSnapshotCmd.PersistentFlags().Lookup("ancient-path"))
-	viper.BindPFlag("snapshot.blockHeight", stateSnapshotCmd.PersistentFlags().Lookup("block-height"))
-	viper.BindPFlag("snapshot.workers", stateSnapshotCmd.PersistentFlags().Lookup("workers"))
-	viper.BindPFlag("snapshot.recoveryFile", stateSnapshotCmd.PersistentFlags().Lookup("recovery-file"))
-	viper.BindPFlag("snapshot.mode", stateSnapshotCmd.PersistentFlags().Lookup("snapshot-mode"))
-	viper.BindPFlag("file.outputDir", stateSnapshotCmd.PersistentFlags().Lookup("output-dir"))
+	viper.BindPFlag(snapshot.LVL_DB_PATH_TOML, stateSnapshotCmd.PersistentFlags().Lookup(snapshot.LVL_DB_PATH_CLI))
+	viper.BindPFlag(snapshot.ANCIENT_DB_PATH_TOML, stateSnapshotCmd.PersistentFlags().Lookup(snapshot.ANCIENT_DB_PATH_CLI))
+	viper.BindPFlag(snapshot.SNAPSHOT_BLOCK_HEIGHT_TOML, stateSnapshotCmd.PersistentFlags().Lookup(snapshot.SNAPSHOT_BLOCK_HEIGHT_CLI))
+	viper.BindPFlag(snapshot.SNAPSHOT_WORKERS_TOML, stateSnapshotCmd.PersistentFlags().Lookup(snapshot.SNAPSHOT_WORKERS_CLI))
+	viper.BindPFlag(snapshot.SNAPSHOT_RECOVERY_FILE_TOML, stateSnapshotCmd.PersistentFlags().Lookup(snapshot.SNAPSHOT_RECOVERY_FILE_CLI))
+	viper.BindPFlag(snapshot.SNAPSHOT_MODE_TOML, stateSnapshotCmd.PersistentFlags().Lookup(snapshot.SNAPSHOT_MODE_CLI))
+	viper.BindPFlag(snapshot.FILE_OUTPUT_DIR_TOML, stateSnapshotCmd.PersistentFlags().Lookup(snapshot.FILE_OUTPUT_DIR_CLI))
 }
@@ -0,0 +1,27 @@
+version: '3.2'
+
+services:
+  migrations:
+    restart: on-failure
+    depends_on:
+      - ipld-eth-db
+    image: vulcanize/ipld-eth-db:v4.2.0-alpha
+    environment:
+      DATABASE_USER: "vdbm"
+      DATABASE_NAME: "vulcanize_testing"
+      DATABASE_PASSWORD: "password"
+      DATABASE_HOSTNAME: "ipld-eth-db"
+      DATABASE_PORT: 5432
+
+  ipld-eth-db:
+    image: timescale/timescaledb:latest-pg14
+    restart: always
+    command: ["postgres", "-c", "log_statement=all"]
+    environment:
+      POSTGRES_USER: "vdbm"
+      POSTGRES_DB: "vulcanize_testing"
+      POSTGRES_PASSWORD: "password"
+    ports:
+      - "127.0.0.1:8077:5432"
+    volumes:
+      - /tmp:/tmp
@@ -5,8 +5,19 @@
    user     = "postgres"

 [leveldb]
-    path = "/Users/user/Library/Ethereum/geth/chaindata"
-    ancient = "/Users/user/Library/Ethereum/geth/chaindata/ancient"
+    path = "/Users/iannorden/Library/Ethereum/geth/chaindata"
+    ancient = "/Users/iannorden/Library/Ethereum/geth/chaindata/ancient"
+
+[log]
+    level = "info"
+    file = "log_file"
+
+[prom]
+    metrics = true
+    http = true
+    httpAddr = "0.0.0.0"
+    httpPort = 9101
+    dbStats = true

 [snapshot]
    mode = "file"
@@ -14,5 +25,15 @@
    blockHeight = -1
    recoveryFile = "recovery_file"

+    startHeight = 1
+    endHeight = 12
+
 [file]
    outputDir = "output_dir/"
+
+[ethereum]
+    clientName = "Geth" # $ETH_CLIENT_NAME
+    nodeID = "arch1" # $ETH_NODE_ID
+    networkID = "1" # $ETH_NETWORK_ID
+    chainID = "1" # $ETH_CHAIN_ID
+    genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3" # $ETH_GENESIS_BLOCK
@@ -0,0 +1,307 @@
+// Copyright © 2022 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 fixture
+
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
+)
+
+type Block struct {
+	Hash         common.Hash
+	Number       *big.Int
+	StateNodes   []snapt.Node
+	StorageNodes [][]snapt.Node
+}
+
+var InPlaceSnapshotBlocks = []Block{
+	// Genesis block
+	{
+		Hash:   common.HexToHash("0xe1bdb963128f645aa674b52a8c7ce00704762f27e2a6896abebd7954878f40e4"),
+		Number: big.NewInt(0),
+		StateNodes: []snapt.Node{
+			// State node for main account with balance.
+			{
+				NodeType: 2,
+				Path:     []byte{},
+				Key:      common.HexToHash("0x67ab3c0dd775f448af7fb41243415ed6fb975d1530a2d828f69bea7346231ad7"),
+				Value:    []byte{248, 119, 161, 32, 103, 171, 60, 13, 215, 117, 244, 72, 175, 127, 180, 18, 67, 65, 94, 214, 251, 151, 93, 21, 48, 162, 216, 40, 246, 155, 234, 115, 70, 35, 26, 215, 184, 83, 248, 81, 128, 141, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 160, 86, 232, 31, 23, 27, 204, 85, 166, 255, 131, 69, 230, 146, 192, 248, 110, 91, 72, 224, 27, 153, 108, 173, 192, 1, 98, 47, 181, 227, 99, 180, 33, 160, 197, 210, 70, 1, 134, 247, 35, 60, 146, 126, 125, 178, 220, 199, 3, 192, 229, 0, 182, 83, 202, 130, 39, 59, 123, 250, 216, 4, 93, 133, 164, 112},
+			},
+		},
+	},
+	// Contract Test1 deployed by main account.
+	{
+		Hash:   common.HexToHash("0x46ce57b700e470d0c0820ede662ecc0d0c78cf87237cb12a40a7ff5ff9cc8ac5"),
+		Number: big.NewInt(1),
+		StateNodes: []snapt.Node{
+			// Branch root node.
+			{
+				NodeType: 0,
+				Path:     []byte{},
+				Value:    []byte{248, 81, 128, 128, 128, 128, 128, 128, 160, 173, 52, 73, 195, 118, 160, 81, 100, 138, 50, 127, 27, 188, 85, 147, 215, 187, 244, 219, 228, 93, 25, 72, 253, 160, 45, 16, 239, 130, 223, 160, 26, 128, 128, 160, 137, 52, 229, 60, 211, 96, 171, 177, 51, 19, 204, 180, 24, 252, 28, 70, 234, 7, 73, 20, 117, 230, 32, 223, 188, 6, 191, 75, 123, 64, 163, 197, 128, 128, 128, 128, 128, 128, 128},
+			},
+			// State node for sender account.
+			{
+				NodeType: 2,
+				Path:     []byte{6},
+				Key:      common.HexToHash("0x67ab3c0dd775f448af7fb41243415ed6fb975d1530a2d828f69bea7346231ad7"),
+				Value:    []byte{248, 118, 160, 55, 171, 60, 13, 215, 117, 244, 72, 175, 127, 180, 18, 67, 65, 94, 214, 251, 151, 93, 21, 48, 162, 216, 40, 246, 155, 234, 115, 70, 35, 26, 215, 184, 83, 248, 81, 1, 141, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 160, 86, 232, 31, 23, 27, 204, 85, 166, 255, 131, 69, 230, 146, 192, 248, 110, 91, 72, 224, 27, 153, 108, 173, 192, 1, 98, 47, 181, 227, 99, 180, 33, 160, 197, 210, 70, 1, 134, 247, 35, 60, 146, 126, 125, 178, 220, 199, 3, 192, 229, 0, 182, 83, 202, 130, 39, 59, 123, 250, 216, 4, 93, 133, 164, 112},
+			},
+			// State node for deployment of contract Test1.
+			{
+				NodeType: 2,
+				Path:     []byte{9},
+				Key:      common.HexToHash("0x9397e33dedda4763aea143fc6151ebcd9a93f62db7a6a556d46c585d82ad2afc"),
+				Value:    []byte{248, 105, 160, 51, 151, 227, 61, 237, 218, 71, 99, 174, 161, 67, 252, 97, 81, 235, 205, 154, 147, 246, 45, 183, 166, 165, 86, 212, 108, 88, 93, 130, 173, 42, 252, 184, 70, 248, 68, 1, 128, 160, 243, 143, 159, 99, 199, 96, 208, 136, 215, 221, 4, 247, 67, 97, 155, 98, 145, 246, 59, 238, 189, 139, 223, 83, 6, 40, 249, 14, 156, 250, 82, 215, 160, 47, 62, 207, 242, 160, 167, 130, 233, 6, 187, 196, 80, 96, 6, 188, 150, 74, 176, 201, 7, 65, 32, 174, 97, 1, 76, 26, 86, 141, 49, 62, 214},
+			},
+		},
+		StorageNodes: [][]snapt.Node{
+			{},
+			{},
+			{
+				// Storage node for contract Test1 state variable initialCount.
+				{
+					NodeType: 2,
+					Path:     []byte{},
+					Key:      common.HexToHash("0xb10e2d527612073b26eecdfd717e6a320cf44b4afac2b0732d9fcbe2b7fa0cf6"),
+					Value:    []byte{227, 161, 32, 177, 14, 45, 82, 118, 18, 7, 59, 38, 238, 205, 253, 113, 126, 106, 50, 12, 244, 75, 74, 250, 194, 176, 115, 45, 159, 203, 226, 183, 250, 12, 246, 1},
+				},
+			},
+		},
+	},
+	// Contract Test2 deployed by main account.
+	{
+		Hash:   common.HexToHash("0xa848b156fe4e61d8dac0a833720794e8c58e93fa6db369af6f0d9a19ada9d723"),
+		Number: big.NewInt(2),
+		StateNodes: []snapt.Node{
+			// Branch root node.
+			{
+				NodeType: 0,
+				Path:     []byte{},
+				Value:    []byte{248, 81, 128, 128, 128, 128, 128, 128, 160, 191, 248, 9, 223, 101, 212, 255, 213, 196, 146, 160, 239, 69, 178, 134, 139, 81, 22, 255, 149, 90, 253, 178, 172, 102, 87, 249, 225, 224, 173, 183, 55, 128, 128, 160, 165, 200, 234, 64, 112, 157, 130, 31, 236, 38, 20, 68, 99, 247, 81, 161, 76, 62, 186, 246, 84, 121, 39, 155, 102, 134, 188, 109, 89, 220, 31, 212, 128, 128, 128, 128, 128, 128, 128},
+			},
+			// State node for sender account.
+			{
+				NodeType: 2,
+				Path:     []byte{6},
+				Key:      common.HexToHash("0x67ab3c0dd775f448af7fb41243415ed6fb975d1530a2d828f69bea7346231ad7"),
+				Value:    []byte{248, 118, 160, 55, 171, 60, 13, 215, 117, 244, 72, 175, 127, 180, 18, 67, 65, 94, 214, 251, 151, 93, 21, 48, 162, 216, 40, 246, 155, 234, 115, 70, 35, 26, 215, 184, 83, 248, 81, 2, 141, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 160, 86, 232, 31, 23, 27, 204, 85, 166, 255, 131, 69, 230, 146, 192, 248, 110, 91, 72, 224, 27, 153, 108, 173, 192, 1, 98, 47, 181, 227, 99, 180, 33, 160, 197, 210, 70, 1, 134, 247, 35, 60, 146, 126, 125, 178, 220, 199, 3, 192, 229, 0, 182, 83, 202, 130, 39, 59, 123, 250, 216, 4, 93, 133, 164, 112},
+			},
+			// State node for deployment of contract Test2.
+			{
+				NodeType: 2,
+				Path:     []byte{10},
+				Key:      common.HexToHash("0xa44b5f4b47ded891709350af6a6e4d56602228a70279bdad4f0f64042445b4b9"),
+				Value:    []byte{248, 105, 160, 52, 75, 95, 75, 71, 222, 216, 145, 112, 147, 80, 175, 106, 110, 77, 86, 96, 34, 40, 167, 2, 121, 189, 173, 79, 15, 100, 4, 36, 69, 180, 185, 184, 70, 248, 68, 1, 128, 160, 130, 30, 37, 86, 162, 144, 200, 100, 5, 248, 22, 10, 45, 102, 32, 66, 164, 49, 186, 69, 107, 157, 178, 101, 199, 155, 184, 55, 192, 75, 229, 240, 160, 86, 36, 245, 233, 5, 167, 42, 118, 181, 35, 178, 216, 149, 56, 146, 147, 19, 8, 140, 137, 234, 0, 160, 27, 220, 33, 204, 6, 152, 239, 177, 52},
+			},
+		},
+		StorageNodes: [][]snapt.Node{
+			{},
+			{},
+			{
+				// Storage node for contract Test2 state variable test.
+				{
+					NodeType: 2,
+					Path:     []byte{},
+					Key:      common.HexToHash("0x290decd9548b62a8d60345a988386fc84ba6bc95484008f6362f93160ef3e563"),
+					Value:    []byte{227, 161, 32, 41, 13, 236, 217, 84, 139, 98, 168, 214, 3, 69, 169, 136, 56, 111, 200, 75, 166, 188, 149, 72, 64, 8, 246, 54, 47, 147, 22, 14, 243, 229, 99, 1},
+				},
+			},
+		},
+	},
+	// Increment contract Test1 state variable count using main account.
+	{
+		Hash:   common.HexToHash("0x9fc4aaaab26f0b43ac609c99ae50925e5dc9a25f103c0511fcff38c6b3158302"),
+		Number: big.NewInt(3),
+		StateNodes: []snapt.Node{
+			// Branch root node.
+			{
+				NodeType: 0,
+				Path:     []byte{},
+				Value:    []byte{248, 113, 128, 128, 128, 128, 128, 128, 160, 70, 53, 190, 199, 124, 254, 86, 213, 42, 126, 117, 155, 2, 223, 56, 167, 130, 118, 10, 150, 65, 46, 207, 169, 167, 250, 209, 64, 37, 205, 153, 51, 128, 128, 160, 165, 200, 234, 64, 112, 157, 130, 31, 236, 38, 20, 68, 99, 247, 81, 161, 76, 62, 186, 246, 84, 121, 39, 155, 102, 134, 188, 109, 89, 220, 31, 212, 128, 128, 160, 214, 109, 199, 206, 145, 11, 213, 44, 206, 214, 36, 181, 134, 92, 243, 178, 58, 88, 158, 42, 31, 125, 71, 148, 188, 122, 252, 100, 250, 182, 85, 159, 128, 128, 128, 128},
+			},
+			// State node for sender account.
+			{
+				NodeType: 2,
+				Path:     []byte{6},
+				Key:      common.HexToHash("0x67ab3c0dd775f448af7fb41243415ed6fb975d1530a2d828f69bea7346231ad7"),
+				Value:    []byte{248, 118, 160, 55, 171, 60, 13, 215, 117, 244, 72, 175, 127, 180, 18, 67, 65, 94, 214, 251, 151, 93, 21, 48, 162, 216, 40, 246, 155, 234, 115, 70, 35, 26, 215, 184, 83, 248, 81, 3, 141, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 160, 86, 232, 31, 23, 27, 204, 85, 166, 255, 131, 69, 230, 146, 192, 248, 110, 91, 72, 224, 27, 153, 108, 173, 192, 1, 98, 47, 181, 227, 99, 180, 33, 160, 197, 210, 70, 1, 134, 247, 35, 60, 146, 126, 125, 178, 220, 199, 3, 192, 229, 0, 182, 83, 202, 130, 39, 59, 123, 250, 216, 4, 93, 133, 164, 112},
+			},
+			// State node for contract Test1 transaction.
+			{
+				NodeType: 2,
+				Path:     []byte{9},
+				Key:      common.HexToHash("0x9397e33dedda4763aea143fc6151ebcd9a93f62db7a6a556d46c585d82ad2afc"),
+				Value:    []byte{248, 105, 160, 51, 151, 227, 61, 237, 218, 71, 99, 174, 161, 67, 252, 97, 81, 235, 205, 154, 147, 246, 45, 183, 166, 165, 86, 212, 108, 88, 93, 130, 173, 42, 252, 184, 70, 248, 68, 1, 128, 160, 167, 171, 204, 110, 30, 52, 74, 189, 215, 97, 245, 227, 176, 141, 250, 205, 8, 182, 138, 101, 51, 150, 155, 174, 234, 246, 30, 128, 253, 230, 36, 228, 160, 47, 62, 207, 242, 160, 167, 130, 233, 6, 187, 196, 80, 96, 6, 188, 150, 74, 176, 201, 7, 65, 32, 174, 97, 1, 76, 26, 86, 141, 49, 62, 214},
+			},
+		},
+		StorageNodes: [][]snapt.Node{
+			{},
+			{},
+			{
+				// Branch root node.
+				{
+					NodeType: 0,
+					Path:     []byte{},
+					Value:    []byte{248, 81, 128, 128, 160, 79, 197, 241, 58, 178, 249, 186, 12, 45, 168, 139, 1, 81, 171, 14, 124, 244, 216, 93, 8, 204, 164, 92, 205, 146, 60, 106, 183, 99, 35, 235, 40, 128, 128, 128, 128, 128, 128, 128, 128, 160, 244, 152, 74, 17, 246, 26, 41, 33, 69, 97, 65, 223, 136, 222, 110, 26, 113, 13, 40, 104, 27, 145, 175, 121, 76, 90, 114, 30, 71, 131, 156, 215, 128, 128, 128, 128, 128},
+				},
+				// Storage node for contract Test1 state variable count.
+				{
+					NodeType: 2,
+					Path:     []byte{2},
+					Key:      common.HexToHash("0x290decd9548b62a8d60345a988386fc84ba6bc95484008f6362f93160ef3e563"),
+					Value:    []byte{226, 160, 57, 13, 236, 217, 84, 139, 98, 168, 214, 3, 69, 169, 136, 56, 111, 200, 75, 166, 188, 149, 72, 64, 8, 246, 54, 47, 147, 22, 14, 243, 229, 99, 1},
+				},
+				// Storage node for contract Test1 state variable initialCount.
+				{
+					NodeType: 2,
+					Path:     []byte{11},
+					Key:      common.HexToHash("0xb10e2d527612073b26eecdfd717e6a320cf44b4afac2b0732d9fcbe2b7fa0cf6"),
+					Value:    []byte{226, 160, 49, 14, 45, 82, 118, 18, 7, 59, 38, 238, 205, 253, 113, 126, 106, 50, 12, 244, 75, 74, 250, 194, 176, 115, 45, 159, 203, 226, 183, 250, 12, 246, 1},
+				},
+			},
+		},
+	},
+}
+
+// Expected state nodes at snapshot height.
+var ExpectedStateNodes = []snapt.Node{
+	{
+		NodeType: 0,
+		Path:     []byte{},
+		Value:    []byte{248, 113, 128, 128, 128, 128, 128, 128, 160, 70, 53, 190, 199, 124, 254, 86, 213, 42, 126, 117, 155, 2, 223, 56, 167, 130, 118, 10, 150, 65, 46, 207, 169, 167, 250, 209, 64, 37, 205, 153, 51, 128, 128, 160, 165, 200, 234, 64, 112, 157, 130, 31, 236, 38, 20, 68, 99, 247, 81, 161, 76, 62, 186, 246, 84, 121, 39, 155, 102, 134, 188, 109, 89, 220, 31, 212, 128, 128, 160, 214, 109, 199, 206, 145, 11, 213, 44, 206, 214, 36, 181, 134, 92, 243, 178, 58, 88, 158, 42, 31, 125, 71, 148, 188, 122, 252, 100, 250, 182, 85, 159, 128, 128, 128, 128},
+	},
+	{
+		NodeType: 2,
+		Path:     []byte{6},
+		Key:      common.HexToHash("0x67ab3c0dd775f448af7fb41243415ed6fb975d1530a2d828f69bea7346231ad7"),
+		Value:    []byte{248, 118, 160, 55, 171, 60, 13, 215, 117, 244, 72, 175, 127, 180, 18, 67, 65, 94, 214, 251, 151, 93, 21, 48, 162, 216, 40, 246, 155, 234, 115, 70, 35, 26, 215, 184, 83, 248, 81, 3, 141, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 160, 86, 232, 31, 23, 27, 204, 85, 166, 255, 131, 69, 230, 146, 192, 248, 110, 91, 72, 224, 27, 153, 108, 173, 192, 1, 98, 47, 181, 227, 99, 180, 33, 160, 197, 210, 70, 1, 134, 247, 35, 60, 146, 126, 125, 178, 220, 199, 3, 192, 229, 0, 182, 83, 202, 130, 39, 59, 123, 250, 216, 4, 93, 133, 164, 112},
+	},
+	{
+		NodeType: 2,
+		Path:     []byte{9},
+		Key:      common.HexToHash("0x9397e33dedda4763aea143fc6151ebcd9a93f62db7a6a556d46c585d82ad2afc"),
+		Value:    []byte{248, 105, 160, 51, 151, 227, 61, 237, 218, 71, 99, 174, 161, 67, 252, 97, 81, 235, 205, 154, 147, 246, 45, 183, 166, 165, 86, 212, 108, 88, 93, 130, 173, 42, 252, 184, 70, 248, 68, 1, 128, 160, 167, 171, 204, 110, 30, 52, 74, 189, 215, 97, 245, 227, 176, 141, 250, 205, 8, 182, 138, 101, 51, 150, 155, 174, 234, 246, 30, 128, 253, 230, 36, 228, 160, 47, 62, 207, 242, 160, 167, 130, 233, 6, 187, 196, 80, 96, 6, 188, 150, 74, 176, 201, 7, 65, 32, 174, 97, 1, 76, 26, 86, 141, 49, 62, 214},
+	},
+	{
+		NodeType: 2,
+		Path:     []byte{10},
+		Key:      common.HexToHash("0xa44b5f4b47ded891709350af6a6e4d56602228a70279bdad4f0f64042445b4b9"),
+		Value:    []byte{248, 105, 160, 52, 75, 95, 75, 71, 222, 216, 145, 112, 147, 80, 175, 106, 110, 77, 86, 96, 34, 40, 167, 2, 121, 189, 173, 79, 15, 100, 4, 36, 69, 180, 185, 184, 70, 248, 68, 1, 128, 160, 130, 30, 37, 86, 162, 144, 200, 100, 5, 248, 22, 10, 45, 102, 32, 66, 164, 49, 186, 69, 107, 157, 178, 101, 199, 155, 184, 55, 192, 75, 229, 240, 160, 86, 36, 245, 233, 5, 167, 42, 118, 181, 35, 178, 216, 149, 56, 146, 147, 19, 8, 140, 137, 234, 0, 160, 27, 220, 33, 204, 6, 152, 239, 177, 52},
+	},
+}
+
+type StorageNodeWithState struct {
+	snapt.Node
+	StatePath []byte
+}
+
+// Expected storage nodes at snapshot height.
+var ExpectedStorageNodes = []StorageNodeWithState{
+	{
+		Node: snapt.Node{
+			NodeType: 0,
+			Path:     []byte{},
+			Value:    []byte{248, 81, 128, 128, 160, 79, 197, 241, 58, 178, 249, 186, 12, 45, 168, 139, 1, 81, 171, 14, 124, 244, 216, 93, 8, 204, 164, 92, 205, 146, 60, 106, 183, 99, 35, 235, 40, 128, 128, 128, 128, 128, 128, 128, 128, 160, 244, 152, 74, 17, 246, 26, 41, 33, 69, 97, 65, 223, 136, 222, 110, 26, 113, 13, 40, 104, 27, 145, 175, 121, 76, 90, 114, 30, 71, 131, 156, 215, 128, 128, 128, 128, 128},
+		},
+		StatePath: []byte{9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{2},
+			Key:      common.HexToHash("0x290decd9548b62a8d60345a988386fc84ba6bc95484008f6362f93160ef3e563"),
+			Value:    []byte{226, 160, 57, 13, 236, 217, 84, 139, 98, 168, 214, 3, 69, 169, 136, 56, 111, 200, 75, 166, 188, 149, 72, 64, 8, 246, 54, 47, 147, 22, 14, 243, 229, 99, 1},
+		},
+		StatePath: []byte{9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{11},
+			Key:      common.HexToHash("0xb10e2d527612073b26eecdfd717e6a320cf44b4afac2b0732d9fcbe2b7fa0cf6"),
+			Value:    []byte{226, 160, 49, 14, 45, 82, 118, 18, 7, 59, 38, 238, 205, 253, 113, 126, 106, 50, 12, 244, 75, 74, 250, 194, 176, 115, 45, 159, 203, 226, 183, 250, 12, 246, 1},
+		},
+		StatePath: []byte{9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{},
+			Key:      common.HexToHash("0x290decd9548b62a8d60345a988386fc84ba6bc95484008f6362f93160ef3e563"),
+			Value:    []byte{227, 161, 32, 41, 13, 236, 217, 84, 139, 98, 168, 214, 3, 69, 169, 136, 56, 111, 200, 75, 166, 188, 149, 72, 64, 8, 246, 54, 47, 147, 22, 14, 243, 229, 99, 1},
+		},
+		StatePath: []byte{10},
+	},
+}
+
+// Block header at snapshot height.
+// Required in database when executing inPlaceStateSnapshot.
+var Block4_Header = types.Header{
+	ParentHash:  common.HexToHash("0x9fc4aaaab26f0b43ac609c99ae50925e5dc9a25f103c0511fcff38c6b3158302"),
+	UncleHash:   common.HexToHash("0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347"),
+	Coinbase:    common.HexToAddress("0x0000000000000000000000000000000000000000"),
+	Root:        common.HexToHash("0x53580584816f617295ea26c0e17641e0120cab2f0a8ffb53a866fd53aa8e8c2d"),
+	TxHash:      common.HexToHash("0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"),
+	ReceiptHash: common.HexToHash("0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"),
+	Bloom:       types.Bloom{},
+	Difficulty:  big.NewInt(+2),
+	Number:      big.NewInt(4),
+	GasLimit:    4704588,
+	GasUsed:     0,
+	Time:        1492010458,
+	Extra:       []byte{215, 131, 1, 6, 0, 132, 103, 101, 116, 104, 135, 103, 111, 49, 46, 55, 46, 51, 133, 108, 105, 110, 117, 120, 0, 0, 0, 0, 0, 0, 0, 0, 159, 30, 250, 30, 250, 114, 175, 19, 140, 145, 89, 102, 198, 57, 84, 74, 2, 85, 230, 40, 142, 24, 140, 34, 206, 145, 104, 193, 13, 190, 70, 218, 61, 136, 180, 170, 6, 89, 48, 17, 159, 184, 134, 33, 11, 240, 26, 8, 79, 222, 93, 59, 196, 141, 138, 163, 139, 202, 146, 228, 252, 197, 33, 81, 0},
+	MixDigest:   common.Hash{},
+	Nonce:       types.BlockNonce{},
+	BaseFee:     nil,
+}
+
+/*
+pragma solidity ^0.8.0;
+
+contract Test1 {
+    uint256 private count;
+    uint256 private initialCount;
+
+    event Increment(uint256 count);
+
+    constructor() {
+      initialCount = 1;
+    }
+
+    function incrementCount() public returns (uint256) {
+      count = count + 1;
+      emit Increment(count);
+
+      return count;
+    }
+}
+*/
+
+/*
+pragma solidity ^0.8.0;
+
+contract Test2 {
+    uint256 private test;
+
+    constructor() {
+    	test = 1;
+    }
+}
+*/
@@ -1,20 +1,117 @@
 module github.com/vulcanize/ipld-eth-state-snapshot

-go 1.15
+go 1.18

 require (
-	github.com/btcsuite/btcd/btcec/v2 v2.2.0 // indirect
-	github.com/ethereum/go-ethereum v1.10.17
+	github.com/ethereum/go-ethereum v1.10.19
 	github.com/golang/mock v1.6.0
 	github.com/ipfs/go-cid v0.1.0
 	github.com/ipfs/go-ipfs-blockstore v1.1.2
 	github.com/ipfs/go-ipfs-ds-help v1.1.0
-	github.com/jackc/pgx/v4 v4.15.0
+	github.com/jmoiron/sqlx v1.2.0
 	github.com/multiformats/go-multihash v0.1.0
+	github.com/prometheus/client_golang v1.3.0
 	github.com/sirupsen/logrus v1.6.0
 	github.com/spf13/cobra v1.0.0
 	github.com/spf13/viper v1.7.0
-	github.com/vulcanize/go-eth-state-node-iterator v1.0.1
+	github.com/vulcanize/go-eth-state-node-iterator v1.1.1
 )

-replace github.com/ethereum/go-ethereum v1.10.17 => github.com/vulcanize/go-ethereum v1.10.17-statediff-4.0.0-alpha
+require (
+	github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6 // indirect
+	github.com/VictoriaMetrics/fastcache v1.6.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/btcsuite/btcd/btcec/v2 v2.2.0 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
+	github.com/fsnotify/fsnotify v1.5.1 // indirect
+	github.com/georgysavva/scany v0.2.9 // indirect
+	github.com/go-kit/kit v0.10.0 // indirect
+	github.com/go-ole/go-ole v1.2.1 // indirect
+	github.com/go-stack/stack v1.8.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/go-cmp v0.5.6 // indirect
+	github.com/google/uuid v1.3.0 // indirect
+	github.com/gopherjs/gopherjs v0.0.0-20190430165422-3e4dfb77656c // indirect
+	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect
+	github.com/hashicorp/hcl v1.0.0 // indirect
+	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
+	github.com/inconshreveable/mousetrap v1.0.0 // indirect
+	github.com/ipfs/bbloom v0.0.4 // indirect
+	github.com/ipfs/go-block-format v0.0.3 // indirect
+	github.com/ipfs/go-datastore v0.5.1 // indirect
+	github.com/ipfs/go-ipfs-util v0.0.2 // indirect
+	github.com/ipfs/go-ipld-format v0.2.0 // indirect
+	github.com/ipfs/go-log v1.0.5 // indirect
+	github.com/ipfs/go-log/v2 v2.4.0 // indirect
+	github.com/ipfs/go-metrics-interface v0.0.1 // indirect
+	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
+	github.com/jackc/pgconn v1.11.0 // indirect
+	github.com/jackc/pgio v1.0.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgproto3/v2 v2.2.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
+	github.com/jackc/pgtype v1.10.0 // indirect
+	github.com/jackc/pgx/v4 v4.15.0 // indirect
+	github.com/jackc/puddle v1.2.1 // indirect
+	github.com/jbenet/goprocess v0.1.4 // indirect
+	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
+	github.com/konsorten/go-windows-terminal-sequences v1.0.3 // indirect
+	github.com/kr/pretty v0.3.0 // indirect
+	github.com/lib/pq v1.10.2 // indirect
+	github.com/magiconair/properties v1.8.1 // indirect
+	github.com/mattn/go-isatty v0.0.14 // indirect
+	github.com/mattn/go-runewidth v0.0.9 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1 // indirect
+	github.com/minio/sha256-simd v1.0.0 // indirect
+	github.com/mitchellh/mapstructure v1.4.1 // indirect
+	github.com/mr-tron/base58 v1.2.0 // indirect
+	github.com/multiformats/go-base32 v0.0.4 // indirect
+	github.com/multiformats/go-base36 v0.1.0 // indirect
+	github.com/multiformats/go-multibase v0.0.3 // indirect
+	github.com/multiformats/go-varint v0.0.6 // indirect
+	github.com/olekukonko/tablewriter v0.0.5 // indirect
+	github.com/onsi/ginkgo v1.16.5 // indirect
+	github.com/onsi/gomega v1.13.0 // indirect
+	github.com/opentracing/opentracing-go v1.2.0 // indirect
+	github.com/pelletier/go-toml v1.2.0 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.1.0 // indirect
+	github.com/prometheus/common v0.7.0 // indirect
+	github.com/prometheus/procfs v0.0.8 // indirect
+	github.com/prometheus/tsdb v0.7.1 // indirect
+	github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible // indirect
+	github.com/smartystreets/assertions v1.0.0 // indirect
+	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	github.com/spf13/afero v1.1.2 // indirect
+	github.com/spf13/cast v1.3.0 // indirect
+	github.com/spf13/jwalterweatherman v1.0.0 // indirect
+	github.com/spf13/pflag v1.0.3 // indirect
+	github.com/stretchr/objx v0.2.0 // indirect
+	github.com/stretchr/testify v1.7.0 // indirect
+	github.com/subosito/gotenv v1.2.0 // indirect
+	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 // indirect
+	github.com/tklauser/go-sysconf v0.3.5 // indirect
+	github.com/tklauser/numcpus v0.2.2 // indirect
+	go.uber.org/atomic v1.9.0 // indirect
+	go.uber.org/goleak v1.1.11 // indirect
+	go.uber.org/multierr v1.7.0 // indirect
+	go.uber.org/zap v1.19.1 // indirect
+	golang.org/x/crypto v0.0.0-20211209193657-4570a0811e8b // indirect
+	golang.org/x/net v0.0.0-20211209124913-491a49abca63 // indirect
+	golang.org/x/sys v0.0.0-20211209171907-798191bca915 // indirect
+	golang.org/x/text v0.3.7 // indirect
+	google.golang.org/appengine v1.6.6 // indirect
+	google.golang.org/protobuf v1.27.1 // indirect
+	gopkg.in/ini.v1 v1.51.0 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	lukechampine.com/blake3 v1.1.7 // indirect
+)
+
+replace github.com/ethereum/go-ethereum v1.10.19 => github.com/vulcanize/go-ethereum v1.10.19-statediff-4.1.0-alpha
@@ -0,0 +1,157 @@
+// VulcanizeDB
+// Copyright © 2022 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 (
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
+)
+
+// DBStatsGetter is an interface that gets sql.DBStats.
+type DBStatsGetter interface {
+	Stats() sql.Stats
+}
+
+// DBStatsCollector implements the prometheus.Collector interface.
+type DBStatsCollector struct {
+	sg DBStatsGetter
+
+	// descriptions of exported metrics
+	maxOpenDesc           *prometheus.Desc
+	openDesc              *prometheus.Desc
+	inUseDesc             *prometheus.Desc
+	idleDesc              *prometheus.Desc
+	waitedForDesc         *prometheus.Desc
+	blockedSecondsDesc    *prometheus.Desc
+	closedMaxIdleDesc     *prometheus.Desc
+	closedMaxLifetimeDesc *prometheus.Desc
+}
+
+// NewDBStatsCollector creates a new DBStatsCollector.
+func NewDBStatsCollector(dbName string, sg DBStatsGetter) *DBStatsCollector {
+	labels := prometheus.Labels{"db_name": dbName}
+	return &DBStatsCollector{
+		sg: sg,
+		maxOpenDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "max_open"),
+			"Maximum number of open connections to the database.",
+			nil,
+			labels,
+		),
+		openDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "open"),
+			"The number of established connections both in use and idle.",
+			nil,
+			labels,
+		),
+		inUseDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "in_use"),
+			"The number of connections currently in use.",
+			nil,
+			labels,
+		),
+		idleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "idle"),
+			"The number of idle connections.",
+			nil,
+			labels,
+		),
+		waitedForDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "waited_for"),
+			"The total number of connections waited for.",
+			nil,
+			labels,
+		),
+		blockedSecondsDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "blocked_seconds"),
+			"The total time blocked waiting for a new connection.",
+			nil,
+			labels,
+		),
+		closedMaxIdleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "closed_max_idle"),
+			"The total number of connections closed due to SetMaxIdleConns.",
+			nil,
+			labels,
+		),
+		closedMaxLifetimeDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "closed_max_lifetime"),
+			"The total number of connections closed due to SetConnMaxLifetime.",
+			nil,
+			labels,
+		),
+	}
+}
+
+// Describe implements the prometheus.Collector interface.
+func (c DBStatsCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.maxOpenDesc
+	ch <- c.openDesc
+	ch <- c.inUseDesc
+	ch <- c.idleDesc
+	ch <- c.waitedForDesc
+	ch <- c.blockedSecondsDesc
+	ch <- c.closedMaxIdleDesc
+	ch <- c.closedMaxLifetimeDesc
+}
+
+// Collect implements the prometheus.Collector interface.
+func (c DBStatsCollector) Collect(ch chan<- prometheus.Metric) {
+	stats := c.sg.Stats()
+
+	ch <- prometheus.MustNewConstMetric(
+		c.maxOpenDesc,
+		prometheus.GaugeValue,
+		float64(stats.MaxOpen()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.openDesc,
+		prometheus.GaugeValue,
+		float64(stats.Open()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.inUseDesc,
+		prometheus.GaugeValue,
+		float64(stats.InUse()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.idleDesc,
+		prometheus.GaugeValue,
+		float64(stats.Idle()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.waitedForDesc,
+		prometheus.CounterValue,
+		float64(stats.WaitCount()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.blockedSecondsDesc,
+		prometheus.CounterValue,
+		stats.WaitDuration().Seconds(),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.closedMaxIdleDesc,
+		prometheus.CounterValue,
+		float64(stats.MaxIdleClosed()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.closedMaxLifetimeDesc,
+		prometheus.CounterValue,
+		float64(stats.MaxLifetimeClosed()),
+	)
+}
@@ -0,0 +1,90 @@
+// VulcanizeDB
+// Copyright © 2022 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 (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const (
+	namespace = "ipld_eth_state_snapshot"
+
+	connSubsystem  = "connections"
+	statsSubsystem = "stats"
+)
+
+var (
+	metrics bool
+
+	stateNodeCount   prometheus.Counter
+	storageNodeCount prometheus.Counter
+	codeNodeCount    prometheus.Counter
+)
+
+func Init() {
+	metrics = true
+
+	stateNodeCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		Name:      "state_node_count",
+		Help:      "Number of state nodes processed",
+	})
+
+	storageNodeCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		Name:      "storage_node_count",
+		Help:      "Number of storage nodes processed",
+	})
+
+	codeNodeCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		Name:      "code_node_count",
+		Help:      "Number of code nodes processed",
+	})
+}
+
+// RegisterDBCollector create metric collector for given connection
+func RegisterDBCollector(name string, db DBStatsGetter) {
+	if metrics {
+		prometheus.Register(NewDBStatsCollector(name, db))
+	}
+}
+
+// IncStateNodeCount increments the number of state nodes processed
+func IncStateNodeCount() {
+	if metrics {
+		stateNodeCount.Inc()
+	}
+}
+
+// IncStorageNodeCount increments the number of storage nodes processed
+func IncStorageNodeCount() {
+	if metrics {
+		storageNodeCount.Inc()
+	}
+}
+
+// IncCodeNodeCount increments the number of code nodes processed
+func IncCodeNodeCount() {
+	if metrics {
+		codeNodeCount.Inc()
+	}
+}
@@ -0,0 +1,47 @@
+// VulcanizeDB
+// Copyright © 2022 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 (
+	"errors"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/sirupsen/logrus"
+)
+
+var errPromHTTP = errors.New("can't start http server for prometheus")
+
+// Serve start listening http
+func Serve(addr string) *http.Server {
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+	srv := http.Server{
+		Addr:    addr,
+		Handler: mux,
+	}
+	go func() {
+		if err := srv.ListenAndServe(); err != nil {
+			logrus.
+				WithError(err).
+				WithField("module", "prom").
+				WithField("addr", addr).
+				Fatal(errPromHTTP)
+		}
+	}()
+	return &srv
+}
@@ -26,15 +26,6 @@ import (
 	"github.com/spf13/viper"
 )

-const (
-	ANCIENT_DB_PATH   = "ANCIENT_DB_PATH"
-	ETH_CLIENT_NAME   = "ETH_CLIENT_NAME"
-	ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
-	ETH_NETWORK_ID    = "ETH_NETWORK_ID"
-	ETH_NODE_ID       = "ETH_NODE_ID"
-	LVL_DB_PATH       = "LVL_DB_PATH"
-)
-
 // SnapshotMode specifies the snapshot data output method
 type SnapshotMode string

@@ -78,26 +69,38 @@ func NewConfig(mode SnapshotMode) (*Config, error) {
 	return ret, ret.Init(mode)
 }

+func NewInPlaceSnapshotConfig() *Config {
+	ret := &Config{
+		&EthConfig{},
+		&DBConfig{},
+		&FileConfig{},
+	}
+	ret.DB.Init()
+
+	return ret
+}
+
 // Init Initialises config
 func (c *Config) Init(mode SnapshotMode) error {
-	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(ETH_NODE_ID_TOML, ETH_NODE_ID)
+	viper.BindEnv(ETH_CLIENT_NAME_TOML, ETH_CLIENT_NAME)
+	viper.BindEnv(ETH_GENESIS_BLOCK_TOML, ETH_GENESIS_BLOCK)
+	viper.BindEnv(ETH_NETWORK_ID_TOML, ETH_NETWORK_ID)
+	viper.BindEnv(ETH_CHAIN_ID_TOML, ETH_CHAIN_ID)

 	c.Eth.NodeInfo = ethNode.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:           viper.GetString(ETH_NODE_ID_TOML),
+		ClientName:   viper.GetString(ETH_CLIENT_NAME_TOML),
+		GenesisBlock: viper.GetString(ETH_GENESIS_BLOCK_TOML),
+		NetworkID:    viper.GetString(ETH_NETWORK_ID_TOML),
+		ChainID:      viper.GetUint64(ETH_CHAIN_ID_TOML),
 	}

-	viper.BindEnv("leveldb.ancient", ANCIENT_DB_PATH)
-	viper.BindEnv("leveldb.path", LVL_DB_PATH)
+	viper.BindEnv(ANCIENT_DB_PATH_TOML, ANCIENT_DB_PATH)
+	viper.BindEnv(LVL_DB_PATH_TOML, LVL_DB_PATH)

-	c.Eth.AncientDBPath = viper.GetString("leveldb.ancient")
-	c.Eth.LevelDBPath = viper.GetString("leveldb.path")
+	c.Eth.AncientDBPath = viper.GetString(ANCIENT_DB_PATH_TOML)
+	c.Eth.LevelDBPath = viper.GetString(LVL_DB_PATH_TOML)

 	switch mode {
 	case FileSnapshot:
@@ -111,34 +114,34 @@ func (c *Config) Init(mode SnapshotMode) error {
 }

 func (c *DBConfig) Init() {
-	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.maxLifetime", "DATABASE_MAX_CONN_LIFETIME")
+	viper.BindEnv(DATABASE_NAME_TOML, DATABASE_NAME)
+	viper.BindEnv(DATABASE_HOSTNAME_TOML, DATABASE_HOSTNAME)
+	viper.BindEnv(DATABASE_PORT_TOML, DATABASE_PORT)
+	viper.BindEnv(DATABASE_USER_TOML, DATABASE_USER)
+	viper.BindEnv(DATABASE_PASSWORD_TOML, DATABASE_PASSWORD)
+	viper.BindEnv(DATABASE_MAX_IDLE_CONNECTIONS_TOML, DATABASE_MAX_IDLE_CONNECTIONS)
+	viper.BindEnv(DATABASE_MAX_OPEN_CONNECTIONS_TOML, DATABASE_MAX_OPEN_CONNECTIONS)
+	viper.BindEnv(DATABASE_MAX_CONN_LIFETIME_TOML, DATABASE_MAX_CONN_LIFETIME)

 	dbParams := postgres.Config{}
 	// DB params
-	dbParams.DatabaseName = viper.GetString("database.name")
-	dbParams.Hostname = viper.GetString("database.hostname")
-	dbParams.Port = viper.GetInt("database.port")
-	dbParams.Username = viper.GetString("database.user")
-	dbParams.Password = viper.GetString("database.password")
+	dbParams.DatabaseName = viper.GetString(DATABASE_NAME_TOML)
+	dbParams.Hostname = viper.GetString(DATABASE_HOSTNAME_TOML)
+	dbParams.Port = viper.GetInt(DATABASE_PORT_TOML)
+	dbParams.Username = viper.GetString(DATABASE_USER_TOML)
+	dbParams.Password = viper.GetString(DATABASE_PASSWORD_TOML)
 	// Connection config
-	dbParams.MaxIdle = viper.GetInt("database.maxIdle")
-	dbParams.MaxConns = viper.GetInt("database.maxOpen")
-	dbParams.MaxConnLifetime = time.Duration(viper.GetInt("database.maxLifetime")) * time.Second
+	dbParams.MaxIdle = viper.GetInt(DATABASE_MAX_IDLE_CONNECTIONS_TOML)
+	dbParams.MaxConns = viper.GetInt(DATABASE_MAX_OPEN_CONNECTIONS_TOML)
+	dbParams.MaxConnLifetime = time.Duration(viper.GetInt(DATABASE_MAX_CONN_LIFETIME_TOML)) * time.Second

 	c.ConnConfig = dbParams
 	c.URI = dbParams.DbConnectionString()
 }

 func (c *FileConfig) Init() error {
-	viper.BindEnv("file.outputDir", "FILE_OUTPUT_DIR")
-	c.OutputDir = viper.GetString("file.outputDir")
+	viper.BindEnv(FILE_OUTPUT_DIR_TOML, FILE_OUTPUT_DIR)
+	c.OutputDir = viper.GetString(FILE_OUTPUT_DIR_TOML)
 	if c.OutputDir == "" {
 		logrus.Infof("no output directory set, using default: %s", defaultOutputDir)
 		c.OutputDir = defaultOutputDir
@@ -0,0 +1,133 @@
+// Copyright © 2020 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 snapshot
+
+// ENV variables
+const (
+	SNAPSHOT_BLOCK_HEIGHT  = "SNAPSHOT_BLOCK_HEIGHT"
+	SNAPSHOT_WORKERS       = "SNAPSHOT_WORKERS"
+	SNAPSHOT_RECOVERY_FILE = "SNAPSHOT_RECOVERY_FILE"
+	SNAPSHOT_MODE          = "SNAPSHOT_MODE"
+	SNAPSHOT_START_HEIGHT  = "SNAPSHOT_START_HEIGHT"
+	SNAPSHOT_END_HEIGHT    = "SNAPSHOT_END_HEIGHT"
+
+	LOGRUS_LEVEL = "LOGRUS_LEVEL"
+	LOGRUS_FILE  = "LOGRUS_FILE"
+
+	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"
+
+	FILE_OUTPUT_DIR = "FILE_OUTPUT_DIR"
+
+	ANCIENT_DB_PATH = "ANCIENT_DB_PATH"
+	LVL_DB_PATH     = "LVL_DB_PATH"
+
+	ETH_CLIENT_NAME   = "ETH_CLIENT_NAME"
+	ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
+	ETH_NETWORK_ID    = "ETH_NETWORK_ID"
+	ETH_NODE_ID       = "ETH_NODE_ID"
+	ETH_CHAIN_ID      = "ETH_CHAIN_ID"
+
+	DATABASE_NAME                 = "DATABASE_NAME"
+	DATABASE_HOSTNAME             = "DATABASE_HOSTNAME"
+	DATABASE_PORT                 = "DATABASE_PORT"
+	DATABASE_USER                 = "DATABASE_USER"
+	DATABASE_PASSWORD             = "DATABASE_PASSWORD"
+	DATABASE_MAX_IDLE_CONNECTIONS = "DATABASE_MAX_IDLE_CONNECTIONS"
+	DATABASE_MAX_OPEN_CONNECTIONS = "DATABASE_MAX_OPEN_CONNECTIONS"
+	DATABASE_MAX_CONN_LIFETIME    = "DATABASE_MAX_CONN_LIFETIME"
+)
+
+// TOML bindings
+const (
+	SNAPSHOT_BLOCK_HEIGHT_TOML  = "snapshot.blockHeight"
+	SNAPSHOT_WORKERS_TOML       = "snapshot.workers"
+	SNAPSHOT_RECOVERY_FILE_TOML = "snapshot.recoveryFile"
+	SNAPSHOT_MODE_TOML          = "snapshot.mode"
+	SNAPSHOT_START_HEIGHT_TOML  = "snapshot.startHeight"
+	SNAPSHOT_END_HEIGHT_TOML    = "snapshot.endHeight"
+
+	LOGRUS_LEVEL_TOML = "log.level"
+	LOGRUS_FILE_TOML  = "log.file"
+
+	PROM_METRICS_TOML   = "prom.metrics"
+	PROM_HTTP_TOML      = "prom.http"
+	PROM_HTTP_ADDR_TOML = "prom.httpAddr"
+	PROM_HTTP_PORT_TOML = "prom.httpPort"
+	PROM_DB_STATS_TOML  = "prom.dbStats"
+
+	FILE_OUTPUT_DIR_TOML = "file.outputDir"
+
+	ANCIENT_DB_PATH_TOML = "leveldb.ancient"
+	LVL_DB_PATH_TOML     = "leveldb.path"
+
+	ETH_CLIENT_NAME_TOML   = "ethereum.clientName"
+	ETH_GENESIS_BLOCK_TOML = "ethereum.genesisBlock"
+	ETH_NETWORK_ID_TOML    = "ethereum.networkID"
+	ETH_NODE_ID_TOML       = "ethereum.nodeID"
+	ETH_CHAIN_ID_TOML      = "ethereum.chainID"
+
+	DATABASE_NAME_TOML                 = "database.name"
+	DATABASE_HOSTNAME_TOML             = "database.hostname"
+	DATABASE_PORT_TOML                 = "database.port"
+	DATABASE_USER_TOML                 = "database.user"
+	DATABASE_PASSWORD_TOML             = "database.password"
+	DATABASE_MAX_IDLE_CONNECTIONS_TOML = "database.maxIdle"
+	DATABASE_MAX_OPEN_CONNECTIONS_TOML = "database.maxOpen"
+	DATABASE_MAX_CONN_LIFETIME_TOML    = "database.maxLifetime"
+)
+
+// CLI flags
+const (
+	SNAPSHOT_BLOCK_HEIGHT_CLI  = "block-height"
+	SNAPSHOT_WORKERS_CLI       = "workers"
+	SNAPSHOT_RECOVERY_FILE_CLI = "recovery-file"
+	SNAPSHOT_MODE_CLI          = "snapshot-mode"
+	SNAPSHOT_START_HEIGHT_CLI  = "start-height"
+	SNAPSHOT_END_HEIGHT_CLI    = "end-height"
+
+	LOGRUS_LEVEL_CLI = "log-level"
+	LOGRUS_FILE_CLI  = "log-file"
+
+	PROM_METRICS_CLI   = "prom-metrics"
+	PROM_HTTP_CLI      = "prom-http"
+	PROM_HTTP_ADDR_CLI = "prom-httpAddr"
+	PROM_HTTP_PORT_CLI = "prom-httpPort"
+	PROM_DB_STATS_CLI  = "prom-dbStats"
+
+	FILE_OUTPUT_DIR_CLI = "output-dir"
+
+	ANCIENT_DB_PATH_CLI = "ancient-path"
+	LVL_DB_PATH_CLI     = "leveldb-path"
+
+	ETH_CLIENT_NAME_CLI   = "ethereum-client-name"
+	ETH_GENESIS_BLOCK_CLI = "ethereum-genesis-block"
+	ETH_NETWORK_ID_CLI    = "ethereum-network-id"
+	ETH_NODE_ID_CLI       = "ethereum-node-id"
+	ETH_CHAIN_ID_CLI      = "ethereum-chain-id"
+
+	DATABASE_NAME_CLI                 = "database-name"
+	DATABASE_HOSTNAME_CLI             = "database-hostname"
+	DATABASE_PORT_CLI                 = "database-port"
+	DATABASE_USER_CLI                 = "database-user"
+	DATABASE_PASSWORD_CLI             = "database-password"
+	DATABASE_MAX_IDLE_CONNECTIONS_CLI = "database-max-idle"
+	DATABASE_MAX_OPEN_CONNECTIONS_CLI = "database-max-open"
+	DATABASE_MAX_CONN_LIFETIME_CLI    = "database-max-lifetime"
+)
@@ -35,6 +35,7 @@ import (
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
 	nodeinfo "github.com/ethereum/go-ethereum/statediff/indexer/node"
 	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/prom"
 	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
 )

@@ -224,6 +225,8 @@ func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, height *
 	}
 	// increment state node counter.
 	atomic.AddUint64(&p.stateNodeCounter, 1)
+	prom.IncStateNodeCount()
+
 	// increment current batch size counter
 	p.currBatchSize += 2
 	return err
@@ -250,6 +253,8 @@ func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, height
 	}
 	// increment storage node counter.
 	atomic.AddUint64(&p.storageNodeCounter, 1)
+	prom.IncStorageNodeCount()
+
 	// increment current batch size counter
 	p.currBatchSize += 2
 	return nil
@@ -269,6 +274,8 @@ func (p *publisher) PublishCode(height *big.Int, codeHash common.Hash, codeBytes
 	}
 	// increment code node counter.
 	atomic.AddUint64(&p.codeNodeCounter, 1)
+	prom.IncCodeNodeCount()
+
 	p.currBatchSize++
 	return nil
 }
@@ -9,8 +9,9 @@ import (
 	"path/filepath"
 	"testing"

+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
-	"github.com/jackc/pgx/v4"

 	fixt "github.com/vulcanize/ipld-eth-state-snapshot/fixture"
 	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
@@ -88,27 +89,23 @@ func TestPgCopy(t *testing.T) {
 	pub := writeFiles(t, dir)

 	ctx := context.Background()
-	conn, err := pgx.Connect(ctx, pgConfig.DbConnectionString())
+	driver, err := postgres.NewSQLXDriver(ctx, pgConfig, nodeInfo)
 	test.NoError(t, err)
+	db := postgres.NewPostgresDB(driver)

-	// clear existing test data
-	pgDeleteTable := `DELETE FROM %s`
-	for _, tbl := range allTables {
-		_, err = conn.Exec(ctx, fmt.Sprintf(pgDeleteTable, tbl.Name))
-		test.NoError(t, err)
-	}
+	sql.TearDownDB(t, db)

 	// copy from files
 	pgCopyStatement := `COPY %s FROM '%s' CSV`
 	for _, tbl := range perBlockTables {
 		stm := fmt.Sprintf(pgCopyStatement, tbl.Name, TableFile(pub.dir, tbl.Name))
-		_, err = conn.Exec(ctx, stm)
+		_, err = db.Exec(ctx, stm)
 		test.NoError(t, err)
 	}
 	for i := uint32(0); i < pub.txCounter; i++ {
 		for _, tbl := range perNodeTables {
 			stm := fmt.Sprintf(pgCopyStatement, tbl.Name, TableFile(pub.txDir(i), tbl.Name))
-			_, err = conn.Exec(ctx, stm)
+			_, err = db.Exec(ctx, stm)
 			test.NoError(t, err)
 		}
 	}
@@ -122,7 +119,7 @@ func TestPgCopy(t *testing.T) {
 		BlockHash string
 	}
 	var header res
-	err = conn.QueryRow(ctx, pgQueryHeader, fixt.Block1_Header.Number.Uint64()).Scan(
+	err = db.QueryRow(ctx, pgQueryHeader, fixt.Block1_Header.Number.Uint64()).Scan(
 		&header.CID, &header.BlockHash)
 	test.NoError(t, err)

@@ -0,0 +1,61 @@
+// Copyright © 2022 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 snapshot
+
+import (
+	"github.com/jmoiron/sqlx"
+	"github.com/sirupsen/logrus"
+	. "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
+)
+
+const (
+	stateSnapShotPgStr   = "SELECT state_snapshot($1, $2)"
+	storageSnapShotPgStr = "SELECT storage_snapshot($1, $2)"
+)
+
+type InPlaceSnapshotParams struct {
+	StartHeight uint64
+	EndHeight   uint64
+}
+
+func CreateInPlaceSnapshot(config *Config, params InPlaceSnapshotParams) error {
+	db, err := sqlx.Connect("postgres", config.DB.ConnConfig.DbConnectionString())
+	if err != nil {
+		return err
+	}
+
+	tx, err := db.Begin()
+	if err != nil {
+		return err
+	}
+
+	defer func() {
+		err = CommitOrRollback(tx, err)
+		if err != nil {
+			logrus.Errorf("CommitOrRollback failed: %s", err)
+		}
+	}()
+
+	if _, err = tx.Exec(stateSnapShotPgStr, params.StartHeight, params.EndHeight); err != nil {
+		return err
+	}
+
+	if _, err = tx.Exec(storageSnapShotPgStr, params.StartHeight, params.EndHeight); err != nil {
+		return err
+	}
+
+	return nil
+}
@@ -0,0 +1,160 @@
+// Copyright © 2022 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 snapshot
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
+	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
+	"github.com/ethereum/go-ethereum/statediff/indexer/models"
+	"github.com/multiformats/go-multihash"
+
+	fixt "github.com/vulcanize/ipld-eth-state-snapshot/fixture"
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot/pg"
+	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
+	"github.com/vulcanize/ipld-eth-state-snapshot/test"
+)
+
+var (
+	pgConfig       = test.DefaultPgConfig
+	nodeInfo       = test.DefaultNodeInfo
+	snapshotHeight = 4
+
+	allTables = []*snapt.Table{
+		&snapt.TableIPLDBlock,
+		&snapt.TableNodeInfo,
+		&snapt.TableHeader,
+		&snapt.TableStateNode,
+		&snapt.TableStorageNode,
+	}
+)
+
+func writeData(t *testing.T, db *postgres.DB) snapt.Publisher {
+	pub := pg.NewPublisher(db)
+	tx, err := pub.BeginTx()
+	test.NoError(t, err)
+
+	for _, block := range fixt.InPlaceSnapshotBlocks {
+		headerID := block.Hash.String()
+
+		for _, stateNode := range block.StateNodes {
+			test.NoError(t, pub.PublishStateNode(&stateNode, headerID, block.Number, tx))
+		}
+
+		for index, stateStorageNodes := range block.StorageNodes {
+			stateNode := block.StateNodes[index]
+
+			for _, storageNode := range stateStorageNodes {
+				test.NoError(t, pub.PublishStorageNode(&storageNode, headerID, block.Number, stateNode.Path, tx))
+			}
+		}
+
+	}
+
+	test.NoError(t, tx.Commit())
+
+	test.NoError(t, pub.PublishHeader(&fixt.Block4_Header))
+	return pub
+}
+
+func TestCreateInPlaceSnapshot(t *testing.T) {
+	test.NeedsDB(t)
+	ctx := context.Background()
+	driver, err := postgres.NewSQLXDriver(ctx, pgConfig, nodeInfo)
+	test.NoError(t, err)
+	db := postgres.NewPostgresDB(driver)
+
+	sql.TearDownDB(t, db)
+
+	_ = writeData(t, db)
+
+	params := InPlaceSnapshotParams{StartHeight: uint64(0), EndHeight: uint64(snapshotHeight)}
+	config := &Config{
+		Eth: &EthConfig{
+			NodeInfo: test.DefaultNodeInfo,
+		},
+		DB: &DBConfig{
+			URI:        pgConfig.DbConnectionString(),
+			ConnConfig: pgConfig,
+		},
+	}
+	err = CreateInPlaceSnapshot(config, params)
+	test.NoError(t, err)
+
+	// Check inplace snapshot was created for state_cids
+	stateNodes := make([]models.StateNodeModel, 0)
+	pgQueryStateCids := `SELECT cast(state_cids.block_number AS TEXT), state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id, state_cids.mh_key
+					  FROM eth.state_cids
+					  WHERE eth.state_cids.block_number = $1
+					  ORDER BY state_cids.state_path`
+	err = db.Select(ctx, &stateNodes, pgQueryStateCids, snapshotHeight)
+	test.NoError(t, err)
+	test.ExpectEqual(t, 4, len(stateNodes))
+	expectedStateNodes := fixt.ExpectedStateNodes
+
+	pgIpfsGet := `SELECT data FROM public.blocks
+					WHERE key = $1 AND block_number = $2`
+
+	for index, stateNode := range stateNodes {
+		var data []byte
+		err = db.Get(ctx, &data, pgIpfsGet, stateNode.MhKey, snapshotHeight)
+		test.NoError(t, err)
+
+		expectedStateNode := expectedStateNodes[index]
+		expectedCID, _ := ipld.RawdataToCid(ipld.MEthStateTrie, expectedStateNode.Value, multihash.KECCAK_256)
+		test.ExpectEqual(t, strconv.Itoa(snapshotHeight), stateNode.BlockNumber)
+		test.ExpectEqual(t, fixt.Block4_Header.Hash().String(), stateNode.HeaderID)
+		test.ExpectEqual(t, expectedCID.String(), stateNode.CID)
+		test.ExpectEqual(t, int(expectedStateNode.NodeType), stateNode.NodeType)
+		test.ExpectEqual(t, expectedStateNode.Key, common.HexToHash(stateNode.StateKey))
+		test.ExpectEqual(t, false, stateNode.Diff)
+		test.ExpectEqualBytes(t, expectedStateNode.Path, stateNode.Path)
+		test.ExpectEqualBytes(t, expectedStateNode.Value, data)
+	}
+
+	// Check inplace snapshot was created for storage_cids
+	storageNodes := make([]models.StorageNodeModel, 0)
+	pgQueryStorageCids := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, storage_cids.state_path, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path, storage_cids.mh_key, storage_cids.header_id
+					  FROM eth.storage_cids
+					  WHERE eth.storage_cids.block_number = $1
+					  ORDER BY storage_cids.state_path, storage_cids.storage_path`
+	err = db.Select(ctx, &storageNodes, pgQueryStorageCids, snapshotHeight)
+	test.NoError(t, err)
+
+	for index, storageNode := range storageNodes {
+		expectedStorageNode := fixt.ExpectedStorageNodes[index]
+		expectedStorageCID, _ := ipld.RawdataToCid(ipld.MEthStorageTrie, expectedStorageNode.Value, multihash.KECCAK_256)
+
+		test.ExpectEqual(t, strconv.Itoa(snapshotHeight), storageNode.BlockNumber)
+		test.ExpectEqual(t, fixt.Block4_Header.Hash().String(), storageNode.HeaderID)
+		test.ExpectEqual(t, expectedStorageCID.String(), storageNode.CID)
+		test.ExpectEqual(t, int(expectedStorageNode.NodeType), storageNode.NodeType)
+		test.ExpectEqual(t, expectedStorageNode.Key, common.HexToHash(storageNode.StorageKey))
+		test.ExpectEqual(t, expectedStorageNode.StatePath, storageNode.StatePath)
+		test.ExpectEqual(t, expectedStorageNode.Path, storageNode.Path)
+		test.ExpectEqual(t, false, storageNode.Diff)
+
+		var data []byte
+		err = db.Get(ctx, &data, pgIpfsGet, storageNode.MhKey, snapshotHeight)
+		test.NoError(t, err)
+		test.ExpectEqualBytes(t, expectedStorageNode.Value, data)
+	}
+}
@@ -28,12 +28,14 @@ import (
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	dshelp "github.com/ipfs/go-ipfs-ds-help"
 	"github.com/multiformats/go-multihash"
+	"github.com/sirupsen/logrus"
 	log "github.com/sirupsen/logrus"

 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
 	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/prom"
 	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
 )

@@ -117,7 +119,12 @@ func (p *publisher) PublishHeader(header *types.Header) (err error) {
 		return err
 	}
 	tx := pubTx{snapTx, nil}
-	defer func() { err = snapt.CommitOrRollback(tx, err) }()
+	defer func() {
+		err = snapt.CommitOrRollback(tx, err)
+		if err != nil {
+			logrus.Errorf("CommitOrRollback failed: %s", err)
+		}
+	}()

 	if _, err = tx.publishIPLD(headerNode.Cid(), headerNode.RawData(), header.Number); err != nil {
 		return err
@@ -151,6 +158,7 @@ func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, height *
 	}
 	// increment state node counter.
 	atomic.AddUint64(&p.stateNodeCounter, 1)
+	prom.IncStateNodeCount()

 	// increment current batch size counter
 	p.currBatchSize += 2
@@ -177,6 +185,7 @@ func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, height
 	}
 	// increment storage node counter.
 	atomic.AddUint64(&p.storageNodeCounter, 1)
+	prom.IncStorageNodeCount()

 	// increment current batch size counter
 	p.currBatchSize += 2
@@ -198,6 +207,7 @@ func (p *publisher) PublishCode(height *big.Int, codeHash common.Hash, codeBytes

 	// increment code node counter.
 	atomic.AddUint64(&p.codeNodeCounter, 1)
+	prom.IncCodeNodeCount()

 	p.currBatchSize++
 	return nil
@@ -2,12 +2,11 @@ package pg

 import (
 	"context"
-	"fmt"
 	"testing"

+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
-	"github.com/jackc/pgx/v4"

 	fixt "github.com/vulcanize/ipld-eth-state-snapshot/fixture"
 	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
@@ -27,10 +26,8 @@ var (
 	}
 )

-func writeData(t *testing.T) *publisher {
-	driver, err := postgres.NewPGXDriver(context.Background(), pgConfig, nodeInfo)
-	test.NoError(t, err)
-	pub := NewPublisher(postgres.NewPostgresDB(driver))
+func writeData(t *testing.T, db *postgres.DB) *publisher {
+	pub := NewPublisher(db)
 	test.NoError(t, pub.PublishHeader(&fixt.Block1_Header))
 	tx, err := pub.BeginTx()
 	test.NoError(t, err)
@@ -47,17 +44,13 @@ func TestBasic(t *testing.T) {
 	test.NeedsDB(t)

 	ctx := context.Background()
-	conn, err := pgx.Connect(ctx, pgConfig.DbConnectionString())
+	driver, err := postgres.NewSQLXDriver(ctx, pgConfig, nodeInfo)
 	test.NoError(t, err)
+	db := postgres.NewPostgresDB(driver)

-	// clear existing test data
-	pgDeleteTable := `DELETE FROM %s`
-	for _, tbl := range allTables {
-		_, err = conn.Exec(ctx, fmt.Sprintf(pgDeleteTable, tbl.Name))
-		test.NoError(t, err)
-	}
+	sql.TearDownDB(t, db)

-	_ = writeData(t)
+	_ = writeData(t, db)

 	// check header was successfully committed
 	pgQueryHeader := `SELECT cid, block_hash
@@ -68,7 +61,7 @@ func TestBasic(t *testing.T) {
 		BlockHash string
 	}
 	var header res
-	err = conn.QueryRow(ctx, pgQueryHeader, fixt.Block1_Header.Number.Uint64()).Scan(
+	err = db.QueryRow(ctx, pgQueryHeader, fixt.Block1_Header.Number.Uint64()).Scan(
 		&header.CID, &header.BlockHash)
 	test.NoError(t, err)

@@ -30,6 +30,7 @@ import (
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/trie"
+	"github.com/sirupsen/logrus"
 	log "github.com/sirupsen/logrus"

 	iter "github.com/vulcanize/go-eth-state-node-iterator"
@@ -56,8 +57,6 @@ type Service struct {
 }

 func NewLevelDB(con *EthConfig) (ethdb.Database, error) {
-	println(con.LevelDBPath)
-	println(con.AncientDBPath)
 	edb, err := rawdb.NewLevelDBDatabaseWithFreezer(
 		con.LevelDBPath, 1024, 256, con.AncientDBPath, "ipld-eth-state-snapshot", true,
 	)
@@ -104,18 +103,21 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 	if err != nil {
 		return err
 	}
+
 	headerID := header.Hash().String()
 	s.tracker = newTracker(s.recoveryFile, int(params.Workers))
-	go s.tracker.run()
-	go s.tracker.captureSignal()
+	s.tracker.captureSignal()

 	var iters []trie.NodeIterator
 	// attempt to restore from recovery file if it exists
 	iters, err = s.tracker.restore(tree)
 	if err != nil {
+		log.Errorf("restore error: %s", err.Error())
 		return err
 	}
+
 	if iters != nil {
+		log.Debugf("restored iterators; count: %d", len(iters))
 		if params.Workers < uint(len(iters)) {
 			return fmt.Errorf(
 				"number of recovered workers (%d) is greater than number configured (%d)",
@@ -123,6 +125,7 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 			)
 		}
 	} else { // nothing to restore
+		log.Debugf("no iterators to restore")
 		if params.Workers > 1 {
 			iters = iter.SubtrieIterators(tree, params.Workers)
 		} else {
@@ -136,7 +139,7 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 	defer func() {
 		err := s.tracker.haltAndDump()
 		if err != nil {
-			log.Error("failed to write recovery file: ", err)
+			log.Errorf("failed to write recovery file: %v", err)
 		}
 	}()

@@ -201,7 +204,12 @@ func (s *Service) createSnapshot(it trie.NodeIterator, headerID string, height *
 	if err != nil {
 		return err
 	}
-	defer func() { err = CommitOrRollback(tx, err) }()
+	defer func() {
+		err = CommitOrRollback(tx, err)
+		if err != nil {
+			logrus.Errorf("CommitOrRollback failed: %s", err)
+		}
+	}()

 	for it.Next(true) {
 		res, err := resolveNode(it, s.stateDB.TrieDB())
@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"

 	"github.com/golang/mock/gomock"

@@ -45,7 +46,8 @@ func TestCreateSnapshot(t *testing.T) {
 		pub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).
 			AnyTimes()
 		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-			Times(len(fixt.Block1_StateNodePaths))
+			// Use MinTimes as duplicate nodes are expected at boundaries
+			MinTimes(len(fixt.Block1_StateNodePaths))

 		// TODO: fixtures for storage node
 		// pub.EXPECT().PublishStorageNode(gomock.Eq(fixt.StorageNode), gomock.Eq(int64(0)), gomock.Any())
@@ -117,6 +119,9 @@ func TestRecovery(t *testing.T) {
 			t.Fatal("cannot stat recovery file:", err)
 		}

+		// Wait for earlier snapshot process to complete
+		time.Sleep(5 * time.Second)
+
 		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 		err = service.CreateSnapshot(params)
 		if err != nil {
@@ -22,7 +22,11 @@ type trackedIter struct {
 func (it *trackedIter) Next(descend bool) bool {
 	ret := it.NodeIterator.Next(descend)
 	if !ret {
-		it.tracker.stopChan <- it
+		if it.tracker.running {
+			it.tracker.stopChan <- it
+		} else {
+			log.Errorf("iterator stopped after tracker halted: path=%x", it.Path())
+		}
 	}
 	return ret
 }
@@ -34,9 +38,7 @@ type iteratorTracker struct {
 	stopChan  chan *trackedIter
 	started   map[*trackedIter]struct{}
 	stopped   []*trackedIter
-
-	haltChan chan struct{}
-	done     chan struct{}
+	running   bool
 }

 func newTracker(file string, buf int) iteratorTracker {
@@ -45,8 +47,7 @@ func newTracker(file string, buf int) iteratorTracker {
 		startChan:    make(chan *trackedIter, buf),
 		stopChan:     make(chan *trackedIter, buf),
 		started:      map[*trackedIter]struct{}{},
-		haltChan:     make(chan struct{}),
-		done:         make(chan struct{}),
+		running:      true,
 	}
 }

@@ -62,23 +63,7 @@ func (tr *iteratorTracker) captureSignal() {
 	}()
 }

-// listens for starts/stops and manages current state
-func (tr *iteratorTracker) run() {
-loop:
-	for {
-		select {
-		case start := <-tr.startChan:
-			tr.started[start] = struct{}{}
-		case stop := <-tr.stopChan:
-			tr.stopped = append(tr.stopped, stop)
-		case <-tr.haltChan:
-			break loop
-		default:
-		}
-	}
-	tr.done <- struct{}{}
-}
-
+// Wraps an iterator in a trackedIter. This should not be called once halts are possible.
 func (tr *iteratorTracker) tracked(it trie.NodeIterator) (ret *trackedIter) {
 	ret = &trackedIter{it, tr}
 	tr.startChan <- ret
@@ -87,7 +72,7 @@ func (tr *iteratorTracker) tracked(it trie.NodeIterator) (ret *trackedIter) {

 // dumps iterator path and bounds to a text file so it can be restored later
 func (tr *iteratorTracker) dump() error {
-	log.Info("Dumping recovery state to:", tr.recoveryFile)
+	log.Debug("Dumping recovery state to: ", tr.recoveryFile)
 	var rows [][]string
 	for it, _ := range tr.started {
 		var endPath []byte
@@ -118,6 +103,7 @@ func (tr *iteratorTracker) restore(tree state.Trie) ([]trie.NodeIterator, error)
 		}
 		return nil, err
 	}
+	log.Debug("Restoring recovery state from: ", tr.recoveryFile)
 	defer file.Close()
 	in := csv.NewReader(file)
 	in.FieldsPerRecord = 2
@@ -137,15 +123,19 @@ func (tr *iteratorTracker) restore(tree state.Trie) ([]trie.NodeIterator, error)
 			}
 		}

-		it := iter.NewPrefixBoundIterator(tree, paths[0], paths[1])
+		// Force the lower bound path to an even length
+		if len(paths[0])&0b1 == 1 {
+			decrementPath(paths[0]) // decrement first to avoid skipped nodes
+			paths[0] = append(paths[0], 0)
+		}
+		it := iter.NewPrefixBoundIterator(tree.NodeIterator(iter.HexToKeyBytes(paths[0])), paths[1])
 		ret = append(ret, tr.tracked(it))
 	}
 	return ret, nil
 }

 func (tr *iteratorTracker) haltAndDump() error {
-	tr.haltChan <- struct{}{}
-	<-tr.done
+	tr.running = false

 	// drain any pending events
 	close(tr.startChan)
@@ -6,6 +6,7 @@ import (

 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"

+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/prom"
 	file "github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot/file"
 	pg "github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot/pg"
 	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
@@ -18,9 +19,35 @@ func NewPublisher(mode SnapshotMode, config *Config) (snapt.Publisher, error) {
 		if err != nil {
 			return nil, err
 		}
+
+		prom.RegisterDBCollector(config.DB.ConnConfig.DatabaseName, driver)
+
 		return pg.NewPublisher(postgres.NewPostgresDB(driver)), nil
 	case FileSnapshot:
 		return file.NewPublisher(config.File.OutputDir, config.Eth.NodeInfo)
 	}
 	return nil, fmt.Errorf("invalid snapshot mode: %s", mode)
 }
+
+// Subtracts 1 from the last byte in a path slice, carrying if needed.
+// Does nothing, returning false, for all-zero inputs.
+func decrementPath(path []byte) bool {
+	// check for all zeros
+	allzero := true
+	for i := 0; i < len(path); i++ {
+		allzero = allzero && path[i] == 0
+	}
+	if allzero {
+		return false
+	}
+	for i := len(path) - 1; i >= 0; i-- {
+		val := path[i]
+		path[i]--
+		if val == 0 {
+			path[i] = 0xf
+		} else {
+			return true
+		}
+	}
+	return true
+}
@@ -7,7 +7,7 @@ var TableIPLDBlock = Table{
 		{"key", text},
 		{"data", bytea},
 	},
-	`ON CONFLICT DO NOTHING`,
+	"ON CONFLICT (key, block_number) DO NOTHING",
 }

 var TableNodeInfo = Table{
@@ -41,7 +41,7 @@ var TableHeader = Table{
 		{"times_validated", integer},
 		{"coinbase", varchar},
 	},
-	"ON CONFLICT DO NOTHING",
+	"ON CONFLICT (block_hash, block_number) DO UPDATE SET (parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase) = (EXCLUDED.parent_hash, EXCLUDED.cid, EXCLUDED.td, EXCLUDED.node_id, EXCLUDED.reward, EXCLUDED.state_root, EXCLUDED.tx_root, EXCLUDED.receipt_root, EXCLUDED.uncle_root, EXCLUDED.bloom, EXCLUDED.timestamp, EXCLUDED.mh_key, eth.header_cids.times_validated + 1, EXCLUDED.coinbase)",
 }

 var TableStateNode = Table{
@@ -56,7 +56,7 @@ var TableStateNode = Table{
 		{"diff", boolean},
 		{"mh_key", text},
 	},
-	`ON CONFLICT DO NOTHING`,
+	"ON CONFLICT (header_id, state_path, block_number) DO UPDATE SET (state_leaf_key, cid, node_type, diff, mh_key) = (EXCLUDED.state_leaf_key, EXCLUDED.cid, EXCLUDED.node_type, EXCLUDED.diff, EXCLUDED.mh_key)",
 }

 var TableStorageNode = Table{
@@ -72,5 +72,5 @@ var TableStorageNode = Table{
 		{"diff", boolean},
 		{"mh_key", text},
 	},
-	"ON CONFLICT DO NOTHING",
+	"ON CONFLICT (header_id, state_path, storage_path, block_number) DO UPDATE SET (storage_leaf_key, cid, node_type, diff, mh_key) = (EXCLUDED.storage_leaf_key, EXCLUDED.cid, EXCLUDED.node_type, EXCLUDED.diff, EXCLUDED.mh_key)",
 }
@@ -20,10 +20,10 @@ var (
 	}
 	DefaultPgConfig = postgres.Config{
 		Hostname:     "localhost",
-		Port:         5432,
-		DatabaseName: "vulcanize_test",
-		Username:     "vulcanize",
-		Password:     "vulcanize_password",
+		Port:         8077,
+		DatabaseName: "vulcanize_testing",
+		Username:     "vdbm",
+		Password:     "password",

 		MaxIdle:         0,
 		MaxConnLifetime: 0,
Author	SHA1	Message	Date
prathamesh0andnabarun	ac00ee280f	Upgrade geth (#51 ) * Fix Makefile for installing mockgen tool * Upgrade geth * Add sleep in the failing recovery test Co-authored-by: nabarun <nabarun@deepstacksoft.com>	2022-07-13 11:39:36 +05:30
prathamesh0andnabarun	b241bf05eb	Add command to take an in-place snapshot (#42 ) * Add command to take an in-place snapshot * Add test data for in place snapshot unit test * Implement unit test for inplace snapshot * Add check for storage IPLD * Run unit tests sequentially * Add github workflow for unit test * Add missing checks for state and storage cid fields * Add more storage nodes to test * Update ipld-eth-db version for tests * Add comments for inplace snapshot test data * Add in-place snapshot cmd in readme * Implement defer pattern for db transaction * Log transaction commit or rollback error Co-authored-by: nabarun <nabarun@deepstacksoft.com>	2022-06-15 12:51:26 +05:30
Ashwin Phatak	c2ecea6d61	Merge pull request #45 from vulcanize/ng-merge-fix-iterator Merge changes to fix iterator and update	2022-06-10 14:24:33 +05:30
nabarun	baef795834	Merge changes to fix iterator and update	2022-06-10 14:12:34 +05:30
nabarun	4979e4090d	Merge pull request #43 from roysc/fix-iterator-and-update Fix iterator and update	2022-06-10 14:03:43 +05:30
nabarun	59011cba3c	Update go-eth-state-node-iterator version	2022-06-10 14:02:13 +05:30
roysc	8371344756	clean up logging	2022-06-08 20:12:22 +08:00
roysc	ef391d8d31	simplify tracker, prevent send to closed chan	2022-06-08 20:12:22 +08:00
roysc	838dabcba6	update iterator mod	2022-06-08 20:12:22 +08:00
roysc	8572c36ee7	decrement path from restored iterator to cover node gaps	2022-06-08 20:12:22 +08:00
Ashwin Phatak	74f82a641d	Fix dependency	2022-06-08 20:05:38 +08:00
roysc	3e24972a89	try to fix hanging iterator	2022-06-08 20:05:38 +08:00
Ashwin Phatak	7129044eae	Log progress info	2022-06-08 18:32:10 +08:00
Abdul Rabbani	c9d398b5af	Merge pull request #40 from vulcanize/feature/update-go-geth-sharding Update go and geth versions	2022-06-01 09:22:04 -04:00
Abdul Rabbani	8aab6a5137	Update geth versions	2022-06-01 09:20:45 -04:00
Abdul Rabbani	cd2559e434	Merge branch 'master' into feature/update-go-geth-sharding	2022-06-01 09:19:18 -04:00
Abdul Rabbani	d213644ff4	Merge pull request #39 from vulcanize/feature/update-go-geth Updating Go and Geth Version	2022-06-01 09:10:35 -04:00
Abdul Rabbani	a04777cb78	Updating Go and Geth Version Ran Tests: ``` 09:07:50:~/GitHub/cerc/ipld-eth-state-snapshot go test ./... -v # github.com/vulcanize/ipld-eth-state-snapshot/cmd cmd/stateSnapshot.go:47:3: (*github.com/sirupsen/logrus.Entry).Fatal call has possible formatting directive %v ? github.com/vulcanize/ipld-eth-state-snapshot [no test files] === RUN TestCreateSnapshot === RUN TestCreateSnapshot/case time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" === RUN TestCreateSnapshot/case#01 time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" === RUN TestCreateSnapshot/case#02 time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" === RUN TestCreateSnapshot/case#03 time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" --- PASS: TestCreateSnapshot (0.20s) --- PASS: TestCreateSnapshot/case (0.15s) --- PASS: TestCreateSnapshot/case#01 (0.01s) --- PASS: TestCreateSnapshot/case#02 (0.02s) --- PASS: TestCreateSnapshot/case#03 (0.03s) === RUN TestRecovery === RUN TestRecovery/case time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" time="2022-06-01T09:08:03-04:00" level=info msg="Dumping recovery state to: /var/folders/xf/cg688d9j7f39ktmd_4ttwxqh0000gn/T/TestRecoverycase3574218035/001/recover.csv" time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" === RUN TestRecovery/case#01 time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" time="2022-06-01T09:08:03-04:00" level=info msg="Dumping recovery state to: /var/folders/xf/cg688d9j7f39ktmd_4ttwxqh0000gn/T/TestRecoverycase#011814952517/001/recover.csv" time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" === RUN TestRecovery/case#02 time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" time="2022-06-01T09:08:03-04:00" level=info msg="Dumping recovery state to: /var/folders/xf/cg688d9j7f39ktmd_4ttwxqh0000gn/T/TestRecoverycase#021650290358/001/recover.csv" time="2022-06-01T09:08:03-04:00" level=info msg="Creating snapshot at height 1" time="2022-06-01T09:08:03-04:00" level=info msg="head hash: 0xa7684ac44d48494670b2e0d9085b7750e7341620f0a271db146ed5e70c1db854 head height: 1" --- PASS: TestRecovery (0.40s) --- PASS: TestRecovery/case (0.12s) --- PASS: TestRecovery/case#01 (0.19s) --- PASS: TestRecovery/case#02 (0.09s) PASS ok github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot 1.109s === RUN TestWriting --- PASS: TestWriting (0.00s) === RUN TestPgCopy publisher_test.go:84: set TEST_WITH_DB to enable test --- SKIP: TestPgCopy (0.00s) PASS ok github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot/file 0.279s === RUN TestBasic publisher_test.go:47: set TEST_WITH_DB to enable test --- SKIP: TestBasic (0.00s) PASS ok github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot/pg 0.647s ```	2022-06-01 09:09:19 -04:00
Ashwin Phatak	c785c12274	Merge pull request #35 from vulcanize/ng-file-ethereum-nodeId Add ethereum config to fix db import in file mode	2022-05-25 18:38:09 +05:30
nabarun	23c4d22885	Add ethereum config to fix db import in file mode	2022-05-25 18:15:08 +05:30
Ashwin Phatak	d3a149899b	Merge pull request #34 from deep-stack/pm-merge-metrics Merge changes to add env bindings and prometheus metrics	2022-05-23 18:14:06 +05:30
prathamesh	dc27487b66	Merge changes to add env bindings and prometheus metrics	2022-05-23 17:11:51 +05:30
prathamesh0	cce45bda57	Add prometheus metrics collection (#33 ) * Upgrade geth * Add prometheus metrics collection * Update README	2022-05-23 16:56:48 +05:30
Ian Norden	751e064f8f	Merge pull request #28 from vulcanize/release-v3.0.2 add env bindings	2022-05-17 22:36:08 -05:00
i-norden	004f3e6665	constants for CLI binding strings	2022-05-17 22:33:20 -05:00
prathamesh0	0a1ea3ae4a	Fix insert statements and upgrade dependencies (#31 )	2022-05-17 19:19:52 +05:30
i-norden	31105a009c	add env bindings	2022-04-04 14:35:18 -05:00