Merge pull request #225 from deep-stack/pm-v4-schema-fixes

Updates to use v4 multi-node DB setup

.github/workflows/on-pr.yml

@@ -15,7 +15,7 @@ jobs:
     name: Run geth unit test
     strategy:
       matrix:
-        go-version: [ 1.16.x]
+        go-version: [1.16.x]
         platform: [ubuntu-latest]
     runs-on: ${{ matrix.platform }}
     env:
@@ -43,7 +43,7 @@ jobs:
       GOPATH: /tmp/go
     strategy:
       matrix:
-        go-version: [ 1.16.x]
+        go-version: [1.16.x]
         platform: [ubuntu-latest]
     runs-on: ${{ matrix.platform }}
     steps:
@@ -58,9 +58,6 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v2
 
-      - name: Start database
-        run:  docker-compose -f docker-compose.yml up -d ipld-eth-db
-
       - name: Run unit tests
         run:
-          make statedifftest
+          ./scripts/run_unit_test.sh

docker-compose.yml

@@ -2,16 +2,13 @@ version: "3.2"
 
 services:
   ipld-eth-db:
-    restart: always
+    restart: on-failure
-    image: vulcanize/ipld-eth-db:v3.2.0
+    depends_on:
+      - access-node
+    image: migrations-test
     environment:
-      POSTGRES_USER: "vdbm"
+      DATABASE_USER: "postgres"
-      POSTGRES_DB: "vulcanize_testing_v3"
+      DATABASE_NAME: "vulcanize_testing_v4"
-      POSTGRES_PASSWORD: "password"
+      DATABASE_PASSWORD: "password"
-    volumes:
+      DATABASE_HOSTNAME: "access-node"
-      - geth_node:/var/lib/postgresql/data
+      DATABASE_PORT: 5432
-    ports:
-      - "127.0.0.1:5432:5432"
-
-volumes:
-  geth_node:

go.sum

@@ -585,7 +585,6 @@ github.com/whyrusleeping/go-logging v0.0.0-20170515211332-0457bb6b88fc/go.mod h1
 github.com/willf/bitset v1.1.3/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
 github.com/xlab/treeprint v0.0.0-20180616005107-d6fb6747feb6/go.mod h1:ce1O1j6UtZfjr22oyGxGLbauSBp2YVXpARAosm7dHBg=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
-github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 github.com/zenazn/goji v0.9.0/go.mod h1:7S9M489iMyHBNxwZnk9/EHS098H4/F6TATF2mIxtB1Q=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
@@ -734,10 +733,8 @@ golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210316164454-77fc1eacc6aa/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210324051608-47abb6519492/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210420205809-ac73e9fd8988/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912 h1:uCLL3g5wH2xjxVREVuAbP9JM5PPKjRbXKRa6IBjkzmU=
 golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -792,8 +789,6 @@ golang.org/x/tools v0.0.0-20200103221440-774c71fcf114/go.mod h1:TB2adYChydJhpapK
 golang.org/x/tools v0.0.0-20200108203644-89082a384178/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.1.0 h1:po9/4sTYwZU9lPhi1tOrb4hCv3qrhiQ77LZfGa2OjwY=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
-golang.org/x/tools v0.1.2 h1:kRBLX7v7Af8W7Gdbbc908OJcdgtK8bOz9Uaj8/F1ACA=
-golang.org/x/tools v0.1.2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/xerrors v0.0.0-20190410155217-1f06c39b4373/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

scripts/run_unit_test.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+
+set -e
+
+# Build database image
+docker build -t migrations-test https://github.com/vulcanize/ipld-eth-db.git#sharding
+
+mkdir -p out
+
+# Remove existing docker-multi-node directory
+rm -rf out/docker-multi-node/
+
+# Copy over files to setup multi-node database
+ID=$(docker create migrations-test)
+docker cp $ID:/app/docker-multi-node out/docker-multi-node/
+docker rm -v $ID
+
+# Spin up multi-node database
+docker-compose -f out/docker-multi-node/docker-compose.test.yml -f docker-compose.yml up ipld-eth-db
+sleep 20
+
+# Run unit tests
+go clean -testcache
+make statedifftest
+
+# Clean up
+docker-compose -f out/docker-multi-node/docker-compose.test.yml -f docker-compose.yml down --remove-orphans --volumes
+rm -rf out/docker-multi-node/

