Merge pull request #69 from cerc-io/ian/v4_dev

Update to use v4 vdb geth v1.11.5
missing .meta files in fixtures
2023-03-31 12:50:23 -05:00 · 2023-03-31 12:36:30 -05:00 · 2023-03-31 10:39:27 -05:00 · 2023-03-31 10:39:13 -05:00 · 2023-03-31 10:39:00 -05:00 · 2022-11-08 14:11:11 -05:00
66 changed files with 2023 additions and 558 deletions
@@ -15,7 +15,7 @@ jobs:

      - uses: actions/setup-go@v3
        with:
-          go-version: ">=1.18.0"
+          go-version: ">=1.19.0"
          check-latest: true

      - name: Checkout code
@@ -3,3 +3,4 @@
 ipld-eth-state-snapshot
 mocks/
 .vscode
+output_dir*/
@@ -22,6 +22,7 @@ Config format:
    workers      = 4                # degree of concurrency, the state trie is subdivided into sections that are traversed and processed concurrently
    blockHeight  = -1               # blockheight to perform the snapshot at (-1 indicates to use the latest blockheight found in leveldb)
    recoveryFile = "recovery_file"  # specifies a file to output recovery information on error or premature closure
+    accounts = []                   # list of accounts (addresses) to take the snapshot for # SNAPSHOT_ACCOUNTS

 [leveldb]
    # path to geth leveldb
@@ -72,6 +73,17 @@ Config format:
    ./ipld-eth-state-snapshot stateSnapshot --config={path to toml config file}
    ```

+    * Account selective snapshot: To restrict the snapshot to a list of accounts (addresses), provide the addresses in config parameter `snapshot.accounts` or env variable `SNAPSHOT_ACCOUNTS`. Only nodes related to provided addresses will be indexed.
+
+        Example:
+
+        ```toml
+        [snapshot]
+            accounts = [
+                "0x825a6eec09e44Cb0fa19b84353ad0f7858d7F61a"
+            ]
+        ```
+
 * For in-place snapshot in the database:

    ```bash