statediff/indexer/database/file/indexer_test.go

@@ -51,8 +51,8 @@ var (
 	sqlxdb    *sqlx.DB
 	err       error
 	ind       interfaces.StateDiffIndexer
-	ipfsPgGet = `SELECT data FROM public.blocks
+	ipfsPgGet = `SELECT COALESCE(data, '') as data FROM public.blocks
-					WHERE key = $1`
+					WHERE key = $1 AND block_number = $2`
 	tx1, tx2, tx3, tx4, tx5, rct1, rct2, rct3, rct4, rct5                          []byte
 	mockBlock                                                                      *types.Block
 	headerCID, trx1CID, trx2CID, trx3CID, trx4CID, trx5CID                         cid.Cid
@@ -262,7 +262,7 @@ func TestFileIndexer(t *testing.T) {
 		mhKey := dshelp.MultihashToDsKey(dc.Hash())
 		prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
 		var data []byte
-		err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey)
+		err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey, mocks.BlockNumber.Uint64())
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -301,7 +301,7 @@ func TestFileIndexer(t *testing.T) {
 			mhKey := dshelp.MultihashToDsKey(dc.Hash())
 			prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
 			var data []byte
-			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey)
+			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey, mocks.BlockNumber.Uint64())
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -425,7 +425,7 @@ func TestFileIndexer(t *testing.T) {
 		}
 		for i := range rcts {
 			results := make([]logIPLD, 0)
-			pgStr = `SELECT log_cids.index, log_cids.address, log_cids.topic0, log_cids.topic1, data FROM eth.log_cids
+			pgStr = `SELECT log_cids.index, log_cids.address, COALESCE(log_cids.topic0, '') as topic0, COALESCE(log_cids.topic1, '') as topic1, data FROM eth.log_cids
     				INNER JOIN eth.receipt_cids ON (log_cids.rct_id = receipt_cids.tx_id)
 					INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key)
 					WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC`
@@ -480,7 +480,7 @@ func TestFileIndexer(t *testing.T) {
 
 		for idx, c := range rcts {
 			result := make([]models.IPLDModel, 0)
-			pgStr = `SELECT data
+			pgStr = `SELECT COALESCE(data, '') as data
 					FROM eth.receipt_cids
 					INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = public.blocks.key)
 					WHERE receipt_cids.leaf_cid = $1`
@@ -506,11 +506,11 @@ func TestFileIndexer(t *testing.T) {
 			mhKey := dshelp.MultihashToDsKey(dc.Hash())
 			prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
 			var data []byte
-			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey)
+			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey, mocks.BlockNumber.Uint64())
 			if err != nil {
 				t.Fatal(err)
 			}