@@ -179,3 +191,19 @@ Config format:
    ```

 * NOTE: `COPY` command on CSVs inserts empty strings as `NULL` in the DB. Passing `FORCE_NOT_NULL <COLUMN_NAME>` forces it to insert empty strings instead. This is required to maintain compatibility of the imported snapshot data with the data generated by statediffing. Reference: https://www.postgresql.org/docs/14/sql-copy.html
+
+### Troubleshooting
+
+* Run the following command to find any rows (in data dumps in `file` mode) having unexpected number of columns:
+
+    ```bash
+    ./scripts/find-bad-rows.sh -i <input-file> -c <expected-columns> -o [output-file] -d true
+    ```
+
+* Run the following command to select rows (from data dumps in `file` mode) other than the ones having unexpected number of columns:
+
+    ```bash
+    ./scripts/filter-bad-rows.sh -i <input-file> -c <expected-columns> -o <output-file>
+    ```
+
+* See [scripts](./scripts) for more details.
@@ -21,7 +21,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"

-	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot"
+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/snapshot"
 )

 // inPlaceStateSnapshotCmd represents the inPlaceStateSnapshot command
@@ -25,8 +25,8 @@ import (
 	"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"
+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/prom"
+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/snapshot"
 )

 var (
@@ -22,7 +22,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"

-	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot"
+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/snapshot"
 )

 // stateSnapshotCmd represents the stateSnapshot command
@@ -69,13 +69,12 @@ func stateSnapshot() {
 		logWithCommand.Fatal(err)
 	}
 	workers := viper.GetUint(snapshot.SNAPSHOT_WORKERS_TOML)
-
 	if height < 0 {
-		if err := snapshotService.CreateLatestSnapshot(workers); err != nil {
+		if err := snapshotService.CreateLatestSnapshot(workers, config.Service.AllowedAccounts); err != nil {
 			logWithCommand.Fatal(err)
 		}
 	} else {
-		params := snapshot.SnapshotParams{Workers: workers, Height: uint64(height)}
+		params := snapshot.SnapshotParams{Workers: workers, Height: uint64(height), WatchedAddresses: config.Service.AllowedAccounts}
 		if err := snapshotService.CreateSnapshot(params); err != nil {
 			logWithCommand.Fatal(err)
 		}
@@ -93,6 +92,7 @@ func init() {
 	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")
+	stateSnapshotCmd.PersistentFlags().StringArray(snapshot.SNAPSHOT_ACCOUNTS_CLI, nil, "list of account addresses to limit snapshot to")

 	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))
@@ -101,4 +101,5 @@ func init() {
 	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))
+	viper.BindPFlag(snapshot.SNAPSHOT_ACCOUNTS_TOML, stateSnapshotCmd.PersistentFlags().Lookup(snapshot.SNAPSHOT_ACCOUNTS_CLI))
 }
@@ -5,7 +5,7 @@ services:
    restart: on-failure
    depends_on:
      - ipld-eth-db
-    image: vulcanize/ipld-eth-db:v4.2.1-alpha
+    image: git.vdb.to/cerc-io/ipld-eth-db/ipld-eth-db:v4.2.1-alpha-unstable
    environment:
      DATABASE_USER: "vdbm"
      DATABASE_NAME: "vulcanize_testing"
@@ -0,0 +1,7 @@
+*/*.log
+*/CURRENT*
+*/LOCK
+*/LOG
+*/MANIFEST-*
+*/ancient/FLOCK
+*/ancient/*.meta
@@ -0,0 +1 @@
+MANIFEST-000005
@@ -0,0 +1,15 @@
+=============== Mar 31, 2023 (CDT) ===============
+12:22:01.281499 log@legend F·NumFile S·FileSize N·Entry C·BadEntry B·BadBlock Ke·KeyError D·DroppedEntry L·Level Q·SeqNum T·TimeElapsed
+12:22:01.281586 table@recovery F·2
+12:22:01.281592 table@recovery recovering @2
+12:22:01.281901 table@recovery recovered @2 Gk·13 Ck·0 Cb·0 S·918 Q·13
+12:22:01.281909 table@recovery recovering @4
+12:22:01.282725 table@recovery recovered @4 Gk·416 Ck·0 Cb·0 S·35872 Q·430
+12:22:01.282737 table@recovery recovered F·2 N·429 Gk·429 Ck·0 Q·430
+12:22:01.298688 version@stat F·[2] S·35KiB[35KiB] Sc·[0.50]
+12:22:01.308338 db@open opening
+12:22:01.308738 version@stat F·[2] S·35KiB[35KiB] Sc·[0.50]
+12:22:01.314746 db@janitor F·4 G·0
+12:22:01.314778 db@open done T·6.427167ms
+12:22:01.317510 db@close closing
+12:22:01.317545 db@close done T·34.833µs
@@ -0,0 +1 @@
+��
@@ -0,0 +1 @@
+��
@@ -0,0 +1 @@
+��
@@ -0,0 +1 @@
+��
@@ -0,0 +1 @@
+��
@@ -3,25 +3,25 @@ package fixture
 import (
 	"os"
 	"path/filepath"
-	"runtime"
 )

 // TODO: embed some mainnet data
 // import "embed"
 //_go:embed mainnet_data.tar.gz

-var (
-	ChaindataPath, AncientdataPath string
-)
+// GetChainDataPath returns the absolute paths to chain data in 'fixture/' given the chain (chain | chain2)
+func GetChainDataPath(chain string) (string, string) {
+	path := filepath.Join("..", "..", "fixture", chain)

-func init() {
-	_, path, _, _ := runtime.Caller(0)
-	wd := filepath.Dir(path)
-
-	ChaindataPath = filepath.Join(wd, "..", "fixture", "chaindata")
-	AncientdataPath = filepath.Join(ChaindataPath, "ancient")
-
-	if _, err := os.Stat(ChaindataPath); err != nil {
-		panic("must populate chaindata at " + ChaindataPath)
+	chaindataPath, err := filepath.Abs(path)
+	if err != nil {
+		panic("cannot resolve path " + path)
 	}
+	ancientdataPath := filepath.Join(chaindataPath, "ancient")
+
+	if _, err := os.Stat(chaindataPath); err != nil {
+		panic("must populate chaindata at " + chaindataPath)
+	}
+
+	return chaindataPath, ancientdataPath
 }
@@ -1,6 +0,0 @@
-*.log
-CURRENT*
-LOCK
-LOG
-MANIFEST-*
-ancient/FLOCK
@@ -0,0 +1 @@
+MANIFEST-000003
@@ -0,0 +1,13 @@
+=============== Mar 31, 2023 (CDT) ===============
+12:22:01.090151 log@legend F·NumFile S·FileSize N·Entry C·BadEntry B·BadBlock Ke·KeyError D·DroppedEntry L·Level Q·SeqNum T·TimeElapsed
+12:22:01.090565 table@recovery F·1
+12:22:01.090576 table@recovery recovering @2
+12:22:01.106640 table@recovery recovered @2 Gk·146827 Ck·0 Cb·0 S·5765300 Q·146827
+12:22:01.106661 table@recovery recovered F·1 N·146827 Gk·146827 Ck·0 Q·146827
+12:22:01.133566 version@stat F·[1] S·5MiB[5MiB] Sc·[0.25]
+12:22:01.145740 db@open opening
+12:22:01.146636 version@stat F·[1] S·5MiB[5MiB] Sc·[0.25]
+12:22:01.151049 db@janitor F·3 G·0
+12:22:01.151090 db@open done T·5.315792ms
+12:22:01.161329 db@close closing
+12:22:01.161389 db@close done T·58.667µs
@@ -0,0 +1 @@
+��
@@ -0,0 +1 @@
+��
@@ -0,0 +1 @@
+��
@@ -0,0 +1 @@
+��
@@ -0,0 +1 @@
+��
@@ -18,9 +18,9 @@ package fixture
 import (
 	"math/big"

+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
 	"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 {
@@ -6,7 +6,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"

-	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
 )

 var Block1_Header = types.Header{
@@ -31,6 +31,283 @@ var Block1_Header = types.Header{
 var Block1_StateNode0 = snapt.Node{
 	NodeType: 0,
 	Path:     []byte{12, 0},
-	Key:      common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000"),
+	Key:      common.Hash{},
 	Value:    []byte{248, 113, 160, 147, 141, 92, 6, 119, 63, 191, 125, 121, 193, 230, 153, 223, 49, 102, 109, 236, 50, 44, 161, 215, 28, 224, 171, 111, 118, 230, 79, 99, 18, 99, 4, 160, 117, 126, 95, 187, 60, 115, 90, 36, 51, 167, 59, 86, 20, 175, 63, 118, 94, 230, 107, 202, 41, 253, 234, 165, 214, 221, 181, 45, 9, 202, 244, 148, 128, 128, 128, 128, 128, 128, 128, 128, 128, 128, 128, 128, 128, 160, 247, 170, 155, 102, 71, 245, 140, 90, 255, 89, 193, 131, 99, 31, 85, 161, 78, 90, 0, 204, 46, 253, 15, 71, 120, 19, 109, 123, 255, 0, 188, 27, 128},
 }
+
+// Header for last block at height 32
+var Chain2_Block32_Header = types.Header{
+	ParentHash:  common.HexToHash("0x6983c921c053d1f637449191379f61ba844013c71e5ebfacaff77f8a8bd97042"),
+	UncleHash:   common.HexToHash("0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347"),
+	Coinbase:    common.HexToAddress("0x0000000000000000000000000000000000000000"),
+	Root:        common.HexToHash("0xeaa5866eb37e33fc3cfe1376b2ad7f465e7213c14e6834e1cfcef9552b2e5d5d"),
+	TxHash:      common.HexToHash("0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"),
+	ReceiptHash: common.HexToHash("0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"),
+	Bloom:       types.Bloom{},
+	Difficulty:  big.NewInt(2),
+	Number:      big.NewInt(32),
+	GasLimit:    8253773,
+	GasUsed:     0,
+	Time:        1658408469,
+	Extra:       []byte{216, 131, 1, 10, 19, 132, 103, 101, 116, 104, 136, 103, 111, 49, 46, 49, 56, 46, 50, 133, 108, 105, 110, 117, 120, 0, 0, 0, 0, 0, 0, 0, 113, 250, 240, 25, 148, 32, 193, 94, 196, 10, 99, 63, 251, 130, 170, 0, 176, 201, 149, 55, 230, 58, 218, 112, 84, 153, 122, 83, 134, 52, 176, 99, 53, 54, 63, 12, 226, 81, 38, 176, 57, 117, 92, 205, 237, 81, 203, 232, 220, 228, 166, 254, 206, 136, 7, 253, 2, 61, 47, 217, 235, 24, 140, 92, 1},
+	MixDigest:   common.Hash{},
+	Nonce:       types.BlockNonce{},
+	BaseFee:     nil,
+}
+
+// State nodes for all paths at height 32
+// Total 7
+var Chain2_Block32_StateNodes = []snapt.Node{
+	{
+		NodeType: 0,
+		Path:     []byte{},
+		Key:      common.Hash{},
+		Value:    []byte{248, 145, 128, 128, 128, 160, 151, 6, 152, 177, 246, 151, 39, 79, 71, 219, 192, 153, 253, 0, 46, 66, 56, 238, 116, 176, 237, 244, 79, 132, 49, 29, 30, 82, 108, 53, 191, 204, 128, 128, 160, 46, 224, 200, 157, 30, 24, 225, 92, 222, 131, 123, 169, 124, 86, 228, 124, 79, 136, 236, 83, 185, 22, 67, 136, 5, 73, 46, 110, 136, 138, 101, 63, 128, 128, 160, 104, 220, 31, 84, 240, 26, 100, 148, 110, 49, 52, 120, 81, 119, 30, 251, 196, 107, 11, 134, 124, 238, 93, 61, 109, 109, 181, 208, 10, 189, 17, 92, 128, 128, 160, 171, 149, 11, 254, 75, 39, 224, 164, 133, 151, 153, 47, 109, 134, 15, 169, 139, 206, 132, 93, 220, 210, 0, 225, 235, 118, 121, 247, 173, 12, 135, 133, 128, 128, 128, 128},
+	},
+	{
+		NodeType: 0,
+		Path:     []byte{3},
+		Key:      common.Hash{},
+		Value:    []byte{248, 81, 128, 128, 128, 160, 209, 34, 171, 171, 30, 147, 168, 199, 137, 152, 249, 118, 14, 166, 1, 169, 116, 224, 82, 196, 237, 83, 255, 188, 228, 197, 7, 178, 144, 137, 77, 55, 128, 128, 128, 128, 128, 160, 135, 96, 108, 173, 177, 63, 201, 196, 26, 204, 72, 118, 17, 30, 76, 117, 155, 63, 68, 187, 4, 249, 78, 69, 161, 82, 178, 234, 164, 48, 158, 173, 128, 128, 128, 128, 128, 128, 128},
+	},
+	{
+		NodeType: 2,
+		Path:     []byte{3, 3},
+		Key:      common.HexToHash("0x33153abc667e873b6036c8a46bdd847e2ade3f89b9331c78ef2553fea194c50d"),
+		Value:    []byte{248, 105, 160, 32, 21, 58, 188, 102, 126, 135, 59, 96, 54, 200, 164, 107, 221, 132, 126, 42, 222, 63, 137, 185, 51, 28, 120, 239, 37, 83, 254, 161, 148, 197, 13, 184, 70, 248, 68, 1, 128, 160, 168, 127, 48, 6, 204, 116, 51, 247, 216, 182, 191, 182, 185, 124, 223, 202, 239, 15, 67, 91, 253, 165, 42, 2, 54, 10, 211, 250, 242, 149, 205, 139, 160, 224, 22, 140, 8, 116, 27, 79, 113, 64, 185, 215, 180, 38, 38, 236, 164, 5, 87, 211, 15, 88, 153, 138, 185, 94, 186, 125, 137, 164, 198, 141, 192},
+	},
+	{
+		NodeType: 2,
+		Path:     []byte{3, 9},
+		Key:      common.HexToHash("0x39fc293fc702e42b9c023f094826545db42fc0fdf2ba031bb522d5ef917a6edb"),
+		Value:    []byte{248, 105, 160, 32, 252, 41, 63, 199, 2, 228, 43, 156, 2, 63, 9, 72, 38, 84, 93, 180, 47, 192, 253, 242, 186, 3, 27, 181, 34, 213, 239, 145, 122, 110, 219, 184, 70, 248, 68, 1, 128, 160, 25, 80, 158, 144, 166, 222, 32, 247, 189, 42, 34, 60, 40, 240, 56, 105, 251, 184, 132, 209, 219, 59, 60, 16, 221, 204, 228, 74, 76, 113, 37, 226, 160, 224, 22, 140, 8, 116, 27, 79, 113, 64, 185, 215, 180, 38, 38, 236, 164, 5, 87, 211, 15, 88, 153, 138, 185, 94, 186, 125, 137, 164, 198, 141, 192},
+	},
+	{
+		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, 10, 141, 1, 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, 54, 174, 96, 33, 243, 186, 113, 120, 188, 222, 254, 210, 63, 40, 4, 130, 154, 156, 66, 247, 130, 93, 88, 113, 144, 78, 47, 252, 174, 140, 130, 45, 160, 29, 80, 58, 104, 206, 141, 36, 93, 124, 217, 67, 93, 183, 43, 71, 98, 114, 126, 124, 105, 229, 48, 218, 194, 109, 83, 20, 76, 13, 102, 156, 130},
+	},
+	{
+		NodeType: 2,
+		Path:     []byte{12},
+		Key:      common.HexToHash("0xcabc5edb305583e33f66322ceee43088aa99277da772feb5053512d03a0a702b"),
+		Value:    []byte{248, 105, 160, 58, 188, 94, 219, 48, 85, 131, 227, 63, 102, 50, 44, 238, 228, 48, 136, 170, 153, 39, 125, 167, 114, 254, 181, 5, 53, 18, 208, 58, 10, 112, 43, 184, 70, 248, 68, 1, 128, 160, 54, 174, 96, 33, 243, 186, 113, 120, 188, 222, 254, 210, 63, 40, 4, 130, 154, 156, 66, 247, 130, 93, 88, 113, 144, 78, 47, 252, 174, 140, 130, 45, 160, 29, 80, 58, 104, 206, 141, 36, 93, 124, 217, 67, 93, 183, 43, 71, 98, 114, 126, 124, 105, 229, 48, 218, 194, 109, 83, 20, 76, 13, 102, 156, 130},
+	},
+}
+
+// Storage nodes for all paths at height 32
+// Total 18
+var Chain2_Block32_StorageNodes = []StorageNodeWithState{
+	{
+		Node: snapt.Node{
+			NodeType: 0,
+			Path:     []byte{},
+			Key:      common.HexToHash(""),
+			Value:    []byte{248, 145, 128, 128, 128, 128, 160, 46, 77, 227, 140, 57, 224, 108, 238, 40, 82, 145, 79, 210, 174, 54, 248, 0, 145, 137, 64, 229, 230, 148, 145, 250, 132, 89, 198, 8, 249, 245, 133, 128, 160, 146, 250, 117, 217, 106, 75, 51, 124, 196, 244, 29, 16, 47, 173, 5, 90, 86, 19, 15, 48, 179, 174, 60, 171, 112, 154, 92, 70, 232, 164, 141, 165, 128, 160, 107, 250, 27, 137, 190, 180, 7, 172, 62, 97, 13, 157, 215, 114, 55, 219, 14, 244, 163, 155, 192, 255, 34, 143, 154, 149, 33, 227, 166, 135, 164, 93, 128, 128, 128, 160, 173, 131, 221, 2, 30, 147, 11, 230, 58, 166, 18, 25, 90, 56, 198, 126, 196, 130, 131, 1, 213, 112, 129, 155, 96, 143, 121, 231, 218, 97, 216, 200, 128, 128, 128, 128},
+		},
+		StatePath: []byte{3, 3},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 0,
+			Path:     []byte{4},
+			Key:      common.HexToHash(""),
+			Value:    []byte{248, 81, 160, 167, 145, 134, 15, 219, 140, 96, 62, 101, 242, 176, 129, 164, 160, 200, 221, 13, 1, 246, 167, 156, 45, 205, 192, 88, 236, 235, 80, 105, 178, 123, 2, 128, 128, 128, 128, 128, 128, 128, 128, 128, 128, 128, 128, 128, 160, 18, 136, 22, 150, 26, 170, 67, 152, 182, 246, 95, 49, 193, 199, 219, 163, 97, 25, 243, 70, 126, 235, 163, 59, 44, 16, 37, 37, 247, 50, 229, 70, 128, 128},
+		},
+		StatePath: []byte{3, 3},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{4, 0},
+			Key:      common.HexToHash("0x405787fa12a823e0f2b7631cc41b3ba8828b3321ca811111fa75cd3aa3bb5ace"),
+			Value:    []byte{236, 160, 32, 87, 135, 250, 18, 168, 35, 224, 242, 183, 99, 28, 196, 27, 59, 168, 130, 139, 51, 33, 202, 129, 17, 17, 250, 117, 205, 58, 163, 187, 90, 206, 138, 137, 54, 53, 201, 173, 197, 222, 160, 0, 0},
+		},
+		StatePath: []byte{3, 3},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{4, 14},
+			Key:      common.HexToHash("0x4e2cec6f4784547e5042a1638086e3188929f34f3c0005f8dec366c96e8195ac"),
+			Value:    []byte{226, 160, 32, 44, 236, 111, 71, 132, 84, 126, 80, 66, 161, 99, 128, 134, 227, 24, 137, 41, 243, 79, 60, 0, 5, 248, 222, 195, 102, 201, 110, 129, 149, 172, 100},
+		},
+		StatePath: []byte{3, 3},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{6},
+			Key:      common.HexToHash("0x6aa02a11dd4d2597318b71d493b145ddf6ae0817a9d3947f45d529a6a75f2bef"),
+			Value:    []byte{236, 160, 58, 160, 42, 17, 221, 77, 37, 151, 49, 139, 113, 212, 147, 177, 69, 221, 246, 174, 8, 23, 169, 211, 148, 127, 69, 213, 41, 166, 167, 95, 43, 239, 138, 137, 54, 53, 201, 173, 197, 222, 159, 255, 156},
+		},
+		StatePath: []byte{3, 3},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{8},
+			Key:      common.HexToHash("0x8a35acfbc15ff81a39ae7d344fd709f28e8600b4aa8c65c6b64bfe7fe36bd19b"),
+			Value:    []byte{248, 67, 160, 58, 53, 172, 251, 193, 95, 248, 26, 57, 174, 125, 52, 79, 215, 9, 242, 142, 134, 0, 180, 170, 140, 101, 198, 182, 75, 254, 127, 227, 107, 209, 155, 161, 160, 71, 76, 68, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 6},
+		},
+		StatePath: []byte{3, 9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{8},
+			Key:      common.HexToHash("0x8a35acfbc15ff81a39ae7d344fd709f28e8600b4aa8c65c6b64bfe7fe36bd19b"),
+			Value:    []byte{248, 67, 160, 58, 53, 172, 251, 193, 95, 248, 26, 57, 174, 125, 52, 79, 215, 9, 242, 142, 134, 0, 180, 170, 140, 101, 198, 182, 75, 254, 127, 227, 107, 209, 155, 161, 160, 71, 76, 68, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 6},
+		},
+		StatePath: []byte{3, 3},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{12},
+			Key:      common.HexToHash("0xc2575a0e9e593c00f959f8c92f12db2869c3395a3b0502d05e2516446f71f85b"),
+			Value:    []byte{248, 67, 160, 50, 87, 90, 14, 158, 89, 60, 0, 249, 89, 248, 201, 47, 18, 219, 40, 105, 195, 57, 90, 59, 5, 2, 208, 94, 37, 22, 68, 111, 113, 248, 91, 161, 160, 71, 111, 108, 100, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8},
+		},
+		StatePath: []byte{3, 9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{12},
+			Key:      common.HexToHash("0xc2575a0e9e593c00f959f8c92f12db2869c3395a3b0502d05e2516446f71f85b"),
+			Value:    []byte{248, 67, 160, 50, 87, 90, 14, 158, 89, 60, 0, 249, 89, 248, 201, 47, 18, 219, 40, 105, 195, 57, 90, 59, 5, 2, 208, 94, 37, 22, 68, 111, 113, 248, 91, 161, 160, 71, 111, 108, 100, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8},
+		},
+		StatePath: []byte{3, 3},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 0,
+			Path:     []byte{},
+			Key:      common.HexToHash(""),
+			Value:    []byte{248, 145, 128, 128, 128, 128, 160, 145, 86, 15, 219, 52, 36, 164, 68, 160, 227, 156, 111, 1, 245, 112, 184, 187, 242, 26, 138, 8, 98, 129, 35, 57, 212, 165, 21, 204, 151, 229, 43, 128, 160, 250, 205, 84, 126, 141, 108, 126, 228, 162, 8, 238, 234, 141, 159, 232, 175, 70, 112, 207, 55, 165, 209, 107, 153, 54, 183, 60, 172, 194, 251, 66, 61, 128, 160, 107, 250, 27, 137, 190, 180, 7, 172, 62, 97, 13, 157, 215, 114, 55, 219, 14, 244, 163, 155, 192, 255, 34, 143, 154, 149, 33, 227, 166, 135, 164, 93, 128, 128, 128, 160, 173, 131, 221, 2, 30, 147, 11, 230, 58, 166, 18, 25, 90, 56, 198, 126, 196, 130, 131, 1, 213, 112, 129, 155, 96, 143, 121, 231, 218, 97, 216, 200, 128, 128, 128, 128},
+		},
+		StatePath: []byte{3, 9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{4},
+			Key:      common.HexToHash("0x405787fa12a823e0f2b7631cc41b3ba8828b3321ca811111fa75cd3aa3bb5ace"),
+			Value:    []byte{236, 160, 48, 87, 135, 250, 18, 168, 35, 224, 242, 183, 99, 28, 196, 27, 59, 168, 130, 139, 51, 33, 202, 129, 17, 17, 250, 117, 205, 58, 163, 187, 90, 206, 138, 137, 54, 53, 201, 173, 197, 222, 160, 0, 0},
+		},
+		StatePath: []byte{3, 9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{6},
+			Key:      common.HexToHash("0x6aa02a11dd4d2597318b71d493b145ddf6ae0817a9d3947f45d529a6a75f2bef"),
+			Value:    []byte{236, 160, 58, 160, 42, 17, 221, 77, 37, 151, 49, 139, 113, 212, 147, 177, 69, 221, 246, 174, 8, 23, 169, 211, 148, 127, 69, 213, 41, 166, 167, 95, 43, 239, 138, 137, 54, 53, 201, 173, 197, 222, 160, 0, 0},
+		},
+		StatePath: []byte{3, 9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 0,
+			Path:     []byte{},
+			Key:      common.HexToHash(""),
+			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, 82, 154, 228, 80, 107, 126, 132, 72, 3, 170, 88, 197, 100, 216, 50, 21, 226, 183, 86, 42, 208, 239, 184, 183, 152, 93, 188, 113, 224, 234, 218, 43, 128, 128, 128, 128, 128},
+		},
+		StatePath: []byte{12},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 0,
+			Path:     []byte{},
+			Key:      common.HexToHash(""),
+			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, 82, 154, 228, 80, 107, 126, 132, 72, 3, 170, 88, 197, 100, 216, 50, 21, 226, 183, 86, 42, 208, 239, 184, 183, 152, 93, 188, 113, 224, 234, 218, 43, 128, 128, 128, 128, 128},
+		},
+		StatePath: []byte{9},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{02},
+			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{12},
+	},
+	{
+		Node: snapt.Node{
+			NodeType: 2,
+			Path:     []byte{02},
+			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, 4},
+		},
+		StatePath: []byte{12},
+	},
+	{
+		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, 4},
+		},
+		StatePath: []byte{9},
+	},
+}
+
+// Contracts used in chain2
+/*
+pragma solidity ^0.8.0;
+
+contract Test {
+    uint256 private count;
+    uint256 private count2;
+
+    event Increment(uint256 count);
+
+    constructor() {
+      count2 = 4;
+    }
+
+    function incrementCount() public returns (uint256) {
+      count = count + 1;
+      emit Increment(count);
+
+      return count;
+    }
+
+    function destroy() public {
+        selfdestruct(payable(msg.sender));
+    }
+
+    function deleteCount2() public {
+        count2 = 0;
+    }
+}
+*/
+
+/*
+pragma solidity ^0.8.0;
+
+import "@openzeppelin/contracts/token/ERC20/ERC20.sol";
+
+contract GLDToken is ERC20 {
+    constructor(uint256 initialSupply) ERC20("Gold", "GLD") {
+        _mint(msg.sender, initialSupply);
+    }
+}
+*/
@@ -1,43 +1,49 @@
-module github.com/vulcanize/ipld-eth-state-snapshot
+module github.com/cerc-io/ipld-eth-state-snapshot

-go 1.18
+go 1.19

 require (
-	github.com/ethereum/go-ethereum v1.10.21
+	github.com/ethereum/go-ethereum v1.11.5
 	github.com/golang/mock v1.6.0
 	github.com/ipfs/go-cid v0.2.0
 	github.com/ipfs/go-ipfs-blockstore v1.2.0
 	github.com/ipfs/go-ipfs-ds-help v1.1.0
-	github.com/jmoiron/sqlx v1.2.0
+	github.com/jmoiron/sqlx v1.3.5
 	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.1.3
+	github.com/prometheus/client_golang v1.14.0
+	github.com/sirupsen/logrus v1.9.0
+	github.com/spf13/cobra v1.5.0
+	github.com/spf13/viper v1.12.0
+	golang.org/x/sync v0.1.0
 )

 require (
+	github.com/DataDog/zstd v1.5.2 // 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/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cockroachdb/errors v1.9.1 // indirect
+	github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect
+	github.com/cockroachdb/pebble v0.0.0-20230209160836-829675f94811 // indirect
+	github.com/cockroachdb/redact v1.1.3 // indirect
+	github.com/deckarep/golang-set/v2 v2.1.0 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
-	github.com/fsnotify/fsnotify v1.5.4 // indirect
-	github.com/georgysavva/scany v0.2.9 // indirect
-	github.com/go-kit/kit v0.10.0 // indirect
+	github.com/fsnotify/fsnotify v1.6.0 // indirect
+	github.com/georgysavva/scany v1.2.1 // indirect
+	github.com/getsentry/sentry-go v0.17.0 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
-	github.com/go-stack/stack v1.8.0 // indirect
+	github.com/go-stack/stack v1.8.1 // indirect
+	github.com/gofrs/flock v0.8.1 // 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/gorilla/websocket v1.4.2 // 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/holiman/uint256 v1.2.0 // 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
@@ -57,17 +63,18 @@ require (
 	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/compress v1.15.15 // 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/kr/pretty v0.3.1 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/lib/pq v1.10.7 // indirect
+	github.com/magiconair/properties v1.8.6 // indirect
+	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/mattn/go-runewidth v0.0.9 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // 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/mitchellh/mapstructure v1.5.0 // 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
@@ -75,40 +82,38 @@ require (
 	github.com/multiformats/go-varint v0.0.6 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
-	github.com/pelletier/go-toml v1.2.0 // indirect
+	github.com/pelletier/go-toml v1.9.5 // indirect
+	github.com/pelletier/go-toml/v2 v2.0.5 // 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.10.0 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.37.1 // indirect
+	github.com/prometheus/procfs v0.9.0 // indirect
+	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	github.com/shirou/gopsutil v3.21.11+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.2 // indirect
-	github.com/subosito/gotenv v1.2.0 // indirect
+	github.com/spf13/afero v1.8.2 // indirect
+	github.com/spf13/cast v1.5.0 // indirect
+	github.com/spf13/jwalterweatherman v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/subosito/gotenv v1.3.0 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20220614013038-64ee5596c38a // indirect
-	github.com/tklauser/go-sysconf v0.3.5 // indirect
-	github.com/tklauser/numcpus v0.2.2 // indirect
+	github.com/tklauser/go-sysconf v0.3.11 // indirect
+	github.com/tklauser/numcpus v0.6.0 // indirect
 	github.com/yusufpapurcu/wmi v1.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/sys v0.0.0-20220520151302-bc2c85ada10a // 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
+	golang.org/x/crypto v0.6.0 // indirect
+	golang.org/x/exp v0.0.0-20230206171751-46f607a40771 // indirect
+	golang.org/x/sys v0.5.0 // indirect
+	golang.org/x/text v0.7.0 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+	gopkg.in/ini.v1 v1.67.0 // indirect
+	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	lukechampine.com/blake3 v1.1.7 // indirect
 )

-replace github.com/ethereum/go-ethereum v1.10.21 => github.com/vulcanize/go-ethereum v1.10.21-statediff-4.1.2-alpha
+replace github.com/ethereum/go-ethereum v1.11.5 => github.com/cerc-io/go-ethereum v1.11.5-statediff-4.3.9-alpha
@@ -18,7 +18,7 @@ package main
 import (
 	"github.com/sirupsen/logrus"

-	"github.com/vulcanize/ipld-eth-state-snapshot/cmd"
+	"github.com/cerc-io/ipld-eth-state-snapshot/cmd"
 )

 func main() {
@@ -19,12 +19,12 @@ package prom
 import (
 	"github.com/prometheus/client_golang/prometheus"

-	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
+	mets "github.com/ethereum/go-ethereum/statediff/indexer/database/metrics"
 )

 // DBStatsGetter is an interface that gets sql.DBStats.
 type DBStatsGetter interface {
-	Stats() sql.Stats
+	Stats() mets.DbStats
 }

 // DBStatsCollector implements the prometheus.Collector interface.
@@ -34,6 +34,8 @@ var (
 	stateNodeCount   prometheus.Counter
 	storageNodeCount prometheus.Counter
 	codeNodeCount    prometheus.Counter
+
+	activeIteratorCount prometheus.Gauge
 )

 func Init() {
@@ -59,6 +61,13 @@ func Init() {
 		Name:      "code_node_count",
 		Help:      "Number of code nodes processed",
 	})
+
+	activeIteratorCount = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		Name:      "active_iterator_count",
+		Help:      "Number of active iterators",
+	})
 }

 // RegisterDBCollector create metric collector for given connection
@@ -88,3 +97,17 @@ func IncCodeNodeCount() {
 		codeNodeCount.Inc()
 	}
 }
+
+// IncActiveIterCount increments the number of active iterators
+func IncActiveIterCount() {
+	if metrics {
+		activeIteratorCount.Inc()
+	}
+}
+
+// DecActiveIterCount decrements the number of active iterators
+func DecActiveIterCount() {
+	if metrics {
+		activeIteratorCount.Dec()
+	}
+}
@@ -19,6 +19,8 @@ import (
 	"fmt"
 	"time"

+	"github.com/ethereum/go-ethereum/common"
+
 	"github.com/sirupsen/logrus"

 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
@@ -38,9 +40,10 @@ const (

 // Config contains params for both databases the service uses
 type Config struct {
-	Eth  *EthConfig
-	DB   *DBConfig
-	File *FileConfig
+	Eth     *EthConfig
+	DB      *DBConfig
+	File    *FileConfig
+	Service *ServiceConfig
 }

 // EthConfig is config parameters for the chain.
@@ -60,11 +63,16 @@ type FileConfig struct {
 	OutputDir string
 }

+type ServiceConfig struct {
+	AllowedAccounts map[common.Address]struct{}
+}
+
 func NewConfig(mode SnapshotMode) (*Config, error) {
 	ret := &Config{
 		&EthConfig{},
 		&DBConfig{},
 		&FileConfig{},
+		&ServiceConfig{},
 	}
 	return ret, ret.Init(mode)
 }
@@ -74,6 +82,7 @@ func NewInPlaceSnapshotConfig() *Config {
 		&EthConfig{},
 		&DBConfig{},
 		&FileConfig{},
+		&ServiceConfig{},
 	}
 	ret.DB.Init()

@@ -110,7 +119,7 @@ func (c *Config) Init(mode SnapshotMode) error {
 	default:
 		return fmt.Errorf("no output mode specified")
 	}
-	return nil
+	return c.Service.Init()
 }

 func (c *DBConfig) Init() {
@@ -148,3 +157,19 @@ func (c *FileConfig) Init() error {
 	}
 	return nil
 }
+
+func (c *ServiceConfig) Init() error {
+	viper.BindEnv(SNAPSHOT_ACCOUNTS_TOML, SNAPSHOT_ACCOUNTS)
+	var allowedAccounts []string
+	viper.UnmarshalKey(SNAPSHOT_ACCOUNTS_TOML, &allowedAccounts)
+	accountsLen := len(allowedAccounts)
+	if accountsLen != 0 {
+		c.AllowedAccounts = make(map[common.Address]struct{}, accountsLen)
+		for _, allowedAccount := range allowedAccounts {
+			c.AllowedAccounts[common.HexToAddress(allowedAccount)] = struct{}{}
+		}
+	} else {
+		logrus.Infof("no snapshot addresses specified, will perform snapshot of entire trie(s)")
+	}
+	return nil
+}
@@ -23,6 +23,7 @@ const (
 	SNAPSHOT_MODE          = "SNAPSHOT_MODE"
 	SNAPSHOT_START_HEIGHT  = "SNAPSHOT_START_HEIGHT"
 	SNAPSHOT_END_HEIGHT    = "SNAPSHOT_END_HEIGHT"
+	SNAPSHOT_ACCOUNTS      = "SNAPSHOT_ACCOUNTS"

 	LOGRUS_LEVEL = "LOGRUS_LEVEL"
 	LOGRUS_FILE  = "LOGRUS_FILE"
@@ -62,6 +63,7 @@ const (
 	SNAPSHOT_MODE_TOML          = "snapshot.mode"
 	SNAPSHOT_START_HEIGHT_TOML  = "snapshot.startHeight"
 	SNAPSHOT_END_HEIGHT_TOML    = "snapshot.endHeight"
+	SNAPSHOT_ACCOUNTS_TOML      = "snapshot.accounts"

 	LOGRUS_LEVEL_TOML = "log.level"
 	LOGRUS_FILE_TOML  = "log.file"
@@ -101,6 +103,7 @@ const (
 	SNAPSHOT_MODE_CLI          = "snapshot-mode"
 	SNAPSHOT_START_HEIGHT_CLI  = "start-height"
 	SNAPSHOT_END_HEIGHT_CLI    = "end-height"
+	SNAPSHOT_ACCOUNTS_CLI      = "snapshot-accounts"

 	LOGRUS_LEVEL_CLI = "log-level"
 	LOGRUS_FILE_CLI  = "log-file"
@@ -32,11 +32,11 @@ import (
 	"github.com/multiformats/go-multihash"
 	"github.com/sirupsen/logrus"

+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/prom"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
 	"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"
 )

 var _ snapt.Publisher = (*publisher)(nil)
@@ -9,13 +9,13 @@ 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/ethereum/go-ethereum/statediff/indexer/test_helpers"

-	fixt "github.com/vulcanize/ipld-eth-state-snapshot/fixture"
-	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
-	"github.com/vulcanize/ipld-eth-state-snapshot/test"
+	fixt "github.com/cerc-io/ipld-eth-state-snapshot/fixture"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
+	"github.com/cerc-io/ipld-eth-state-snapshot/test"
 )

 var (
@@ -91,9 +91,9 @@ func TestPgCopy(t *testing.T) {
 	ctx := context.Background()
 	driver, err := postgres.NewSQLXDriver(ctx, pgConfig, nodeInfo)
 	test.NoError(t, err)
-	db := postgres.NewPostgresDB(driver)
+	db := postgres.NewPostgresDB(driver, false)

-	sql.TearDownDB(t, db)
+	test_helpers.TearDownDB(t, db)

 	// copy from files
 	pgCopyStatement := `COPY %s FROM '%s' CSV`
@@ -16,9 +16,9 @@
 package snapshot

 import (
+	. "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
 	"github.com/jmoiron/sqlx"
 	"github.com/sirupsen/logrus"
-	. "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
 )

 const (
@@ -21,16 +21,16 @@ import (
 	"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/ethereum/go-ethereum/statediff/indexer/test_helpers"
 	"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"
+	fixt "github.com/cerc-io/ipld-eth-state-snapshot/fixture"
+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/snapshot/pg"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
+	"github.com/cerc-io/ipld-eth-state-snapshot/test"
 )

 var (
@@ -80,9 +80,9 @@ func TestCreateInPlaceSnapshot(t *testing.T) {
 	ctx := context.Background()
 	driver, err := postgres.NewSQLXDriver(ctx, pgConfig, nodeInfo)
 	test.NoError(t, err)
-	db := postgres.NewPostgresDB(driver)
+	db := postgres.NewPostgresDB(driver, false)

-	sql.TearDownDB(t, db)
+	test_helpers.TearDownDB(t, db)

 	_ = writeData(t, db)

@@ -31,12 +31,12 @@ import (
 	"github.com/sirupsen/logrus"
 	log "github.com/sirupsen/logrus"

+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/prom"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
 	"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"
 )

 var _ snapt.Publisher = (*publisher)(nil)
@@ -4,13 +4,13 @@ import (
 	"context"
 	"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/ethereum/go-ethereum/statediff/indexer/test_helpers"

-	fixt "github.com/vulcanize/ipld-eth-state-snapshot/fixture"
-	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
-	"github.com/vulcanize/ipld-eth-state-snapshot/test"
+	fixt "github.com/cerc-io/ipld-eth-state-snapshot/fixture"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
+	"github.com/cerc-io/ipld-eth-state-snapshot/test"
 )

 var (
@@ -46,9 +46,9 @@ func TestBasic(t *testing.T) {
 	ctx := context.Background()
 	driver, err := postgres.NewSQLXDriver(ctx, pgConfig, nodeInfo)
 	test.NoError(t, err)
-	db := postgres.NewPostgresDB(driver)
+	db := postgres.NewPostgresDB(driver, false)

-	sql.TearDownDB(t, db)
+	test_helpers.TearDownDB(t, db)

 	_ = writeData(t, db)

@@ -17,10 +17,10 @@ package snapshot

 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"math/big"
-	"sync"

 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
@@ -30,11 +30,12 @@ import (
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/trie"
-	"github.com/sirupsen/logrus"
+	iter "github.com/ethereum/go-ethereum/trie/concurrent_iterator"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"

-	iter "github.com/vulcanize/go-eth-state-node-iterator"
-	. "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/prom"
+	. "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
 )

 var (
@@ -48,18 +49,18 @@ var (
 // Service holds ethDB and stateDB to read data from lvldb and Publisher
 // to publish trie in postgres DB.
 type Service struct {
-	ethDB         ethdb.Database
-	stateDB       state.Database
-	ipfsPublisher Publisher
-	maxBatchSize  uint
-	tracker       iteratorTracker
-	recoveryFile  string
+	watchingAddresses bool
+	ethDB             ethdb.Database
+	stateDB           state.Database
+	ipfsPublisher     Publisher
+	maxBatchSize      uint
+	tracker           iteratorTracker
+	recoveryFile      string
 }

 func NewLevelDB(con *EthConfig) (ethdb.Database, error) {
-	edb, err := rawdb.NewLevelDBDatabaseWithFreezer(
-		con.LevelDBPath, 1024, 256, con.AncientDBPath, "ipld-eth-state-snapshot", true,
-	)
+	kvdb, _ := rawdb.NewLevelDBDatabase(con.LevelDBPath, 1024, 256, "ipld-eth-state-snapshot", true)
+	edb, err := rawdb.NewDatabaseWithFreezer(kvdb, con.AncientDBPath, "ipld-eth-state-snapshot", true)
 	if err != nil {
 		return nil, fmt.Errorf("unable to create NewLevelDBDatabaseWithFreezer: %s", err)
 	}
@@ -78,11 +79,17 @@ func NewSnapshotService(edb ethdb.Database, pub Publisher, recoveryFile string)
 }

 type SnapshotParams struct {
-	Height  uint64
-	Workers uint
+	WatchedAddresses map[common.Address]struct{}
+	Height           uint64
+	Workers          uint
 }

 func (s *Service) CreateSnapshot(params SnapshotParams) error {
+	paths := make([][]byte, 0, len(params.WatchedAddresses))
+	for addr := range params.WatchedAddresses {
+		paths = append(paths, keybytesToHex(crypto.Keccak256(addr.Bytes())))
+	}
+	s.watchingAddresses = len(paths) > 0
 	// extract header from lvldb and publish to PG-IPFS
 	// hold onto the headerID so that we can link the state nodes to this header
 	log.Infof("Creating snapshot at height %d", params.Height)
@@ -105,8 +112,10 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 	}

 	headerID := header.Hash().String()
+
+	ctx, cancelCtx := context.WithCancel(context.Background())
 	s.tracker = newTracker(s.recoveryFile, int(params.Workers))
-	s.tracker.captureSignal()
+	s.tracker.captureSignal(cancelCtx)

 	var iters []trie.NodeIterator
 	// attempt to restore from recovery file if it exists
@@ -124,7 +133,8 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 				len(iters), params.Workers,
 			)
 		}
-	} else { // nothing to restore
+	} else {
+		// nothing to restore
 		log.Debugf("no iterators to restore")
 		if params.Workers > 1 {
 			iters = iter.SubtrieIterators(tree, params.Workers)
@@ -132,7 +142,8 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 			iters = []trie.NodeIterator{tree.NodeIterator(nil)}
 		}
 		for i, it := range iters {
-			iters[i] = s.tracker.tracked(it)
+			// recovered path is nil for fresh iterators
+			iters[i] = s.tracker.tracked(it, nil)
 		}
 	}

@@ -143,22 +154,25 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 		}
 	}()

-	if len(iters) > 0 {
-		return s.createSnapshotAsync(iters, headerID, new(big.Int).SetUint64(params.Height))
-	} else {
-		return s.createSnapshot(iters[0], headerID, new(big.Int).SetUint64(params.Height))
+	switch {
+	case len(iters) > 1:
+		return s.createSnapshotAsync(ctx, iters, headerID, new(big.Int).SetUint64(params.Height), paths)
+	case len(iters) == 1:
+		return s.createSnapshot(ctx, iters[0], headerID, new(big.Int).SetUint64(params.Height), paths)
+	default:
+		return nil
 	}
 }

 // Create snapshot up to head (ignores height param)
-func (s *Service) CreateLatestSnapshot(workers uint) error {
+func (s *Service) CreateLatestSnapshot(workers uint, watchedAddresses map[common.Address]struct{}) error {
 	log.Info("Creating snapshot at head")
 	hash := rawdb.ReadHeadHeaderHash(s.ethDB)
 	height := rawdb.ReadHeaderNumber(s.ethDB, hash)
 	if height == nil {
 		return fmt.Errorf("unable to read header height for header hash %s", hash.String())
 	}
-	return s.CreateSnapshot(SnapshotParams{Height: *height, Workers: workers})
+	return s.CreateSnapshot(SnapshotParams{Height: *height, Workers: workers, WatchedAddresses: watchedAddresses})
 }

 type nodeResult struct {
@@ -166,7 +180,7 @@ type nodeResult struct {
 	elements []interface{}
 }

-func resolveNode(it trie.NodeIterator, trieDB *trie.Database) (*nodeResult, error) {
+func resolveNode(nodePath []byte, it trie.NodeIterator, trieDB *trie.Database) (*nodeResult, error) {
 	// "leaf" nodes are actually "value" nodes, whose parents are the actual leaves
 	if it.Leaf() {
 		return nil, nil
@@ -175,8 +189,10 @@ func resolveNode(it trie.NodeIterator, trieDB *trie.Database) (*nodeResult, erro
 		return nil, nil
 	}

-	path := make([]byte, len(it.Path()))
-	copy(path, it.Path())
+	// use full node path
+	// (it.Path() will give partial path in case of subtrie iterators)
+	path := make([]byte, len(nodePath))
+	copy(path, nodePath)
 	n, err := trieDB.Node(it.Hash())
 	if err != nil {
 		return nil, err
@@ -199,7 +215,19 @@ func resolveNode(it trie.NodeIterator, trieDB *trie.Database) (*nodeResult, erro
 	}, nil
 }

-func (s *Service) createSnapshot(it trie.NodeIterator, headerID string, height *big.Int) error {
+// validPath checks if a path is prefix to any one of the paths in the given list
+func validPath(currentPath []byte, seekingPaths [][]byte) bool {
+	for _, seekingPath := range seekingPaths {
+		if bytes.HasPrefix(seekingPath, currentPath) {
+			return true
+		}
+	}
+	return false
+}
+
+// createSnapshot performs traversal using the given iterator and indexes the nodes
+// optionally filtering them according to a list of paths
+func (s *Service) createSnapshot(ctx context.Context, it trie.NodeIterator, headerID string, height *big.Int, seekingPaths [][]byte) error {
 	tx, err := s.ipfsPublisher.BeginTx()
 	if err != nil {
 		return err
@@ -207,99 +235,234 @@ func (s *Service) createSnapshot(it trie.NodeIterator, headerID string, height *
 	defer func() {
 		err = CommitOrRollback(tx, err)
 		if err != nil {
-			logrus.Errorf("CommitOrRollback failed: %s", err)
+			log.Errorf("CommitOrRollback failed: %s", err)
 		}
 	}()

-	for it.Next(true) {
-		res, err := resolveNode(it, s.stateDB.TrieDB())
-		if err != nil {
-			return err
-		}
-		if res == nil {
-			continue
-		}
+	// path (from recovery dump) to be seeked on recovery
+	// nil in case of a fresh iterator
+	var recoveredPath []byte

-		tx, err = s.ipfsPublisher.PrepareTxForBatch(tx, s.maxBatchSize)
-		if err != nil {
-			return err
-		}
+	// latest path seeked from the concurrent iterator
+	// (updated after a node processed)
+	// nil in case of a fresh iterator; initially holds the recovered path in case of a recovered iterator
+	var seekedPath *[]byte

-		switch res.node.NodeType {
-		case Leaf:
-			// if the node is a leaf, decode the account and publish the associated storage trie
-			// nodes if there are any
-			var account types.StateAccount
-			if err := rlp.DecodeBytes(res.elements[1].([]byte), &account); err != nil {
-				return fmt.Errorf(
-					"error decoding account for leaf node at path %x nerror: %v", res.node.Path, err)
+	// end path for the concurrent iterator
+	var endPath []byte
+
+	if iter, ok := it.(*trackedIter); ok {
+		seekedPath = &iter.seekedPath
+		recoveredPath = append(recoveredPath, *seekedPath...)
+		endPath = iter.endPath
+	} else {
+		return errors.New("untracked iterator")
+	}
+
+	return s.createSubTrieSnapshot(ctx, tx, nil, it, recoveredPath, seekedPath, endPath, headerID, height, seekingPaths)
+}
+
+// createSubTrieSnapshot processes nodes at the next level of a trie using the given subtrie iterator
+// continually updating seekedPath with path of the latest processed node
+func (s *Service) createSubTrieSnapshot(ctx context.Context, tx Tx, prefixPath []byte, subTrieIt trie.NodeIterator, recoveredPath []byte, seekedPath *[]byte, endPath []byte, headerID string, height *big.Int, seekingPaths [][]byte) error {
+	prom.IncActiveIterCount()
+	defer prom.DecActiveIterCount()
+
+	// descend in the first loop iteration to reach first child node
+	descend := true
+	for {
+		select {
+		case <-ctx.Done():
+			return errors.New("ctx cancelled")
+		default:
+			if ok := subTrieIt.Next(descend); !ok {
+				return subTrieIt.Error()
 			}
-			partialPath := trie.CompactToHex(res.elements[0].([]byte))
-			valueNodePath := append(res.node.Path, partialPath...)
-			encodedPath := trie.HexToCompact(valueNodePath)
-			leafKey := encodedPath[1:]
-			res.node.Key = common.BytesToHash(leafKey)
-			err := s.ipfsPublisher.PublishStateNode(&res.node, headerID, height, tx)
+
+			// to avoid descending further
+			descend = false
+
+			// move on to next node if current path is empty
+			// occurs when reaching root node or just before reaching the first child of a subtrie in case of some concurrent iterators
+			if bytes.Equal(subTrieIt.Path(), []byte{}) {
+				// if node path is empty and prefix is nil, it's the root node
+				if prefixPath == nil {
+					// create snapshot of node, if it is a leaf this will also create snapshot of entire storage trie
+					if err := s.createNodeSnapshot(tx, subTrieIt.Path(), subTrieIt, headerID, height); err != nil {
+						return err
+					}
+					updateSeekedPath(seekedPath, subTrieIt.Path())
+				}
+
+				if ok := subTrieIt.Next(true); !ok {
+					// return if no further nodes available
+					return subTrieIt.Error()
+				}
+			}
+
+			// create the full node path as it.Path() doesn't include the path before subtrie root
+			nodePath := append(prefixPath, subTrieIt.Path()...)
+
+			// check iterator upper bound before processing the node
+			// required to avoid processing duplicate nodes:
+			//   if a node is considered more than once,
+			//   it's whole subtrie is re-processed giving large number of duplicate nodoes
+			if !checkUpperPathBound(nodePath, endPath) {
+				// fmt.Println("failed checkUpperPathBound", nodePath, endPath)
+				// explicitly stop the iterator in tracker if upper bound check fails
+				// required since it won't be marked as stopped if further nodes are still available
+				if trackedSubtrieIt, ok := subTrieIt.(*trackedIter); ok {
+					s.tracker.stopIter(trackedSubtrieIt)
+				}
+				return subTrieIt.Error()
+			}
+
+			// skip the current node if it's before recovered path and not along the recovered path
+			// nodes at the same level that are before recovered path are ignored to avoid duplicate nodes
+			// however, nodes along the recovered path are re-considered for redundancy
+			if bytes.Compare(recoveredPath, nodePath) > 0 &&
+				// a node is along the recovered path if it's path is shorter or equal in length
+				// and is part of the recovered path
+				!(len(nodePath) <= len(recoveredPath) && bytes.Equal(recoveredPath[:len(nodePath)], nodePath)) {
+				continue
+			}
+
+			// ignore node if it is not along paths of interest
+			if s.watchingAddresses && !validPath(nodePath, seekingPaths) {
+				// consider this node as processed since it is getting ignored
+				// and update the seeked path
+				updateSeekedPath(seekedPath, nodePath)
+				// move on to the next node
+				continue
+			}
+
+			// if the node is along paths of interest
+			// create snapshot of node, if it is a leaf this will also create snapshot of entire storage trie
+			if err := s.createNodeSnapshot(tx, nodePath, subTrieIt, headerID, height); err != nil {
+				return err
+			}
+			// update seeked path after node has been processed
+			updateSeekedPath(seekedPath, nodePath)
+
+			// create an iterator to traverse and process the next level of this subTrie
+			nextSubTrieIt, err := s.createSubTrieIt(nodePath, subTrieIt.Hash(), recoveredPath)
 			if err != nil {
 				return err
 			}
-
-			// publish any non-nil code referenced by codehash
-			if !bytes.Equal(account.CodeHash, emptyCodeHash) {
-				codeHash := common.BytesToHash(account.CodeHash)
-				codeBytes := rawdb.ReadCode(s.ethDB, codeHash)
-				if len(codeBytes) == 0 {
-					log.Error("Code is missing", "account", common.BytesToHash(it.LeafKey()))
-					return errors.New("missing code")
-				}
-
-				if err = s.ipfsPublisher.PublishCode(height, codeHash, codeBytes, tx); err != nil {
-					return err
-				}
-			}
-
-			if tx, err = s.storageSnapshot(account.Root, headerID, height, res.node.Path, tx); err != nil {
-				return fmt.Errorf("failed building storage snapshot for account %+v\r\nerror: %w", account, err)
-			}
-		case Extension, Branch:
-			res.node.Key = common.BytesToHash([]byte{})
-			if err := s.ipfsPublisher.PublishStateNode(&res.node, headerID, height, tx); err != nil {
+			// pass on the seekedPath of the tracked concurrent iterator to be updated
+			if err := s.createSubTrieSnapshot(ctx, tx, nodePath, nextSubTrieIt, recoveredPath, seekedPath, endPath, headerID, height, seekingPaths); err != nil {
 				return err
 			}
-		default:
-			return errors.New("unexpected node type")
 		}
 	}
+}
+
+// createSubTrieIt creates an iterator to traverse the subtrie of node with the given hash
+// the subtrie iterator is initialized at a node from the recovered path at corresponding level (if avaiable)
+func (s *Service) createSubTrieIt(prefixPath []byte, hash common.Hash, recoveredPath []byte) (trie.NodeIterator, error) {
+	// skip directly to the node from the recovered path at corresponding level
+	// applicable if:
+	//   node path is behind recovered path
+	//   and recovered path includes the prefix path
+	var startPath []byte
+	if bytes.Compare(recoveredPath, prefixPath) > 0 &&
+		len(recoveredPath) > len(prefixPath) &&
+		bytes.Equal(recoveredPath[:len(prefixPath)], prefixPath) {
+		startPath = append(startPath, recoveredPath[len(prefixPath):len(prefixPath)+1]...)
+		// force the lower bound path to an even length
+		// (required by HexToKeyBytes())
+		if len(startPath)&0b1 == 1 {
+			// decrement first to avoid skipped nodes
+			decrementPath(startPath)
+			startPath = append(startPath, 0)
+		}
+	}
+
+	// create subTrie iterator with the given hash
+	subTrie, err := s.stateDB.OpenTrie(hash)
+	if err != nil {
+		return nil, err
+	}
+
+	return subTrie.NodeIterator(iter.HexToKeyBytes(startPath)), nil
+}
+
+// createNodeSnapshot indexes the current node
+// entire storage trie is also indexed (if available)
+func (s *Service) createNodeSnapshot(tx Tx, path []byte, it trie.NodeIterator, headerID string, height *big.Int) error {
+	res, err := resolveNode(path, it, s.stateDB.TrieDB())
+	if err != nil {
+		return err
+	}
+	if res == nil {
+		return nil
+	}
+
+	tx, err = s.ipfsPublisher.PrepareTxForBatch(tx, s.maxBatchSize)
+	if err != nil {
+		return err
+	}
+
+	switch res.node.NodeType {
+	case Leaf:
+		// if the node is a leaf, decode the account and publish the associated storage trie
+		// nodes if there are any
+		var account types.StateAccount
+		if err := rlp.DecodeBytes(res.elements[1].([]byte), &account); err != nil {
+			return fmt.Errorf(
+				"error decoding account for leaf node at path %x nerror: %v", res.node.Path, err)
+		}
+		partialPath := trie.CompactToHex(res.elements[0].([]byte))
+		valueNodePath := append(res.node.Path, partialPath...)
+		encodedPath := trie.HexToCompact(valueNodePath)
+		leafKey := encodedPath[1:]
+		res.node.Key = common.BytesToHash(leafKey)
+		if err := s.ipfsPublisher.PublishStateNode(&res.node, headerID, height, tx); err != nil {
+			return err
+		}
+
+		// publish any non-nil code referenced by codehash
+		if !bytes.Equal(account.CodeHash, emptyCodeHash) {
+			codeHash := common.BytesToHash(account.CodeHash)
+			codeBytes := rawdb.ReadCode(s.ethDB, codeHash)
+			if len(codeBytes) == 0 {
+				log.Error("Code is missing", "account", common.BytesToHash(it.LeafKey()))
+				return errors.New("missing code")
+			}
+
+			if err = s.ipfsPublisher.PublishCode(height, codeHash, codeBytes, tx); err != nil {
+				return err
+			}
+		}
+
+		if _, err = s.storageSnapshot(account.Root, headerID, height, res.node.Path, tx); err != nil {
+			return fmt.Errorf("failed building storage snapshot for account %+v\r\nerror: %w", account, err)
+		}
+	case Extension, Branch:
+		res.node.Key = common.BytesToHash([]byte{})
+		if err := s.ipfsPublisher.PublishStateNode(&res.node, headerID, height, tx); err != nil {
+			return err
+		}
+	default:
+		return errors.New("unexpected node type")
+	}
 	return it.Error()
 }

 // Full-trie concurrent snapshot
-func (s *Service) createSnapshotAsync(iters []trie.NodeIterator, headerID string, height *big.Int) error {
-	errors := make(chan error)
-	var wg sync.WaitGroup
+func (s *Service) createSnapshotAsync(ctx context.Context, iters []trie.NodeIterator, headerID string, height *big.Int, seekingPaths [][]byte) error {
+	// use errgroup with a context to stop all concurrent iterators if one runs into an error
+	// each concurrent iterator completes processing it's current node before stopping
+	g, ctx := errgroup.WithContext(ctx)
 	for _, it := range iters {
-		wg.Add(1)
-		go func(it trie.NodeIterator) {
-			defer wg.Done()
-			if err := s.createSnapshot(it, headerID, height); err != nil {
-				errors <- err
-			}
+		func(it trie.NodeIterator) {
+			g.Go(func() error {
+				return s.createSnapshot(ctx, it, headerID, height, seekingPaths)
+			})
 		}(it)
 	}

-	done := make(chan struct{})
-	go func() {
-		wg.Wait()
-		done <- struct{}{}
-	}()
-
-	var err error
-	select {
-	case err = <-errors:
-	case <-done:
-		close(errors)
-	}
-	return err
+	return g.Wait()
 }

 func (s *Service) storageSnapshot(sr common.Hash, headerID string, height *big.Int, statePath []byte, tx Tx) (Tx, error) {
@@ -314,7 +477,7 @@ func (s *Service) storageSnapshot(sr common.Hash, headerID string, height *big.I

 	it := sTrie.NodeIterator(make([]byte, 0))
 	for it.Next(true) {
-		res, err := resolveNode(it, s.stateDB.TrieDB())
+		res, err := resolveNode(it.Path(), it, s.stateDB.TrieDB())
 		if err != nil {
 			return nil, err
 		}
@@ -2,18 +2,33 @@ package snapshot

 import (
 	"errors"
+	"fmt"
 	"math/big"
+	"math/rand"
 	"os"
 	"path/filepath"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"

+	"github.com/ethereum/go-ethereum/common"
 	"github.com/golang/mock/gomock"

-	fixt "github.com/vulcanize/ipld-eth-state-snapshot/fixture"
-	mock "github.com/vulcanize/ipld-eth-state-snapshot/mocks/snapshot"
-	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
-	"github.com/vulcanize/ipld-eth-state-snapshot/test"
+	fixt "github.com/cerc-io/ipld-eth-state-snapshot/fixture"
+	mock "github.com/cerc-io/ipld-eth-state-snapshot/mocks/snapshot"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
+	"github.com/cerc-io/ipld-eth-state-snapshot/test"
+)
+
+var (
+	stateNodeNotIndexedErr   = "state node not indexed for path %v"
+	storageNodeNotIndexedErr = "storage node not indexed for state path %v, storage path %v"
+
+	unexpectedStateNodeErr   = "got unexpected state node for path %v"
+	unexpectedStorageNodeErr = "got unexpected storage node for state path %v, storage path %v"
+
+	extraNodesIndexedErr = "number of nodes indexed (%v) is more than expected (max %v)"
 )

 func testConfig(leveldbpath, ancientdbpath string) *Config {
@@ -39,23 +54,40 @@ func makeMocks(t *testing.T) (*mock.MockPublisher, *mock.MockTx) {

 func TestCreateSnapshot(t *testing.T) {
 	runCase := func(t *testing.T, workers int) {
+		// map: expected state path -> struct{}{}
+		expectedStateNodePaths := sync.Map{}
+		for _, path := range fixt.Block1_StateNodePaths {
+			expectedStateNodePaths.Store(string(path), struct{}{})
+		}
+
 		pub, tx := makeMocks(t)
 		pub.EXPECT().PublishHeader(gomock.Eq(&fixt.Block1_Header))
 		pub.EXPECT().BeginTx().Return(tx, nil).
 			Times(workers)
 		pub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).
 			AnyTimes()
-		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-			// Use MinTimes as duplicate nodes are expected at boundaries
-			MinTimes(len(fixt.Block1_StateNodePaths))
+		tx.EXPECT().Commit().
+			Times(workers)
+		pub.EXPECT().PublishStateNode(
+			gomock.Any(),
+			gomock.Eq(fixt.Block1_Header.Hash().String()),
+			gomock.Eq(fixt.Block1_Header.Number),
+			gomock.Eq(tx)).
+			DoAndReturn(func(node *snapt.Node, _ string, _ *big.Int, _ snapt.Tx) error {
+				if _, ok := expectedStateNodePaths.Load(string(node.Path)); ok {
+					expectedStateNodePaths.Delete(string(node.Path))
+				} else {
+					t.Fatalf(unexpectedStateNodeErr, node.Path)
+				}
+				return nil
+			}).
+			Times(len(fixt.Block1_StateNodePaths))

 		// TODO: fixtures for storage node
 		// pub.EXPECT().PublishStorageNode(gomock.Eq(fixt.StorageNode), gomock.Eq(int64(0)), gomock.Any())

-		tx.EXPECT().Commit().
-			Times(workers)
-
-		config := testConfig(fixt.ChaindataPath, fixt.AncientdataPath)
+		chainDataPath, ancientDataPath := fixt.GetChainDataPath("chaindata")
+		config := testConfig(chainDataPath, ancientDataPath)
 		edb, err := NewLevelDB(config.Eth)
 		if err != nil {
 			t.Fatal(err)
@@ -73,30 +105,221 @@ func TestCreateSnapshot(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
+
+		// Check if all expected state nodes are indexed
+		expectedStateNodePaths.Range(func(key, value any) bool {
+			t.Fatalf(stateNodeNotIndexedErr, []byte(key.(string)))
+			return true
+		})
 	}

-	testCases := []int{1, 4, 16, 32}
+	testCases := []int{1, 4, 8, 16, 32}
 	for _, tc := range testCases {
 		t.Run("case", func(t *testing.T) { runCase(t, tc) })
 	}
 }

-func failingPublishStateNode(_ *snapt.Node, _ string, _ *big.Int, _ snapt.Tx) error {
-	return errors.New("failingPublishStateNode")
+type indexedNode struct {
+	value     snapt.Node
+	isIndexed bool
+}
+
+type storageNodeKey struct {
+	statePath   string
+	storagePath string
+}
+
+func TestAccountSelectiveSnapshot(t *testing.T) {
+	snapShotHeight := uint64(32)
+	watchedAddresses := map[common.Address]struct{}{
+		common.HexToAddress("0x825a6eec09e44Cb0fa19b84353ad0f7858d7F61a"): {},
+		common.HexToAddress("0x0616F59D291a898e796a1FAD044C5926ed2103eC"): {},
+	}
+
+	expectedStateNodeIndexes := []int{0, 1, 2, 6}
+
+	statePath33 := []byte{3, 3}
+	expectedStorageNodeIndexes33 := []int{0, 1, 2, 3, 4, 6, 8}
+
+	statePath12 := []byte{12}
+	expectedStorageNodeIndexes12 := []int{12, 14, 16}
+
+	runCase := func(t *testing.T, workers int) {
+		expectedStateNodes := sync.Map{}
+
+		for _, expectedStateNodeIndex := range expectedStateNodeIndexes {
+			path := fixt.Chain2_Block32_StateNodes[expectedStateNodeIndex].Path
+			expectedStateNodes.Store(string(path), indexedNode{
+				value:     fixt.Chain2_Block32_StateNodes[expectedStateNodeIndex],
+				isIndexed: false,
+			})
+		}
+
+		expectedStorageNodes := sync.Map{}
+
+		for _, expectedStorageNodeIndex := range expectedStorageNodeIndexes33 {
+			path := fixt.Chain2_Block32_StorageNodes[expectedStorageNodeIndex].Path
+			key := storageNodeKey{
+				statePath:   string(statePath33),
+				storagePath: string(path),
+			}
+			value := indexedNode{
+				value:     fixt.Chain2_Block32_StorageNodes[expectedStorageNodeIndex].Node,
+				isIndexed: false,
+			}
+			expectedStorageNodes.Store(key, value)
+		}
+
+		for _, expectedStorageNodeIndex := range expectedStorageNodeIndexes12 {
+			path := fixt.Chain2_Block32_StorageNodes[expectedStorageNodeIndex].Path
+			key := storageNodeKey{
+				statePath:   string(statePath12),
+				storagePath: string(path),
+			}
+			value := indexedNode{
+				value:     fixt.Chain2_Block32_StorageNodes[expectedStorageNodeIndex].Node,
+				isIndexed: false,
+			}
+			expectedStorageNodes.Store(key, value)
+		}
+
+		pub, tx := makeMocks(t)
+		pub.EXPECT().PublishHeader(gomock.Eq(&fixt.Chain2_Block32_Header))
+		pub.EXPECT().BeginTx().Return(tx, nil).
+			Times(workers)
+		pub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).
+			AnyTimes()
+		tx.EXPECT().Commit().
+			Times(workers)
+		pub.EXPECT().PublishCode(gomock.Eq(fixt.Chain2_Block32_Header.Number), gomock.Any(), gomock.Any(), gomock.Eq(tx)).
+			AnyTimes()
+		pub.EXPECT().PublishStateNode(
+			gomock.Any(),
+			gomock.Eq(fixt.Chain2_Block32_Header.Hash().String()),
+			gomock.Eq(fixt.Chain2_Block32_Header.Number),
+			gomock.Eq(tx)).
+			Do(func(node *snapt.Node, _ string, _ *big.Int, _ snapt.Tx) error {
+				key := string(node.Path)
+				// Check published nodes
+				if expectedStateNode, ok := expectedStateNodes.Load(key); ok {
+					expectedVal := expectedStateNode.(indexedNode).value
+					test.ExpectEqual(t, expectedVal, *node)
+
+					// Mark expected node as indexed
+					expectedStateNodes.Store(key, indexedNode{
+						value:     expectedVal,
+						isIndexed: true,
+					})
+				} else {
+					t.Fatalf(unexpectedStateNodeErr, node.Path)
+				}
+				return nil
+			}).
+			AnyTimes()
+		pub.EXPECT().PublishStorageNode(
+			gomock.Any(),
+			gomock.Eq(fixt.Chain2_Block32_Header.Hash().String()),
+			gomock.Eq(new(big.Int).SetUint64(snapShotHeight)),
+			gomock.Any(),
+			gomock.Eq(tx)).
+			Do(func(node *snapt.Node, _ string, _ *big.Int, statePath []byte, _ snapt.Tx) error {
+				key := storageNodeKey{
+					statePath:   string(statePath),
+					storagePath: string(node.Path),
+				}
+				// Check published nodes
+				if expectedStorageNode, ok := expectedStorageNodes.Load(key); ok {
+					expectedVal := expectedStorageNode.(indexedNode).value
+					test.ExpectEqual(t, expectedVal, *node)
+
+					// Mark expected node as indexed
+					expectedStorageNodes.Store(key, indexedNode{
+						value:     expectedVal,
+						isIndexed: true,
+					})
+				} else {
+					t.Fatalf(unexpectedStorageNodeErr, statePath, node.Path)
+				}
+				return nil
+			}).
+			AnyTimes()
+
+		chainDataPath, ancientDataPath := fixt.GetChainDataPath("chain2data")
+		config := testConfig(chainDataPath, ancientDataPath)
+		edb, err := NewLevelDB(config.Eth)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer edb.Close()
+
+		recovery := filepath.Join(t.TempDir(), "recover.csv")
+		service, err := NewSnapshotService(edb, pub, recovery)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		params := SnapshotParams{Height: snapShotHeight, Workers: uint(workers), WatchedAddresses: watchedAddresses}
+		err = service.CreateSnapshot(params)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		expectedStateNodes.Range(func(key, value any) bool {
+			if !value.(indexedNode).isIndexed {
+				t.Fatalf(stateNodeNotIndexedErr, []byte(key.(string)))
+				return false
+			}
+			return true
+		})
+		expectedStorageNodes.Range(func(key, value any) bool {
+			if !value.(indexedNode).isIndexed {
+				t.Fatalf(storageNodeNotIndexedErr, []byte(key.(storageNodeKey).statePath), []byte(key.(storageNodeKey).storagePath))
+				return false
+			}
+			return true
+		})
+	}
+
+	testCases := []int{1, 4, 8, 16, 32}
+	for _, tc := range testCases {
+		t.Run("case", func(t *testing.T) { runCase(t, tc) })
+	}
 }

 func TestRecovery(t *testing.T) {
-	runCase := func(t *testing.T, workers int) {
-		pub, tx := makeMocks(t)
-		pub.EXPECT().PublishHeader(gomock.Any()).AnyTimes()
-		pub.EXPECT().BeginTx().Return(tx, nil).AnyTimes()
-		pub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).AnyTimes()
-		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-			Times(workers).
-			DoAndReturn(failingPublishStateNode)
-		tx.EXPECT().Commit().AnyTimes()
+	maxPathLength := 4
+	runCase := func(t *testing.T, workers int, interruptAt int32) {
+		// map: expected state path -> number of times it got published
+		expectedStateNodePaths := sync.Map{}
+		for _, path := range fixt.Block1_StateNodePaths {
+			expectedStateNodePaths.Store(string(path), 0)
+		}
+		var indexedStateNodesCount int32

-		config := testConfig(fixt.ChaindataPath, fixt.AncientdataPath)
+		pub, tx := makeMocks(t)
+		pub.EXPECT().PublishHeader(gomock.Eq(&fixt.Block1_Header))
+		pub.EXPECT().BeginTx().Return(tx, nil).MaxTimes(workers)
+		pub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).AnyTimes()
+		tx.EXPECT().Commit().MaxTimes(workers)
+		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+			DoAndReturn(func(node *snapt.Node, _ string, _ *big.Int, _ snapt.Tx) error {
+				// Start throwing an error after a certain number of state nodes have been indexed
+				if indexedStateNodesCount >= interruptAt {
+					return errors.New("failingPublishStateNode")
+				} else {
+					if prevCount, ok := expectedStateNodePaths.Load(string(node.Path)); ok {
+						expectedStateNodePaths.Store(string(node.Path), prevCount.(int)+1)
+						atomic.AddInt32(&indexedStateNodesCount, 1)
+					} else {
+						t.Fatalf(unexpectedStateNodeErr, node.Path)
+					}
+				}
+				return nil
+			}).
+			MaxTimes(int(interruptAt) + workers)
+
+		chainDataPath, ancientDataPath := fixt.GetChainDataPath("chaindata")
+		config := testConfig(chainDataPath, ancientDataPath)
 		edb, err := NewLevelDB(config.Eth)
 		if err != nil {
 			t.Fatal(err)
@@ -119,15 +342,35 @@ func TestRecovery(t *testing.T) {
 			t.Fatal("cannot stat recovery file:", err)
 		}

-		// Wait for earlier snapshot process to complete
-		time.Sleep(5 * time.Second)
+		// Create new mocks for recovery
+		recoveryPub, tx := makeMocks(t)
+		recoveryPub.EXPECT().PublishHeader(gomock.Eq(&fixt.Block1_Header))
+		recoveryPub.EXPECT().BeginTx().Return(tx, nil).AnyTimes()
+		recoveryPub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).AnyTimes()
+		tx.EXPECT().Commit().AnyTimes()
+		recoveryPub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+			DoAndReturn(func(node *snapt.Node, _ string, _ *big.Int, _ snapt.Tx) error {
+				if prevCount, ok := expectedStateNodePaths.Load(string(node.Path)); ok {
+					expectedStateNodePaths.Store(string(node.Path), prevCount.(int)+1)
+					atomic.AddInt32(&indexedStateNodesCount, 1)
+				} else {
+					t.Fatalf(unexpectedStateNodeErr, node.Path)
+				}
+				return nil
+			}).
+			AnyTimes()

-		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
-		err = service.CreateSnapshot(params)
+		// Create a new snapshot service for recovery
+		recoveryService, err := NewSnapshotService(edb, recoveryPub, recovery)
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = recoveryService.CreateSnapshot(params)
 		if err != nil {
 			t.Fatal(err)
 		}

+		// Check if recovery file has been deleted
 		_, err = os.Stat(recovery)
 		if err == nil {
 			t.Fatal("recovery file still present")
@@ -136,11 +379,192 @@ func TestRecovery(t *testing.T) {
 				t.Fatal(err)
 			}
 		}
+
+		// Check if all state nodes are indexed after recovery
+		expectedStateNodePaths.Range(func(key, value any) bool {
+			if value.(int) == 0 {
+				t.Fatalf(stateNodeNotIndexedErr, []byte(key.(string)))
+			}
+			return true
+		})
+
+		// nodes along the recovery path get reindexed
+		maxStateNodesCount := len(fixt.Block1_StateNodePaths) + workers*maxPathLength
+		if indexedStateNodesCount > int32(maxStateNodesCount) {
+			t.Fatalf(extraNodesIndexedErr, indexedStateNodesCount, maxStateNodesCount)
+		}
+	}
+
+	testCases := []int{1, 2, 4, 8, 16, 32}
+	numInterrupts := 3
+	interrupts := make([]int32, numInterrupts)
+	for i := 0; i < numInterrupts; i++ {
+		rand.Seed(time.Now().UnixNano())
+		interrupts[i] = rand.Int31n(int32(len(fixt.Block1_StateNodePaths)))
 	}

-	testCases := []int{1, 4, 32}
 	for _, tc := range testCases {
-		t.Run("case", func(t *testing.T) { runCase(t, tc) })
+		for _, interrupt := range interrupts {
+			t.Run(fmt.Sprint("case", tc, interrupt), func(t *testing.T) { runCase(t, tc, interrupt) })
+		}
+	}
+}
+
+func TestAccountSelectiveRecovery(t *testing.T) {
+	maxPathLength := 2
+	snapShotHeight := uint64(32)
+	watchedAddresses := map[common.Address]struct{}{
+		common.HexToAddress("0x825a6eec09e44Cb0fa19b84353ad0f7858d7F61a"): {},
+		common.HexToAddress("0x0616F59D291a898e796a1FAD044C5926ed2103eC"): {},
 	}

+	expectedStateNodeIndexes := []int{0, 1, 2, 6}
+
+	runCase := func(t *testing.T, workers int, interruptAt int32) {
+		// map: expected state path -> number of times it got published
+		expectedStateNodePaths := sync.Map{}
+		for _, expectedStateNodeIndex := range expectedStateNodeIndexes {
+			path := fixt.Chain2_Block32_StateNodes[expectedStateNodeIndex].Path
+			expectedStateNodePaths.Store(string(path), 0)
+		}
+		var indexedStateNodesCount int32
+
+		pub, tx := makeMocks(t)
+		pub.EXPECT().PublishHeader(gomock.Eq(&fixt.Chain2_Block32_Header))
+		pub.EXPECT().BeginTx().Return(tx, nil).Times(workers)
+		pub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).AnyTimes()
+		tx.EXPECT().Commit().Times(workers)
+		pub.EXPECT().PublishStateNode(
+			gomock.Any(),
+			gomock.Eq(fixt.Chain2_Block32_Header.Hash().String()),
+			gomock.Eq(fixt.Chain2_Block32_Header.Number),
+			gomock.Eq(tx)).
+			DoAndReturn(func(node *snapt.Node, _ string, _ *big.Int, _ snapt.Tx) error {
+				// Start throwing an error after a certain number of state nodes have been indexed
+				if indexedStateNodesCount >= interruptAt {
+					return errors.New("failingPublishStateNode")
+				} else {
+					if prevCount, ok := expectedStateNodePaths.Load(string(node.Path)); ok {
+						expectedStateNodePaths.Store(string(node.Path), prevCount.(int)+1)
+						atomic.AddInt32(&indexedStateNodesCount, 1)
+					} else {
+						t.Fatalf(unexpectedStateNodeErr, node.Path)
+					}
+				}
+				return nil
+			}).
+			MaxTimes(int(interruptAt) + workers)
+		pub.EXPECT().PublishStorageNode(
+			gomock.Any(),
+			gomock.Eq(fixt.Chain2_Block32_Header.Hash().String()),
+			gomock.Eq(new(big.Int).SetUint64(snapShotHeight)),
+			gomock.Any(),
+			gomock.Eq(tx)).
+			AnyTimes()
+		pub.EXPECT().PublishCode(gomock.Eq(fixt.Chain2_Block32_Header.Number), gomock.Any(), gomock.Any(), gomock.Eq(tx)).
+			AnyTimes()
+
+		chainDataPath, ancientDataPath := fixt.GetChainDataPath("chain2data")
+		config := testConfig(chainDataPath, ancientDataPath)
+		edb, err := NewLevelDB(config.Eth)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer edb.Close()
+
+		recovery := filepath.Join(t.TempDir(), "recover.csv")
+		service, err := NewSnapshotService(edb, pub, recovery)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		params := SnapshotParams{Height: snapShotHeight, Workers: uint(workers), WatchedAddresses: watchedAddresses}
+		err = service.CreateSnapshot(params)
+		if err == nil {
+			t.Fatal("expected an error")
+		}
+
+		if _, err = os.Stat(recovery); err != nil {
+			t.Fatal("cannot stat recovery file:", err)
+		}
+
+		// Create new mocks for recovery
+		recoveryPub, tx := makeMocks(t)
+		recoveryPub.EXPECT().PublishHeader(gomock.Eq(&fixt.Chain2_Block32_Header))
+		recoveryPub.EXPECT().BeginTx().Return(tx, nil).MaxTimes(workers)
+		recoveryPub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).AnyTimes()
+		tx.EXPECT().Commit().MaxTimes(workers)
+		recoveryPub.EXPECT().PublishStateNode(
+			gomock.Any(),
+			gomock.Eq(fixt.Chain2_Block32_Header.Hash().String()),
+			gomock.Eq(fixt.Chain2_Block32_Header.Number),
+			gomock.Eq(tx)).
+			DoAndReturn(func(node *snapt.Node, _ string, _ *big.Int, _ snapt.Tx) error {
+				if prevCount, ok := expectedStateNodePaths.Load(string(node.Path)); ok {
+					expectedStateNodePaths.Store(string(node.Path), prevCount.(int)+1)
+					atomic.AddInt32(&indexedStateNodesCount, 1)
+				} else {
+					t.Fatalf(unexpectedStateNodeErr, node.Path)
+				}
+				return nil
+			}).
+			AnyTimes()
+		recoveryPub.EXPECT().PublishStorageNode(
+			gomock.Any(),
+			gomock.Eq(fixt.Chain2_Block32_Header.Hash().String()),
+			gomock.Eq(new(big.Int).SetUint64(snapShotHeight)),
+			gomock.Any(),
+			gomock.Eq(tx)).
+			AnyTimes()
+		recoveryPub.EXPECT().PublishCode(gomock.Eq(fixt.Chain2_Block32_Header.Number), gomock.Any(), gomock.Any(), gomock.Eq(tx)).
+			AnyTimes()
+
+		// Create a new snapshot service for recovery
+		recoveryService, err := NewSnapshotService(edb, recoveryPub, recovery)
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = recoveryService.CreateSnapshot(params)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// Check if recovery file has been deleted
+		_, err = os.Stat(recovery)
+		if err == nil {
+			t.Fatal("recovery file still present")
+		} else {
+			if !os.IsNotExist(err) {
+				t.Fatal(err)
+			}
+		}
+
+		// Check if all expected state nodes are indexed after recovery
+		expectedStateNodePaths.Range(func(key, value any) bool {
+			if value.(int) == 0 {
+				t.Fatalf(stateNodeNotIndexedErr, []byte(key.(string)))
+			}
+			return true
+		})
+
+		// nodes along the recovery path get reindexed
+		maxStateNodesCount := len(expectedStateNodeIndexes) + workers*maxPathLength
+		if indexedStateNodesCount > int32(maxStateNodesCount) {
+			t.Fatalf(extraNodesIndexedErr, indexedStateNodesCount, maxStateNodesCount)
+		}
+	}
+
+	testCases := []int{1, 2, 4, 8, 16, 32}
+	numInterrupts := 2
+	interrupts := make([]int32, numInterrupts)
+	for i := 0; i < numInterrupts; i++ {
+		rand.Seed(time.Now().UnixNano())
+		interrupts[i] = rand.Int31n(int32(len(expectedStateNodeIndexes)))
+	}
+
+	for _, tc := range testCases {
+		for _, interrupt := range interrupts {
+			t.Run(fmt.Sprint("case", tc, interrupt), func(t *testing.T) { runCase(t, tc, interrupt) })
+		}
+	}
 }
@@ -1,6 +1,8 @@
 package snapshot

 import (
+	"bytes"
+	"context"
 	"encoding/csv"
 	"fmt"
 	"os"
@@ -9,18 +11,21 @@ import (

 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/trie"
+	iter "github.com/ethereum/go-ethereum/trie/concurrent_iterator"
 	log "github.com/sirupsen/logrus"
-
-	iter "github.com/vulcanize/go-eth-state-node-iterator"
 )

 type trackedIter struct {
 	trie.NodeIterator
 	tracker *iteratorTracker
+
+	seekedPath []byte // latest path seeked from the tracked iterator
+	endPath    []byte // endPath for the tracked iterator
 }

 func (it *trackedIter) Next(descend bool) bool {
 	ret := it.NodeIterator.Next(descend)
+
 	if !ret {
 		if it.tracker.running {
 			it.tracker.stopChan <- it
@@ -51,45 +56,79 @@ func newTracker(file string, buf int) iteratorTracker {
 	}
 }

-func (tr *iteratorTracker) captureSignal() {
+func (tr *iteratorTracker) captureSignal(cancelCtx context.CancelFunc) {
 	sigChan := make(chan os.Signal, 1)

 	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
 	go func() {
 		sig := <-sigChan
 		log.Errorf("Signal received (%v), stopping", sig)
-		tr.haltAndDump()
-		os.Exit(1)
+		// cancel context on receiving a signal
+		// on ctx cancellation, all the iterators complete processing of their current node before stopping
+		cancelCtx()
 	}()
 }

 // 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}
+func (tr *iteratorTracker) tracked(it trie.NodeIterator, recoveredPath []byte) (ret *trackedIter) {
+	// create seeked path of max capacity (65)
+	iterSeekedPath := make([]byte, 0, 65)
+	// intially populate seeked path with the recovered path
+	// to be used in trie traversal
+	if recoveredPath != nil {
+		iterSeekedPath = append(iterSeekedPath, recoveredPath...)
+	}
+
+	// if the iterator being tracked is a PrefixBoundIterator, capture it's end path
+	// to be used in trie traversal
+	var endPath []byte
+	if boundedIter, ok := it.(*iter.PrefixBoundIterator); ok {
+		endPath = boundedIter.EndPath
+	}
+
+	ret = &trackedIter{it, tr, iterSeekedPath, endPath}
 	tr.startChan <- ret
 	return
 }

+// explicitly stops an iterator
+func (tr *iteratorTracker) stopIter(it *trackedIter) {
+	tr.stopChan <- it
+}
+
 // dumps iterator path and bounds to a text file so it can be restored later
 func (tr *iteratorTracker) dump() error {
 	log.Debug("Dumping recovery state to: ", tr.recoveryFile)
 	var rows [][]string
-	for it, _ := range tr.started {
+	for it := range tr.started {
+		var startPath []byte
 		var endPath []byte
 		if impl, ok := it.NodeIterator.(*iter.PrefixBoundIterator); ok {
+			// if the iterator being tracked is a PrefixBoundIterator,
+			// initialize start and end paths with its bounds
+			startPath = impl.StartPath
 			endPath = impl.EndPath
 		}
+
+		// if seeked path and iterator path are non-empty, use iterator's path as startpath
+		if !bytes.Equal(it.seekedPath, []byte{}) && !bytes.Equal(it.Path(), []byte{}) {
+			startPath = it.Path()
+		}
+
 		rows = append(rows, []string{
-			fmt.Sprintf("%x", it.Path()),
+			fmt.Sprintf("%x", startPath),
 			fmt.Sprintf("%x", endPath),
+			fmt.Sprintf("%x", it.seekedPath),
 		})
 	}
+
 	file, err := os.Create(tr.recoveryFile)
 	if err != nil {
 		return err
 	}
 	defer file.Close()
 	out := csv.NewWriter(file)
+
 	return out.WriteAll(rows)
 }

@@ -106,30 +145,45 @@ func (tr *iteratorTracker) restore(tree state.Trie) ([]trie.NodeIterator, error)
 	log.Debug("Restoring recovery state from: ", tr.recoveryFile)
 	defer file.Close()
 	in := csv.NewReader(file)
-	in.FieldsPerRecord = 2
+	in.FieldsPerRecord = 3
 	rows, err := in.ReadAll()
 	if err != nil {
 		return nil, err
 	}
+
 	var ret []trie.NodeIterator
 	for _, row := range rows {
 		// pick up where each interval left off
-		var paths [2][]byte
-		for i, val := range row {
-			if len(val) != 0 {
-				if _, err = fmt.Sscanf(val, "%x", &paths[i]); err != nil {
-					return nil, err
-				}
+		var startPath []byte
+		var endPath []byte
+		var recoveredPath []byte
+
+		if len(row[0]) != 0 {
+			if _, err = fmt.Sscanf(row[0], "%x", &startPath); err != nil {
+				return nil, err
+			}
+		}
+		if len(row[1]) != 0 {
+			if _, err = fmt.Sscanf(row[1], "%x", &endPath); err != nil {
+				return nil, err
+			}
+		}
+		if len(row[2]) != 0 {
+			if _, err = fmt.Sscanf(row[2], "%x", &recoveredPath); err != nil {
+				return nil, err
 			}
 		}

-		// 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)
+		// force the lower bound path to an even length
+		// (required by HexToKeyBytes())
+		if len(startPath)&0b1 == 1 {
+			// decrement first to avoid skipped nodes
+			decrementPath(startPath)
+			startPath = append(startPath, 0)
 		}
-		it := iter.NewPrefixBoundIterator(tree.NodeIterator(iter.HexToKeyBytes(paths[0])), paths[1])
-		ret = append(ret, tr.tracked(it))
+
+		it := iter.NewPrefixBoundIterator(tree.NodeIterator(iter.HexToKeyBytes(startPath)), startPath, endPath)
+		ret = append(ret, tr.tracked(it, recoveredPath))
 	}
 	return ret, nil
 }
@@ -137,7 +191,7 @@ func (tr *iteratorTracker) restore(tree state.Trie) ([]trie.NodeIterator, error)
 func (tr *iteratorTracker) haltAndDump() error {
 	tr.running = false

-	// drain any pending events
+	// drain any pending iterators
 	close(tr.startChan)
 	for start := range tr.startChan {
 		tr.started[start] = struct{}{}
@@ -159,5 +213,6 @@ func (tr *iteratorTracker) haltAndDump() error {
 		}
 		return err
 	}
+
 	return tr.dump()
 }
@@ -1,15 +1,16 @@
 package snapshot

 import (
+	"bytes"
 	"context"
 	"fmt"

 	"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"
+	"github.com/cerc-io/ipld-eth-state-snapshot/pkg/prom"
+	file "github.com/cerc-io/ipld-eth-state-snapshot/pkg/snapshot/file"
+	pg "github.com/cerc-io/ipld-eth-state-snapshot/pkg/snapshot/pg"
+	snapt "github.com/cerc-io/ipld-eth-state-snapshot/pkg/types"
 )

 func NewPublisher(mode SnapshotMode, config *Config) (snapt.Publisher, error) {
@@ -22,7 +23,7 @@ func NewPublisher(mode SnapshotMode, config *Config) (snapt.Publisher, error) {

 		prom.RegisterDBCollector(config.DB.ConnConfig.DatabaseName, driver)

-		return pg.NewPublisher(postgres.NewPostgresDB(driver)), nil
+		return pg.NewPublisher(postgres.NewPostgresDB(driver, false)), nil
 	case FileSnapshot:
 		return file.NewPublisher(config.File.OutputDir, config.Eth.NodeInfo)
 	}
@@ -51,3 +52,37 @@ func decrementPath(path []byte) bool {
 	}
 	return true
 }
+
+// https://github.com/ethereum/go-ethereum/blob/master/trie/encoding.go#L97
+func keybytesToHex(str []byte) []byte {
+	l := len(str)*2 + 1
+	var nibbles = make([]byte, l)
+	for i, b := range str {
+		nibbles[i*2] = b / 16
+		nibbles[i*2+1] = b % 16
+	}
+	nibbles[l-1] = 16
+	return nibbles
+}
+
+func updateSeekedPath(seekedPath *[]byte, nodePath []byte) {
+	// assumes len(nodePath) <= max len(*seekedPath)
+	*seekedPath = (*seekedPath)[:len(nodePath)]
+	copy(*seekedPath, nodePath)
+}
+
+// checks that the provided node path is before the end path
+func checkUpperPathBound(nodePath, endPath []byte) bool {
+	// every path is before nil endPath
+	if endPath == nil {
+		return true
+	}
+
+	if len(endPath)%2 == 0 {
+		// in case of even length endpath
+		// apply open interval filter since the node at endpath will be covered by the next iterator
+		return bytes.Compare(nodePath, endPath) < 0
+	}
+
+	return bytes.Compare(nodePath, endPath) <= 0
+}
@@ -0,0 +1,73 @@
+## Data Validation
+
+* For a given table in the `ipld-eth-db` schema, we know the number of columns to be expected in each row in the data dump:
+
+  | Table             | Expected columns |
+  | ----------------- |:----------------:|
+  | public.nodes      | 5                |
+  | public.blocks     | 3                |
+  | eth.header_cids   | 16               |
+  | eth.state_cids    | 8                |
+  | eth.storage_cids  | 9                |
+
+### Find Bad Data
+
+* Run the following command to find any rows having unexpected number of columns:
+
+  ```bash
+  ./scripts/find-bad-rows.sh -i <input-file> -c <expected-columns> -o [output-file] -d [include-data]
+  ```
+
+  * `input-file` `-i`: Input data file path
+  * `expected-columns` `-c`: Expected number of columns in each row of the input file
+  * `output-file` `-o`: Output destination file path (default: `STDOUT`)
+  * `include-data` `-d`: Whether to include the data row in the output (`true | false`) (default: `false`)
+  * The output is of format: row number, number of columns, the data row
+
+    Eg:
+
+    ```bash
+    ./scripts/find-bad-rows.sh -i eth.state_cids.csv -c 8 -o res.txt -d true
+    ```
+    
+    Output:
+
+    ```
+    1 9 1500000,xxxxxxxx,0x83952d392f9b0059eea94b10d1a095eefb1943ea91595a16c6698757127d4e1c,,baglacgzasvqcntdahkxhufdnkm7a22s2eetj6mx6nzkarwxtkvy4x3bubdgq,\x0f,0,f,/blocks/,DMQJKYBGZRQDVLT2CRWVGPQNNJNCCJU7GL7G4VAI3LZVK4OL5Q2ARTI
+    ```
+
+    Eg:
+
+    ```bash
+    ./scripts/find-bad-rows.sh -i public.nodes.csv -c 5 -o res.txt -d true
+    ./scripts/find-bad-rows.sh -i public.blocks.csv -c 3 -o res.txt -d true
+    ./scripts/find-bad-rows.sh -i eth.header_cids.csv -c 16 -o res.txt -d true
+    ./scripts/find-bad-rows.sh -i eth.state_cids.csv -c 8 -o res.txt -d true
+    ./scripts/find-bad-rows.sh -i eth.storage_cids.csv -c 9 -o res.txt -d true
+    ```
+
+## Data Cleanup
+
+* In case of column count mismatch, data from `file` mode dumps can't be imported readily into `ipld-eth-db`.
+
+### Filter Bad Data
+
+* Run the following command to filter out rows having unexpected number of columns:
+
+  ```bash
+  ./scripts/filter-bad-rows.sh -i <input-file> -c <expected-columns> -o <output-file>
+  ```
+
+  * `input-file` `-i`: Input data file path
+  * `expected-columns` `-c`: Expected number of columns in each row of the input file
+  * `output-file` `-o`: Output destination file path
+
+    Eg:
+
+    ```bash
+    ./scripts/filter-bad-rows.sh -i public.nodes.csv -c 5 -o cleaned-public.nodes.csv
+    ./scripts/filter-bad-rows.sh -i public.blocks.csv -c 3 -o cleaned-public.blocks.csv
+    ./scripts/filter-bad-rows.sh -i eth.header_cids.csv -c 16 -o cleaned-eth.header_cids.csv
+    ./scripts/filter-bad-rows.sh -i eth.state_cids.csv -c 8 -o cleaned-eth.state_cids.csv
+    ./scripts/filter-bad-rows.sh -i eth.storage_cids.csv -c 9 -o cleaned-eth.storage_cids.csv
+    ```
@@ -0,0 +1,29 @@
+#!/bin/bash
+
+# flags
+# -i <input-file>:        Input data file path
+# -c <expected-columns>:  Expected number of columns in each row of the input file
+# -o [output-file]:       Output destination file path
+
+# eg: ./scripts/filter-bad-rows.sh -i eth.state_cids.csv -c 8 -o cleaned-eth.state_cids.csv
+
+while getopts i:c:o: OPTION
+do
+  case "${OPTION}" in
+    i) inputFile=${OPTARG};;
+    c) expectedColumns=${OPTARG};;
+    o) outputFile=${OPTARG};;
+  esac
+done
+
+timestamp=$(date +%s)
+
+# select only rows having expected number of columns
+if [ -z "${outputFile}" ]; then
+  echo "Invalid destination file arg (-o) ${outputFile}"
+else
+  awk -F"," "NF==${expectedColumns}" ${inputFile} > ${outputFile}
+fi
+
+difference=$(($(date +%s)-timestamp))
+echo Time taken: $(date -d@${difference} -u +%H:%M:%S)
@@ -0,0 +1,43 @@
+#!/bin/bash
+
+# flags
+# -i <input-file>:        Input data file path
+# -c <expected-columns>:  Expected number of columns in each row of the input file
+# -o [output-file]:       Output destination file path (default: STDOUT)
+# -d [include-data]:      Whether to include the data row in output (true | false) (default: false)
+
+# eg: ./scripts/find-bad-rows.sh -i eth.state_cids.csv -c 8 -o res.txt -d true
+# output: 1 9 1500000,xxxxxxxx,0x83952d392f9b0059eea94b10d1a095eefb1943ea91595a16c6698757127d4e1c,,
+#         baglacgzasvqcntdahkxhufdnkm7a22s2eetj6mx6nzkarwxtkvy4x3bubdgq,\x0f,0,f,/blocks/,
+#         DMQJKYBGZRQDVLT2CRWVGPQNNJNCCJU7GL7G4VAI3LZVK4OL5Q2ARTI
+
+while getopts i:c:o:d: OPTION
+do
+  case "${OPTION}" in
+    i) inputFile=${OPTARG};;
+    c) expectedColumns=${OPTARG};;
+    o) outputFile=${OPTARG};;
+    d) data=${OPTARG};;
+  esac
+done
+
+timestamp=$(date +%s)
+
+# if data requested, dump row number, number of columns and the row
+if [ "${data}" = true ] ; then
+  if [ -z "${outputFile}" ]; then
+    awk -F"," "NF!=${expectedColumns} {print NR, NF, \$0}" < ${inputFile}
+  else
+    awk -F"," "NF!=${expectedColumns} {print NR, NF, \$0}" < ${inputFile} > ${outputFile}
+  fi
+# else, dump only row number, number of columns
+else
+  if [ -z "${outputFile}" ]; then
+    awk -F"," "NF!=${expectedColumns} {print NR, NF}" < ${inputFile}
+  else
+    awk -F"," "NF!=${expectedColumns} {print NR, NF}" < ${inputFile} > ${outputFile}
+  fi
+fi
+
+difference=$(($(date +%s)-timestamp))
+echo Time taken: $(date -d@${difference} -u +%H:%M:%S)
Author	SHA1	Message	Date
Ian Norden	e63ffbf0ad	Merge pull request #69 from cerc-io/ian/v4_dev Update to use v4 vdb geth v1.11.5	2023-03-31 12:50:23 -05:00
i-norden	6ede522ae0	missing .meta files in fixtures	2023-03-31 12:36:30 -05:00
i-norden	5c13d59515	refactor to work with v4 vdb geth v1.11.5	2023-03-31 10:39:27 -05:00
i-norden	6fe54aa8b4	use go1.19	2023-03-31 10:39:13 -05:00
i-norden	c9dd85488b	bump vdb v4 geth to v1.11.5	2023-03-31 10:39:00 -05:00
Michael	ac85fe29eb	dependency updates for geth 1.10.26 (#67 )	2022-11-08 14:11:11 -05:00
Michael	9ea78c4ecf	dependency updates for geth 1.10.25 (#65 )	2022-09-26 09:14:15 -04:00
Michael	989bd1c0f1	Cerc refactor (#64 ) * cerc refactor waiting on unpublished dependencies * more cerc_refactor with published ipld-eth-db unstable * TearDownDB refactor * missed second publisher_test TearDownDB	2022-09-20 13:47:34 -04:00
Michael	3f93a989dc	Merge pull request #63 from vulcanize/rebase-1.10.23-wip updates for geth rebase on 1.10.23	2022-09-02 15:39:23 -04:00
Michael Shaw	321e37584d	updates for geth rebase on 1.10.23	2022-09-02 15:21:29 -04:00
prathamesh0	be544a3424	Add helper scripts for data dump correction (#57 ) * Add a script to find bad data in CSV file dumps * Add a script to delete bad rows from CSV file dumps * Add instructions to run the scripts * Reorganize instructions	2022-08-17 15:14:14 +05:30
Ian Nordenandprathamesh	05aeeab581	Account selective snapshot (#46 ) * snapshotter ignores nodes not along a path along those derived from a list of account addresses if one is provided * config and env updates * cmd update * Encode watched address path bytes to hex for comparison * actually ignore the subtries that are not along the paths of interest * Fixes for account selective snapshot * Use non-concurrent iterator when having a single worker * Only index root node when starting path of an iterator is nil * Upgrade deps * Avoid tracking iterators and skip recovery test * Fix recovery mechanism, use sync Map instead of buffered channels * Add test for account selective snapshot * Continue traversal with concurrent iterators with starting path nil * Use errgroup to simplify error handling with concurrent iterators * Check if all the nodes are indexed in the recovery test * Use concurrency safe sync Map in account selective snapshot test * Only track concurrent iterators and refactor code * Fix node and recovered path comparison * Revert back to using buffered channels for tracking iterators * Add a metric to monitor number of active iterators * Update docs * Update seeked path after node is processed * Return error on context cancellation from subtrie iteration * Add tests for account selective snapshot recovery * Explicity enforce concurrent iterator bounds to avoid duplicate nodes * Update full snapshot test to check nodes being indexed * Refactor code to simplify snapshot logic * Remove unnecessary function argument * Use ctx cancellation for handling signals * Add descriptive comments Co-authored-by: prathamesh0 <prathamesh.musale0@gmail.com>	2022-08-03 17:05:04 +05:30