-			postStatePgStr := `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
+			postStatePgStr := `SELECT COALESCE(post_state, '') as post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 			switch c {
 			case rct1CID.String():
 				require.Equal(t, rctLeaf1, data)
@@ -564,7 +564,7 @@ func TestFileIndexer(t *testing.T) {
 
 		// check that state nodes were properly indexed and published
 		stateNodes := make([]models.StateNodeModel, 0)
-		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
+		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, COALESCE(state_cids.state_path, '') as state_path, state_cids.header_id
 				FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 				WHERE header_cids.block_number = $1 AND node_type != 3`
 		err = sqlxdb.Select(&stateNodes, pgStr, mocks.BlockNumber.Uint64())
@@ -580,11 +580,11 @@ func TestFileIndexer(t *testing.T) {
 			}
 			mhKey := dshelp.MultihashToDsKey(dc.Hash())
 			prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
-			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey)
+			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey, mocks.BlockNumber.Uint64())
 			if err != nil {
 				t.Fatal(err)
 			}
-			pgStr = `SELECT cast(block_number AS TEXT), header_id, state_path, cast(balance AS TEXT), nonce, code_hash, storage_root from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
+			pgStr = `SELECT cast(block_number AS TEXT), header_id, state_path, cast(balance AS TEXT), nonce, COALESCE(code_hash, '') as code_hash, storage_root from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
 			var account models.StateAccountModel
 			err = sqlxdb.Get(&account, pgStr, stateNode.HeaderID, stateNode.Path)
 			if err != nil {
@@ -641,7 +641,7 @@ func TestFileIndexer(t *testing.T) {
 			mhKey := dshelp.MultihashToDsKey(dc.Hash())
 			prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
 			require.Equal(t, shared.RemovedNodeMhKey, prefixedKey)
-			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey)
+			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey, mocks.BlockNumber.Uint64())
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -668,7 +668,7 @@ func TestFileIndexer(t *testing.T) {
 
 		// check that storage nodes were properly indexed
 		storageNodes := make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
+		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash
@@ -694,7 +694,7 @@ func TestFileIndexer(t *testing.T) {
 		}
 		mhKey := dshelp.MultihashToDsKey(dc.Hash())
 		prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
-		err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey)
+		err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey, mocks.BlockNumber.Uint64())
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -702,7 +702,7 @@ func TestFileIndexer(t *testing.T) {
 
 		// check that Removed storage nodes were properly indexed
 		storageNodes = make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
+		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash
@@ -747,8 +747,8 @@ func TestFileIndexer(t *testing.T) {
 			}
 			mhKey = dshelp.MultihashToDsKey(dc.Hash())
 			prefixedKey = blockstore.BlockPrefix.String() + mhKey.String()
-			require.Equal(t, shared.RemovedNodeMhKey, prefixedKey)
+			require.Equal(t, shared.RemovedNodeMhKey, prefixedKey, mocks.BlockNumber.Uint64())
-			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey)
+			err = sqlxdb.Get(&data, ipfsPgGet, prefixedKey, mocks.BlockNumber.Uint64())
 			if err != nil {
 				t.Fatal(err)
 			}

statediff/indexer/database/file/mainnet_tests/indexer_test.go

@@ -107,7 +107,7 @@ func setup(t *testing.T, testBlock *types.Block, testReceipts types.Receipts) {
 		require.NoError(t, err)
 	}
 
-	test_helpers.ExpectEqual(t, tx.(*file.BatchTx).BlockNumber, testBlock.Number().Uint64())
+	require.Equal(t, testBlock.Number().String(), tx.(*file.BatchTx).BlockNumber)
 
 	connStr := postgres.DefaultConfig.DbConnectionString()
 

statediff/indexer/database/sql/indexer_shared_test.go

@@ -22,7 +22,7 @@ var (
 	db        sql.Database
 	err       error
 	ind       interfaces.StateDiffIndexer
-	ipfsPgGet = `SELECT data FROM public.blocks
+	ipfsPgGet = `SELECT COALESCE(data, '') as data FROM public.blocks
 					WHERE key = $1 AND block_number = $2`
 	tx1, tx2, tx3, tx4, tx5, rct1, rct2, rct3, rct4, rct5                          []byte
 	mockBlock                                                                      *types.Block

statediff/indexer/database/sql/mainnet_tests/indexer_test.go

@@ -99,7 +99,7 @@ func setup(t *testing.T, testBlock *types.Block, testReceipts types.Receipts) {
 		require.NoError(t, err)
 	}
 
-	test_helpers.ExpectEqual(t, tx.(*sql.BatchTx).BlockNumber, testBlock.Number().Uint64())
+	require.Equal(t, testBlock.Number().String(), tx.(*sql.BatchTx).BlockNumber)
 }
 
 func tearDown(t *testing.T) {

statediff/indexer/database/sql/pgx_indexer_test.go

@@ -21,8 +21,6 @@ import (
 	"math/big"
 	"testing"
 
-	"github.com/lib/pq"
-
 	"github.com/ipfs/go-cid"
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	dshelp "github.com/ipfs/go-ipfs-ds-help"
@@ -207,14 +205,6 @@ func TestPGXIndexer(t *testing.T) {
 				if txRes.Value != transactions[3].Value().String() {
 					t.Fatalf("expected tx value %s got %s", transactions[3].Value().String(), txRes.Value)
 				}
-				// AccessListElementModel is the db model for eth.access_list_entry
-				type AccessListElementModel struct {
-					BlockNumber string         `db:"block_number"`
-					Index       int64          `db:"index"`
-					TxID        string         `db:"tx_id"`
-					Address     string         `db:"address"`
-					StorageKeys pq.StringArray `db:"storage_keys"`
-				}
 				accessListElementModels := make([]models.AccessListElementModel, 0)
 				pgStr = "SELECT cast(access_list_elements.block_number AS TEXT), access_list_elements.index, access_list_elements.tx_id, " +
 					"access_list_elements.address, access_list_elements.storage_keys FROM eth.access_list_elements " +
@@ -267,7 +257,7 @@ func TestPGXIndexer(t *testing.T) {
 				AND transaction_cids.header_id = header_cids.block_hash
 				AND header_cids.block_number = $1
 				ORDER BY transaction_cids.index`
-		logsPgStr := `SELECT log_cids.index, log_cids.address, log_cids.topic0, log_cids.topic1, data FROM eth.log_cids
+		logsPgStr := `SELECT log_cids.index, log_cids.address, COALESCE(log_cids.topic0, '') as topic0, COALESCE(log_cids.topic1, '') as topic1, data FROM eth.log_cids
     				INNER JOIN eth.receipt_cids ON (log_cids.rct_id = receipt_cids.tx_id)
 					INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key)
 					WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC`
@@ -338,7 +328,7 @@ func TestPGXIndexer(t *testing.T) {
 
 		for idx, c := range rcts {
 			result := make([]models.IPLDModel, 0)
-			pgStr = `SELECT data
+			pgStr = `SELECT COALESCE(data, '') as data
 					FROM eth.receipt_cids
 					INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = public.blocks.key)
 					WHERE receipt_cids.leaf_cid = $1`
@@ -369,7 +359,7 @@ func TestPGXIndexer(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			postStatePgStr := `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
+			postStatePgStr := `SELECT COALESCE(post_state, '') as post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 			switch c {
 			case rct1CID.String():
 				require.Equal(t, rctLeaf1, data)
@@ -422,7 +412,7 @@ func TestPGXIndexer(t *testing.T) {
 		defer checkTxClosure(t, 1, 0, 1)
 		// check that state nodes were properly indexed and published
 		stateNodes := make([]models.StateNodeModel, 0)
-		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
+		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, COALESCE(state_cids.state_path, '') as state_path, state_cids.header_id
 				FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 				WHERE header_cids.block_number = $1 AND node_type != 3`
 		err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
@@ -442,7 +432,7 @@ func TestPGXIndexer(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			pgStr = `SELECT cast(block_number AS TEXT), header_id, state_path, cast(balance AS TEXT), nonce, code_hash, storage_root from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
+			pgStr = `SELECT cast(block_number AS TEXT), header_id, state_path, cast(balance AS TEXT), nonce, COALESCE(code_hash, '') as code_hash, storage_root from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
 			var account models.StateAccountModel
 			err = db.Get(context.Background(), &account, pgStr, stateNode.HeaderID, stateNode.Path)
 			if err != nil {
@@ -482,7 +472,7 @@ func TestPGXIndexer(t *testing.T) {
 
 		// check that Removed state nodes were properly indexed and published
 		stateNodes = make([]models.StateNodeModel, 0)
-		pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
+		pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, COALESCE(state_cids.state_path, '') as state_path, state_cids.header_id
 				FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 				WHERE header_cids.block_number = $1 AND node_type = 3`
 		err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
@@ -525,7 +515,7 @@ func TestPGXIndexer(t *testing.T) {
 		defer checkTxClosure(t, 1, 0, 1)
 		// check that storage nodes were properly indexed
 		storageNodes := make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
+		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash
@@ -559,7 +549,7 @@ func TestPGXIndexer(t *testing.T) {
 
 		// check that Removed storage nodes were properly indexed
 		storageNodes = make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
+		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash

statediff/indexer/database/sql/postgres/config.go

@@ -48,10 +48,10 @@ func ResolveDriverType(str string) (DriverType, error) {
 // DefaultConfig are default parameters for connecting to a Postgres sql
 var DefaultConfig = Config{
 	Hostname:     "localhost",
-	Port:         5432,
+	Port:         8066,
 	DatabaseName: "vulcanize_testing_v4",
 	Username:     "postgres",
-	Password:     "",
+	Password:     "password",
 }
 
 // Config holds params for a Postgres db

statediff/indexer/database/sql/postgres/database.go

@@ -40,65 +40,65 @@ type DB struct {
 func (db *DB) InsertHeaderStm() string {
 	return `INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase)
 			VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
-			ON CONFLICT (block_hash) DO UPDATE SET (parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase) = ($3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, eth.header_cids.times_validated + 1, $16)`
+			ON CONFLICT DO NOTHING`
 }
 
 // InsertUncleStm satisfies the sql.Statements interface
 func (db *DB) InsertUncleStm() string {
 	return `INSERT INTO eth.uncle_cids (block_number, block_hash, header_id, parent_hash, cid, reward, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7)
-			ON CONFLICT (block_hash) DO NOTHING`
+			ON CONFLICT DO NOTHING`
 }
 
 // InsertTxStm satisfies the sql.Statements interface
 func (db *DB) InsertTxStm() string {
 	return `INSERT INTO eth.transaction_cids (block_number, header_id, tx_hash, cid, dst, src, index, mh_key, tx_data, tx_type, value) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-			ON CONFLICT (tx_hash) DO NOTHING`
+			ON CONFLICT DO NOTHING`
 }
 
 // InsertAccessListElementStm satisfies the sql.Statements interface
 func (db *DB) InsertAccessListElementStm() string {
 	return `INSERT INTO eth.access_list_elements (block_number, tx_id, index, address, storage_keys) VALUES ($1, $2, $3, $4, $5)
-			ON CONFLICT (tx_id, index) DO NOTHING`
+			ON CONFLICT DO NOTHING`
 }
 
 // InsertRctStm satisfies the sql.Statements interface
 func (db *DB) InsertRctStm() string {
 	return `INSERT INTO eth.receipt_cids (block_number, tx_id, leaf_cid, contract, contract_hash, leaf_mh_key, post_state, post_status, log_root) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-		  	ON CONFLICT (tx_id) DO NOTHING`
+		  	ON CONFLICT DO NOTHING`
 }
 
 // InsertLogStm satisfies the sql.Statements interface
 func (db *DB) InsertLogStm() string {
 	return `INSERT INTO eth.log_cids (block_number, leaf_cid, leaf_mh_key, rct_id, address, index, topic0, topic1, topic2, topic3, log_data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-			ON CONFLICT (rct_id, index) DO NOTHING`
+			ON CONFLICT DO NOTHING`
 }
 
 // InsertStateStm satisfies the sql.Statements interface
 func (db *DB) InsertStateStm() string {
 	return `INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
-			ON CONFLICT (header_id, state_path) DO UPDATE SET (block_number, state_leaf_key, cid, node_type, diff, mh_key) = ($1, $3, $4, $6, $7, $8)`
+			ON CONFLICT DO NOTHING`
 }
 
 // InsertAccountStm satisfies the sql.Statements interface
 func (db *DB) InsertAccountStm() string {
 	return `INSERT INTO eth.state_accounts (block_number, header_id, state_path, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5, $6, $7)
-		  	ON CONFLICT (header_id, state_path) DO NOTHING`
+		  	ON CONFLICT DO NOTHING`
 }
 
 // InsertStorageStm satisfies the sql.Statements interface
 func (db *DB) InsertStorageStm() string {
 	return `INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-		  	ON CONFLICT (header_id, state_path, storage_path) DO UPDATE SET (block_number, storage_leaf_key, cid, node_type, diff, mh_key) = ($1, $4, $5, $7, $8, $9)`
+		  	ON CONFLICT DO NOTHING`
 }
 
 // InsertIPLDStm satisfies the sql.Statements interface
 func (db *DB) InsertIPLDStm() string {
-	return `INSERT INTO public.blocks (block_number, key, data) VALUES ($1, $2, $3) ON CONFLICT (block_number, key) DO NOTHING`
+	return `INSERT INTO public.blocks (block_number, key, data) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING`
 }
 
 // InsertIPLDsStm satisfies the sql.Statements interface
 func (db *DB) InsertIPLDsStm() string {
-	return `INSERT INTO public.blocks (block_number, key, data) VALUES (unnest($1::BIGINT[]), unnest($2::TEXT[]), unnest($3::BYTEA[])) ON CONFLICT (block_number, key) DO NOTHING`
+	return `INSERT INTO public.blocks (block_number, key, data) VALUES (unnest($1::BIGINT[]), unnest($2::TEXT[]), unnest($3::BYTEA[])) ON CONFLICT DO NOTHING`
 }
 
 // InsertKnownGapsStm satisfies the sql.Statements interface

statediff/indexer/database/sql/postgres/pgx_test.go

@@ -24,10 +24,10 @@ import (
 	"testing"
 
 	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/stretchr/testify/require"
 
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/ethereum/go-ethereum/statediff/indexer/node"
-	"github.com/ethereum/go-ethereum/statediff/indexer/test_helpers"
 )
 
 var (
@@ -69,7 +69,7 @@ func TestPostgresPGX(t *testing.T) {
 
 		bi := new(big.Int)
 		bi.SetString("34940183920000000000", 10)
-		test_helpers.ExpectEqual(t, bi.String(), "34940183920000000000")
+		require.Equal(t, "34940183920000000000", bi.String())
 
 		defer dbPool.Exec(ctx, `DROP TABLE IF EXISTS example`)
 		_, err = dbPool.Exec(ctx, "CREATE TABLE example ( id INTEGER, data NUMERIC )")
@@ -91,10 +91,10 @@ func TestPostgresPGX(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		test_helpers.ExpectEqual(t, data, bi.String())
+		require.Equal(t, data, bi.String())
 		actual := new(big.Int)
 		actual.SetString(data, 10)
-		test_helpers.ExpectEqual(t, actual, bi)
+		require.Equal(t, bi, actual)
 	})
 
 	t.Run("throws error when can't connect to the database", func(t *testing.T) {

statediff/indexer/database/sql/postgres/sqlx_test.go

@@ -24,10 +24,10 @@ import (
 
 	"github.com/jmoiron/sqlx"
 	_ "github.com/lib/pq"
+	"github.com/stretchr/testify/require"
 
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/ethereum/go-ethereum/statediff/indexer/node"
-	"github.com/ethereum/go-ethereum/statediff/indexer/test_helpers"
 )
 
 func TestPostgresSQLX(t *testing.T) {
@@ -67,7 +67,7 @@ func TestPostgresSQLX(t *testing.T) {
 
 		bi := new(big.Int)
 		bi.SetString("34940183920000000000", 10)
-		test_helpers.ExpectEqual(t, bi.String(), "34940183920000000000")
+		require.Equal(t, "34940183920000000000", bi.String())
 
 		defer db.Exec(`DROP TABLE IF EXISTS example`)
 		_, err = db.Exec("CREATE TABLE example ( id INTEGER, data NUMERIC )")
@@ -89,10 +89,10 @@ func TestPostgresSQLX(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		test_helpers.ExpectEqual(t, data, bi.String())
+		require.Equal(t, data, bi.String())
 		actual := new(big.Int)
 		actual.SetString(data, 10)
-		test_helpers.ExpectEqual(t, actual, bi)
+		require.Equal(t, bi, actual)
 	})
 
 	t.Run("throws error when can't connect to the database", func(t *testing.T) {

statediff/indexer/database/sql/sqlx_indexer_test.go

@@ -253,7 +253,7 @@ func TestSQLXIndexer(t *testing.T) {
 				AND transaction_cids.header_id = header_cids.block_hash
 				AND header_cids.block_number = $1
 				ORDER BY transaction_cids.index`
-		logsPgStr := `SELECT log_cids.index, log_cids.address, log_cids.topic0, log_cids.topic1, data FROM eth.log_cids
+		logsPgStr := `SELECT log_cids.index, log_cids.address, COALESCE(log_cids.topic0, '') as topic0, COALESCE(log_cids.topic1, '') as topic1, data FROM eth.log_cids
     				INNER JOIN eth.receipt_cids ON (log_cids.rct_id = receipt_cids.tx_id)
 					INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key)
 					WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC`
@@ -322,7 +322,7 @@ func TestSQLXIndexer(t *testing.T) {
 
 		for idx, c := range rcts {
 			result := make([]models.IPLDModel, 0)
-			pgStr = `SELECT data
+			pgStr = `SELECT COALESCE(data, '') as data
 					FROM eth.receipt_cids
 					INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = public.blocks.key)
 					WHERE receipt_cids.leaf_cid = $1`
@@ -352,7 +352,7 @@ func TestSQLXIndexer(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			postStatePgStr := `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
+			postStatePgStr := `SELECT COALESCE(post_state, '') as post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 			switch c {
 			case rct1CID.String():
 				require.Equal(t, rctLeaf1, data)
@@ -405,7 +405,7 @@ func TestSQLXIndexer(t *testing.T) {
 		defer checkTxClosure(t, 0, 0, 0)
 		// check that state nodes were properly indexed and published
 		stateNodes := make([]models.StateNodeModel, 0)
-		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
+		pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, COALESCE(state_cids.state_path, '') as state_path, state_cids.header_id
 				FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 				WHERE header_cids.block_number = $1 AND node_type != 3`
 		err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
@@ -465,7 +465,7 @@ func TestSQLXIndexer(t *testing.T) {
 
 		// check that Removed state nodes were properly indexed and published
 		stateNodes = make([]models.StateNodeModel, 0)
-		pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
+		pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, COALESCE(state_cids.state_path, '') as state_path, state_cids.header_id
 				FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 				WHERE header_cids.block_number = $1 AND node_type = 3`
 		err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
@@ -508,7 +508,7 @@ func TestSQLXIndexer(t *testing.T) {
 		defer checkTxClosure(t, 0, 0, 0)
 		// check that storage nodes were properly indexed
 		storageNodes := make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
+		pgStr := `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash
@@ -542,7 +542,7 @@ func TestSQLXIndexer(t *testing.T) {
 
 		// check that Removed storage nodes were properly indexed
 		storageNodes = make([]models.StorageNodeWithStateKeyModel, 0)
-		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
+		pgStr = `SELECT cast(storage_cids.block_number AS TEXT), storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, COALESCE(storage_cids.storage_path, '') as storage_path
 				FROM eth.storage_cids, eth.state_cids, eth.header_cids
 				WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
 				AND state_cids.header_id = header_cids.block_hash

statediff/indexer/database/sql/writer.go

@@ -47,7 +47,7 @@ func (w *Writer) Close() error {
 /*
 INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase)
 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
-ON CONFLICT (block_hash) DO UPDATE SET (block_number, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase) = ($1, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, eth.header_cids.times_validated + 1, $16)
+ON CONFLICT (block_hash, block_number) DO UPDATE SET (block_number, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase) = ($1, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, eth.header_cids.times_validated + 1, $16)
 */
 func (w *Writer) upsertHeaderCID(tx Tx, header models.HeaderModel) error {
 	_, err := tx.Exec(w.db.Context(), w.db.InsertHeaderStm(),
@@ -63,7 +63,7 @@ func (w *Writer) upsertHeaderCID(tx Tx, header models.HeaderModel) error {
 
 /*
 INSERT INTO eth.uncle_cids (block_number, block_hash, header_id, parent_hash, cid, reward, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7)
-ON CONFLICT (block_hash) DO NOTHING
+ON CONFLICT (block_hash, block_number) DO NOTHING
 */
 func (w *Writer) upsertUncleCID(tx Tx, uncle models.UncleModel) error {
 	_, err := tx.Exec(w.db.Context(), w.db.InsertUncleStm(),
@@ -76,7 +76,7 @@ func (w *Writer) upsertUncleCID(tx Tx, uncle models.UncleModel) error {
 
 /*
 INSERT INTO eth.transaction_cids (block_number, header_id, tx_hash, cid, dst, src, index, mh_key, tx_data, tx_type, value) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-ON CONFLICT (tx_hash) DO NOTHING
+ON CONFLICT (tx_hash, block_number) DO NOTHING
 */
 func (w *Writer) upsertTransactionCID(tx Tx, transaction models.TxModel) error {
 	_, err := tx.Exec(w.db.Context(), w.db.InsertTxStm(),
@@ -91,7 +91,7 @@ func (w *Writer) upsertTransactionCID(tx Tx, transaction models.TxModel) error {
 
 /*
 INSERT INTO eth.access_list_elements (block_number, tx_id, index, address, storage_keys) VALUES ($1, $2, $3, $4, $5)
-ON CONFLICT (tx_id, index) DO NOTHING
+ON CONFLICT (tx_id, index, block_number) DO NOTHING
 */
 func (w *Writer) upsertAccessListElement(tx Tx, accessListElement models.AccessListElementModel) error {
 	_, err := tx.Exec(w.db.Context(), w.db.InsertAccessListElementStm(),
@@ -106,7 +106,7 @@ func (w *Writer) upsertAccessListElement(tx Tx, accessListElement models.AccessL
 
 /*
 INSERT INTO eth.receipt_cids (block_number, tx_id, leaf_cid, contract, contract_hash, leaf_mh_key, post_state, post_status, log_root) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-ON CONFLICT (tx_id) DO NOTHING
+ON CONFLICT (tx_id, block_number) DO NOTHING
 */
 func (w *Writer) upsertReceiptCID(tx Tx, rct *models.ReceiptModel) error {
 	_, err := tx.Exec(w.db.Context(), w.db.InsertRctStm(),
@@ -121,7 +121,7 @@ func (w *Writer) upsertReceiptCID(tx Tx, rct *models.ReceiptModel) error {
 
 /*
 INSERT INTO eth.log_cids (block_number, leaf_cid, leaf_mh_key, rct_id, address, index, topic0, topic1, topic2, topic3, log_data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-ON CONFLICT (rct_id, index) DO NOTHING
+ON CONFLICT (rct_id, index, block_number) DO NOTHING
 */
 func (w *Writer) upsertLogCID(tx Tx, logs []*models.LogsModel) error {
 	for _, log := range logs {
@@ -138,7 +138,7 @@ func (w *Writer) upsertLogCID(tx Tx, logs []*models.LogsModel) error {
 
 /*
 INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
-ON CONFLICT (header_id, state_path) DO UPDATE SET (block_number, state_leaf_key, cid, node_type, diff, mh_key) = ($1 $3, $4, $6, $7, $8)
+ON CONFLICT (header_id, state_path, block_number) DO UPDATE SET (block_number, state_leaf_key, cid, node_type, diff, mh_key) = ($1 $3, $4, $6, $7, $8)
 */
 func (w *Writer) upsertStateCID(tx Tx, stateNode models.StateNodeModel) error {
 	var stateKey string
@@ -156,7 +156,7 @@ func (w *Writer) upsertStateCID(tx Tx, stateNode models.StateNodeModel) error {
 
 /*
 INSERT INTO eth.state_accounts (block_number, header_id, state_path, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5, $6, $7)
-ON CONFLICT (header_id, state_path) DO NOTHING
+ON CONFLICT (header_id, state_path, block_number) DO NOTHING
 */
 func (w *Writer) upsertStateAccount(tx Tx, stateAccount models.StateAccountModel) error {
 	_, err := tx.Exec(w.db.Context(), w.db.InsertAccountStm(),
@@ -170,7 +170,7 @@ func (w *Writer) upsertStateAccount(tx Tx, stateAccount models.StateAccountModel
 
 /*
 INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-ON CONFLICT (header_id, state_path, storage_path) DO UPDATE SET (block_number, storage_leaf_key, cid, node_type, diff, mh_key) = ($1, $4, $5, $7, $8, $9)
+ON CONFLICT (header_id, state_path, storage_path, block_number) DO UPDATE SET (block_number, storage_leaf_key, cid, node_type, diff, mh_key) = ($1, $4, $5, $7, $8, $9)
 */
 func (w *Writer) upsertStorageCID(tx Tx, storageCID models.StorageNodeModel) error {
 	var storageKey string