Remove watched addresses fill service code

Combine the build and push steps.

.github/workflows/on-master.yaml

@@ -1,25 +0,0 @@
-name: Docker Compose Build
-
-on:
-  push:
-    branches:
-      - master
-
-jobs:
-  build:
-    name: Run docker build
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v2
-      - name: Get the version
-        id: vars
-        run: echo ::set-output name=sha::$(echo ${GITHUB_SHA:0:7})
-      - name: Run docker build
-        run: make docker-build
-      - name: Tag docker image
-        run: docker tag vulcanize/ipld-eth-server docker.pkg.github.com/vulcanize/ipld-eth-server/ipld-eth-server:${{steps.vars.outputs.sha}}
-      - name: Docker Login
-        run: echo ${{ secrets.GITHUB_TOKEN }} | docker login https://docker.pkg.github.com -u vulcanize --password-stdin
-      - name: Docker Push
-        run: docker push docker.pkg.github.com/vulcanize/ipld-eth-server/ipld-eth-server:${{steps.vars.outputs.sha}}
-

.github/workflows/on-pr.yaml

@@ -60,15 +60,13 @@ jobs:
   integrationtest:
     name: Run integration tests
     env:
-      STACK_ORCHESTRATOR_REF: 38545ad36144800d16d61e53050bb8e784729f26
-      GO_ETHEREUM_REF: ef8c9f2580dd577a7c5eca538fb0ed64d53dc4a4
+      STACK_ORCHESTRATOR_REF: fcbc74451c5494664fe21f765e89c9c6565c07cb
+      GO_ETHEREUM_REF: 498101102c891c4f8c3cab5649158c642ee1fd6b
       GOPATH: /tmp/go
       DB_WRITE: true
       ETH_FORWARD_ETH_CALLS: false
       ETH_PROXY_ON_ERROR: false
       ETH_HTTP_PATH: "go-ethereum:8545"
-      WATCHED_ADDRESS_GAP_FILLER_ENABLED: false
-      WATCHED_ADDRESS_GAP_FILLER_INTERVAL: 2
     strategy:
       matrix:
         go-version: [1.16.x]
@@ -127,15 +125,13 @@ jobs:
   integrationtest_forwardethcalls:
     name: Run integration tests for direct proxy fall-through of eth_calls
     env:
-      STACK_ORCHESTRATOR_REF: 38545ad36144800d16d61e53050bb8e784729f26
-      GO_ETHEREUM_REF: ef8c9f2580dd577a7c5eca538fb0ed64d53dc4a4
+      STACK_ORCHESTRATOR_REF: fcbc74451c5494664fe21f765e89c9c6565c07cb
+      GO_ETHEREUM_REF: 498101102c891c4f8c3cab5649158c642ee1fd6b
       GOPATH: /tmp/go
       DB_WRITE: false
       ETH_FORWARD_ETH_CALLS: true
       ETH_PROXY_ON_ERROR: false
       ETH_HTTP_PATH: "go-ethereum:8545"
-      WATCHED_ADDRESS_GAP_FILLER_ENABLED: false
-      WATCHED_ADDRESS_GAP_FILLER_INTERVAL: 2
     strategy:
       matrix:
         go-version: [1.16.x]
@@ -190,72 +186,3 @@ jobs:
           while [ "$(curl -s -o /dev/null -w ''%{http_code}'' localhost:8081)" != "200" ]; do echo "waiting for ipld-eth-server..." && sleep 5; done && \
           while [ "$(curl -s -o /dev/null -w ''%{http_code}'' localhost:8545)" != "200" ]; do echo "waiting for geth-statediff..." && sleep 5; done && \
           make integrationtest
-
-  integrationtest_watchedaddress_gapfillingservice:
-    name: Run integration tests for watched addresses with gap filling service enabled
-    env:
-      STACK_ORCHESTRATOR_REF: 38545ad36144800d16d61e53050bb8e784729f26
-      GO_ETHEREUM_REF: ef8c9f2580dd577a7c5eca538fb0ed64d53dc4a4
-      GOPATH: /tmp/go
-      DB_WRITE: true
-      ETH_FORWARD_ETH_CALLS: false
-      ETH_PROXY_ON_ERROR: false
-      ETH_HTTP_PATH: "go-ethereum:8545"
-      WATCHED_ADDRESS_GAP_FILLER_ENABLED: true
-      WATCHED_ADDRESS_GAP_FILLER_INTERVAL: 2
-    strategy:
-      matrix:
-        go-version: [1.16.x]
-        os: [ubuntu-latest]
-    runs-on: ${{ matrix.os }}
-    steps:
-      - name: Create GOPATH
-        run: mkdir -p /tmp/go
-      - name: Install Go
-        uses: actions/setup-go@v2
-        with:
-          go-version: ${{ matrix.go-version }}
-      - uses: actions/checkout@v2
-        with:
-          path: "./ipld-eth-server"
-      - uses: actions/checkout@v2
-        with:
-          ref: ${{ env.STACK_ORCHESTRATOR_REF }}
-          path: "./stack-orchestrator/"
-          repository: vulcanize/stack-orchestrator
-      - uses: actions/checkout@v2
-        with:
-          ref: ${{ env.GO_ETHEREUM_REF }}
-          repository: vulcanize/go-ethereum
-          path: "./go-ethereum/"
-      - name: Create config file
-        run: |
-          echo vulcanize_go_ethereum=$GITHUB_WORKSPACE/go-ethereum/ > ./config.sh
-          echo vulcanize_ipld_eth_server=$GITHUB_WORKSPACE/ipld-eth-server/ >> ./config.sh
-          echo db_write=$DB_WRITE >> ./config.sh
-          echo eth_forward_eth_calls=$ETH_FORWARD_ETH_CALLS >> ./config.sh
-          echo eth_proxy_on_error=$ETH_PROXY_ON_ERROR >> ./config.sh
-          echo eth_http_path=$ETH_HTTP_PATH >> ./config.sh
-          echo watched_addres_gap_filler_enabled=$WATCHED_ADDRESS_GAP_FILLER_ENABLED >> ./config.sh
-          echo watched_addres_gap_filler_interval=$WATCHED_ADDRESS_GAP_FILLER_INTERVAL >> ./config.sh
-          cat ./config.sh
-      - name: Build geth
-        run: |
-          cd $GITHUB_WORKSPACE/stack-orchestrator/helper-scripts
-          ./compile-geth.sh \
-          -p "$GITHUB_WORKSPACE/config.sh" \
-          -e docker
-      - name: Run docker compose
-        run: |
-          docker-compose  \
-          -f "$GITHUB_WORKSPACE/stack-orchestrator/docker/latest/docker-compose-db.yml" \
-          -f "$GITHUB_WORKSPACE/stack-orchestrator/docker/local/docker-compose-go-ethereum.yml" \
-          -f "$GITHUB_WORKSPACE/stack-orchestrator/docker/local/docker-compose-ipld-eth-server.yml" \
-          --env-file "$GITHUB_WORKSPACE/config.sh" \
-          up -d --build
-      - name: Test
-        run: |
-          cd $GITHUB_WORKSPACE/ipld-eth-server
-          while [ "$(curl -s -o /dev/null -w ''%{http_code}'' localhost:8081)" != "200" ]; do echo "waiting for ipld-eth-server..." && sleep 5; done && \
-          while [ "$(curl -s -o /dev/null -w ''%{http_code}'' localhost:8545)" != "200" ]; do echo "waiting for geth-statediff..." && sleep 5; done && \
-          make integrationtest

.github/workflows/publish.yaml

@@ -3,9 +3,27 @@ on:
   release:
     types: [published]
 jobs:
+  build:
+    name: Run docker build
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Get the version
+        id: vars
+        run: echo ::set-output name=sha::$(echo ${GITHUB_SHA:0:7})
+      - name: Run docker build
+        run: make docker-build
+      - name: Tag docker image
+        run: docker tag vulcanize/ipld-eth-server docker.pkg.github.com/vulcanize/ipld-eth-server/ipld-eth-server:${{steps.vars.outputs.sha}}
+      - name: Docker Login
+        run: echo ${{ secrets.GITHUB_TOKEN }} | docker login https://docker.pkg.github.com -u vulcanize --password-stdin
+      - name: Docker Push
+        run: docker push docker.pkg.github.com/vulcanize/ipld-eth-server/ipld-eth-server:${{steps.vars.outputs.sha}}
+
   push_to_registries:
     name: Push Docker image to Docker Hub
     runs-on: ubuntu-latest
+    needs: build
     steps:
       - name: Get the version
         id: vars
@@ -22,4 +40,3 @@ jobs:
         run: docker tag docker.pkg.github.com/vulcanize/ipld-eth-server/ipld-eth-server:${{steps.vars.outputs.sha}} vulcanize/ipld-eth-server:${{steps.vars.outputs.tag}}
       - name: Docker Push to Docker Hub
         run: docker push vulcanize/ipld-eth-server:${{steps.vars.outputs.tag}}
-

.github/workflows/run_unit_test.sh

@@ -8,30 +8,22 @@ temp_dir=$(mktemp -d)
 cd $temp_dir
 git clone -b $(cat /tmp/git_head_ref) "https://github.com/$(cat /tmp/git_repository).git"
 cd ipld-eth-server
-mkdir -p out
 
 ## Remove the branch and github related info. This way future runs wont be confused.
 rm -f /tmp/git_head_ref /tmp/git_repository
 
-# Remove existing docker-tsdb directory
-rm -rf out/docker-tsdb/
-
-# Copy over files to setup TimescaleDB
-ID=$(docker create vulcanize/ipld-eth-db:v4.1.1-alpha)
-docker cp $ID:/app/docker-tsdb out/docker-tsdb/
-docker rm -v $ID
-
-# Spin up TimescaleDB
-docker-compose -f out/docker-tsdb/docker-compose.test.yml -f docker-compose.yml up ipld-eth-db
-trap "docker-compose -f out/docker-tsdb/docker-compose.test.yml -f docker-compose.yml down --remove-orphans --volumes; cd $start_dir ; rm -r $temp_dir" SIGINT SIGTERM ERR
-sleep 45
+# Spin up DB
+docker-compose up -d ipld-eth-db
+trap "docker-compose down --remove-orphans; cd $start_dir ; rm -r $temp_dir" SIGINT SIGTERM ERR
+sleep 10
 
 # Remove old logs so there's no confusion, then run test
 rm -f /tmp/test.log /tmp/return_test.txt
-PGPASSWORD=password DATABASE_USER=vdbm DATABASE_PORT=8066 DATABASE_PASSWORD=password DATABASE_HOSTNAME=127.0.0.1 DATABASE_NAME=vulcanize_testing_v4 make test >  /tmp/test.log
+PGPASSWORD=password DATABASE_USER=vdbm DATABASE_PORT=8077 DATABASE_PASSWORD=password DATABASE_HOSTNAME=127.0.0.1 DATABASE_NAME=vulcanize_testing make test >  /tmp/test.log
 echo $? > /tmp/return_test.txt
 
 # Clean up
-docker-compose -f out/docker-tsdb/docker-compose.test.yml -f docker-compose.yml down --remove-orphans --volumes
+docker-compose down -v --remove-orphans
 cd $start_dir
 rm -fr $temp_dir
+

cmd/serve.go

@@ -33,7 +33,6 @@ import (
 	"github.com/vulcanize/gap-filler/pkg/mux"
 
 	"github.com/vulcanize/ipld-eth-server/v3/pkg/eth"
-	fill "github.com/vulcanize/ipld-eth-server/v3/pkg/fill"
 	"github.com/vulcanize/ipld-eth-server/v3/pkg/graphql"
 	srpc "github.com/vulcanize/ipld-eth-server/v3/pkg/rpc"
 	s "github.com/vulcanize/ipld-eth-server/v3/pkg/serve"
@@ -101,25 +100,12 @@ func serve() {
 		logWithCommand.Info("state validator disabled")
 	}
 
-	var watchedAddressFillService *fill.Service
-	if serverConfig.WatchedAddressGapFillerEnabled {
-		watchedAddressFillService = fill.New(serverConfig)
-		wg.Add(1)
-		go watchedAddressFillService.Start(wg)
-		logWithCommand.Info("watched address gap filler enabled")
-	} else {
-		logWithCommand.Info("watched address gap filler disabled")
-	}
-
 	shutdown := make(chan os.Signal, 1)
 	signal.Notify(shutdown, os.Interrupt)
 	<-shutdown
 	if graphQL != nil {
 		graphQL.Stop()
 	}
-	if watchedAddressFillService != nil {
-		watchedAddressFillService.Stop()
-	}
 	server.Stop()
 	wg.Wait()
 }
@@ -374,10 +360,6 @@ func init() {
 	serveCmd.PersistentFlags().Bool("validator-enabled", false, "turn on the state validator")
 	serveCmd.PersistentFlags().Uint("validator-every-nth-block", 1500, "only validate every Nth block")
 
-	// watched address gap filler flags
-	serveCmd.PersistentFlags().Bool("watched-address-gap-filler-enabled", false, "turn on the watched address gap filler")
-	serveCmd.PersistentFlags().Int("watched-address-gap-filler-interval", 60, "watched address gap fill interval in secs")
-
 	// and their bindings
 	// eth graphql server
 	viper.BindPFlag("eth.server.graphql", serveCmd.PersistentFlags().Lookup("eth-server-graphql"))
@@ -426,8 +408,4 @@ func init() {
 	// state validator flags
 	viper.BindPFlag("validator.enabled", serveCmd.PersistentFlags().Lookup("validator-enabled"))
 	viper.BindPFlag("validator.everyNthBlock", serveCmd.PersistentFlags().Lookup("validator-every-nth-block"))
-
-	// watched address gap filler flags
-	viper.BindPFlag("watch.fill.enabled", serveCmd.PersistentFlags().Lookup("watched-address-gap-filler-enabled"))
-	viper.BindPFlag("watch.fill.interval", serveCmd.PersistentFlags().Lookup("watched-address-gap-filler-interval"))
 }

cmd/validate.go

@@ -22,8 +22,8 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
-	validator "github.com/vulcanize/eth-ipfs-state-validator/v4/pkg"
-	ipfsethdb "github.com/vulcanize/ipfs-ethdb/v4/postgres"
+	validator "github.com/vulcanize/eth-ipfs-state-validator/v3/pkg"
+	ipfsethdb "github.com/vulcanize/ipfs-ethdb/v3/postgres"
 
 	s "github.com/vulcanize/ipld-eth-server/v3/pkg/serve"
 )

docker-compose.yml

@@ -2,16 +2,17 @@ version: '3.2'
 
 services:
   ipld-eth-db:
-    restart: on-failure
-    depends_on:
-      - access-node
-    image: vulcanize/ipld-eth-db:v4.1.1-alpha
+    restart: always
+    image: vulcanize/ipld-eth-db:v3.2.0
     environment:
-      DATABASE_USER: "vdbm"
-      DATABASE_NAME: "vulcanize_testing_v4"
-      DATABASE_PASSWORD: "password"
-      DATABASE_HOSTNAME: "access-node"
-      DATABASE_PORT: 5432
+      POSTGRES_USER: "vdbm"
+      POSTGRES_DB: "vulcanize_testing"
+      POSTGRES_PASSWORD: "password"
+    volumes:
+     - vdb_db_eth_server:/var/lib/postgresql/data
+    ports:
+     - "127.0.0.1:8077:5432"
+    command: ["postgres", "-c", "log_statement=all"]
 
   eth-server:
     restart: unless-stopped
@@ -37,8 +38,6 @@ services:
       ETH_FORWARD_ETH_CALLS: $ETH_FORWARD_ETH_CALLS
       ETH_PROXY_ON_ERROR: $ETH_PROXY_ON_ERROR
       ETH_HTTP_PATH: $ETH_HTTP_PATH
-      WATCHED_ADDRESS_GAP_FILLER_ENABLED: $WATCHED_ADDRESS_GAP_FILLER_ENABLED
-      WATCHED_ADDRESS_GAP_FILLER_INTERVAL: $WATCHED_ADDRESS_GAP_FILLER_INTERVAL
     volumes:
     - type: bind
       source: ./chain.json

environments/example.toml

@@ -28,8 +28,3 @@
     clientName = "Geth" # $ETH_CLIENT_NAME
     genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3" # $ETH_GENESIS_BLOCK
     networkID = "1" # $ETH_NETWORK_ID
-
-[watch]
-    [watch.fill]
-        enabled = false
-        interval = 30

go.mod

@@ -23,9 +23,9 @@ require (
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/cobra v1.4.0
 	github.com/spf13/viper v1.11.0
-	github.com/vulcanize/eth-ipfs-state-validator/v4 v4.0.0-alpha
+	github.com/vulcanize/eth-ipfs-state-validator/v3 v3.0.0
 	github.com/vulcanize/gap-filler v0.3.1
-	github.com/vulcanize/ipfs-ethdb/v4 v4.0.0-alpha
+	github.com/vulcanize/ipfs-ethdb/v3 v3.0.1
 )
 
-replace github.com/ethereum/go-ethereum v1.10.17 => github.com/vulcanize/go-ethereum v1.10.17-statediff-4.0.1-alpha
+replace github.com/ethereum/go-ethereum v1.10.17 => github.com/vulcanize/go-ethereum v1.10.17-statediff-3.2.1

go.sum

@@ -1834,14 +1834,14 @@ github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPU
 github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
 github.com/viant/assertly v0.4.8/go.mod h1:aGifi++jvCrUaklKEKT0BU95igDNaqkvz+49uaYMPRU=
 github.com/viant/toolbox v0.24.0/go.mod h1:OxMCG57V0PXuIP2HNQrtJf2CjqdmbrOx5EkMILuUhzM=
-github.com/vulcanize/eth-ipfs-state-validator/v4 v4.0.0-alpha h1:9TtxGBNl/hVQ8wegrZx//UnAMivImbcNcYTQxHfhjlQ=
-github.com/vulcanize/eth-ipfs-state-validator/v4 v4.0.0-alpha/go.mod h1:cY+XmYdOtBc84s02GdDTFoybJF+mR7WDcr3Z7l4Fb1g=
+github.com/vulcanize/eth-ipfs-state-validator/v3 v3.0.0 h1:hQcbewsJX5Xpd3xdE9n3JpGyuzUqMhwhpQCpbN+9Uiw=
+github.com/vulcanize/eth-ipfs-state-validator/v3 v3.0.0/go.mod h1:CuuygZPciwwLV5vNY1LcIi0PZxXZbOmicw1/IkrPpPI=
 github.com/vulcanize/gap-filler v0.3.1 h1:N5d+jCJo/VTWFvBSbTD7biRhK/OqDZzi1tgA85SIBKs=
 github.com/vulcanize/gap-filler v0.3.1/go.mod h1:qowG1cgshVpBqMokiWro/1xhh0uypw7oAu8FQ42JMy4=
-github.com/vulcanize/go-ethereum v1.10.17-statediff-4.0.1-alpha h1:ANyTG2JJkOaQD3IqkLLl2OjtTC+PKlrdmy6HIddpqZw=
-github.com/vulcanize/go-ethereum v1.10.17-statediff-4.0.1-alpha/go.mod h1:mDwZX+QoWdqzQo6SDG3YVqCYACutcSG6uzpziMvTu28=
-github.com/vulcanize/ipfs-ethdb/v4 v4.0.0-alpha h1:XaltpDMGFqcEe+tsYTmqICeFhR+jTHHzc85YTVigEdE=
-github.com/vulcanize/ipfs-ethdb/v4 v4.0.0-alpha/go.mod h1:sSTqwalaV7CiXq83YkMJpZetoSxgcdzr0AG0NnuaGMM=
+github.com/vulcanize/go-ethereum v1.10.17-statediff-3.2.1 h1:BUIaiqqsWM61KfzIowqrGJ8TM2SWwmral4iJ/nhTSe0=
+github.com/vulcanize/go-ethereum v1.10.17-statediff-3.2.1/go.mod h1:mDwZX+QoWdqzQo6SDG3YVqCYACutcSG6uzpziMvTu28=
+github.com/vulcanize/ipfs-ethdb/v3 v3.0.1 h1:gm+6SgxcNOCz6mgwvhsxORLXsr9yIFvbJDuoAK85ONs=
+github.com/vulcanize/ipfs-ethdb/v3 v3.0.1/go.mod h1:SPBTTl5CqRexYfkI66pbzE5nQziYpoQ+7hH0XveA9IU=
 github.com/wangjia184/sortedset v0.0.0-20160527075905-f5d03557ba30/go.mod h1:YkocrP2K2tcw938x9gCOmT5G5eCD6jsTz0SZuyAqwIE=
 github.com/warpfork/go-testmark v0.3.0 h1:Q81c4u7hT+BR5kNfNQhEF0VT2pmL7+Kk0wD+ORYl7iA=
 github.com/warpfork/go-testmark v0.3.0/go.mod h1:jhEf8FVxd+F17juRubpmut64NEG6I2rgkUhlcqqXwE0=

pkg/eth/backend.go

@@ -43,8 +43,8 @@ import (
 	"github.com/ethereum/go-ethereum/trie"
 	"github.com/jmoiron/sqlx"
 	log "github.com/sirupsen/logrus"
-	validator "github.com/vulcanize/eth-ipfs-state-validator/v4/pkg"
-	ipfsethdb "github.com/vulcanize/ipfs-ethdb/v4/postgres"
+	validator "github.com/vulcanize/eth-ipfs-state-validator/v3/pkg"
+	ipfsethdb "github.com/vulcanize/ipfs-ethdb/v3/postgres"
 
 	ethServerShared "github.com/ethereum/go-ethereum/statediff/indexer/shared"
 
@@ -59,42 +59,31 @@ var (
 
 	// errMissingSignature is returned if a block's extra-data section doesn't seem
 	// to contain a 65 byte secp256k1 signature.
+	errMissingSignature = errors.New("extra-data 65 byte signature suffix missing")
 )
 
 const (
 	RetrieveCanonicalBlockHashByNumber = `SELECT block_hash FROM eth.header_cids
-									INNER JOIN public.blocks ON (
-										header_cids.mh_key = blocks.key
-										AND header_cids.block_number = blocks.block_number
-									)
+									INNER JOIN public.blocks ON (header_cids.mh_key = blocks.key)
 									WHERE block_hash = (SELECT canonical_header_hash($1))`
 	RetrieveCanonicalHeaderByNumber = `SELECT cid, data FROM eth.header_cids
-									INNER JOIN public.blocks ON (
-										header_cids.mh_key = blocks.key
-										AND header_cids.block_number = blocks.block_number
-									)
+									INNER JOIN public.blocks ON (header_cids.mh_key = blocks.key)
 									WHERE block_hash = (SELECT canonical_header_hash($1))`
 	RetrieveTD = `SELECT CAST(td as Text) FROM eth.header_cids
 			WHERE header_cids.block_hash = $1`
-	RetrieveRPCTransaction = `SELECT blocks.data, block_hash, transaction_cids.block_number, index
-			FROM public.blocks, eth.transaction_cids, eth.header_cids
+	RetrieveRPCTransaction = `SELECT blocks.data, block_hash, block_number, index FROM public.blocks, eth.transaction_cids, eth.header_cids
 			WHERE blocks.key = transaction_cids.mh_key
-			AND blocks.block_number = transaction_cids.block_number
 			AND transaction_cids.header_id = header_cids.block_hash
-			AND transaction_cids.block_number = header_cids.block_number
 			AND transaction_cids.tx_hash = $1`
 	RetrieveCodeHashByLeafKeyAndBlockHash = `SELECT code_hash FROM eth.state_accounts, eth.state_cids, eth.header_cids
-											WHERE state_accounts.header_id = state_cids.header_id
-											AND state_accounts.state_path = state_cids.state_path
-											AND state_accounts.block_number = state_cids.block_number
+											WHERE state_accounts.header_id = state_cids.header_id AND state_accounts.state_path = state_cids.state_path
 											AND state_cids.header_id = header_cids.block_hash
-											AND state_cids.block_number = header_cids.block_number
 											AND state_leaf_key = $1
-											AND header_cids.block_number <= (SELECT block_number
+											AND block_number <= (SELECT block_number
 																FROM eth.header_cids
 																WHERE block_hash = $2)
-											AND header_cids.block_hash = (SELECT canonical_header_hash(header_cids.block_number))
-											ORDER BY header_cids.block_number DESC
+											AND header_cids.block_hash = (SELECT canonical_header_hash(block_number))
+											ORDER BY block_number DESC
 											LIMIT 1`
 	RetrieveCodeByMhKey = `SELECT data FROM public.blocks WHERE key = $1`
 )

pkg/eth/cid_retriever.go

@@ -19,7 +19,6 @@ package eth
 import (
 	"fmt"
 	"math/big"
-	"strconv"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -60,7 +59,7 @@ func (ecr *CIDRetriever) RetrieveFirstBlockNumber() (int64, error) {
 // RetrieveLastBlockNumber is used to retrieve the latest block number in the db
 func (ecr *CIDRetriever) RetrieveLastBlockNumber() (int64, error) {
 	var blockNumber int64
-	err := ecr.db.Get(&blockNumber, "SELECT block_number FROM eth.header_cids ORDER BY block_number DESC LIMIT 1")
+	err := ecr.db.Get(&blockNumber, "SELECT block_number FROM eth.header_cids ORDER BY block_number DESC LIMIT 1 ")
 	return blockNumber, err
 }
 
@@ -168,9 +167,9 @@ func (ecr *CIDRetriever) Retrieve(filter SubscriptionSettings, blockNumber int64
 func (ecr *CIDRetriever) RetrieveHeaderCIDs(tx *sqlx.Tx, blockNumber int64) ([]models.HeaderModel, error) {
 	log.Debug("retrieving header cids for block ", blockNumber)
 	headers := make([]models.HeaderModel, 0)
-	pgStr := `SELECT CAST(block_number as Text), block_hash, parent_hash, cid, mh_key, CAST(td as Text), node_id,
-				CAST(reward as Text), state_root, uncle_root,tx_root, receipt_root,bloom, timestamp, times_validated, coinbase
-				FROM eth.header_cids
+	pgStr := `SELECT CAST(block_number as Text), block_hash,parent_hash,cid,mh_key,CAST(td as Text),node_id,
+				CAST(reward as Text), state_root,uncle_root,tx_root,receipt_root,bloom,timestamp,times_validated,
+				coinbase FROM eth.header_cids
 				WHERE block_number = $1`
 	return headers, tx.Select(&headers, pgStr, blockNumber)
 }
@@ -179,8 +178,7 @@ func (ecr *CIDRetriever) RetrieveHeaderCIDs(tx *sqlx.Tx, blockNumber int64) ([]m
 func (ecr *CIDRetriever) RetrieveUncleCIDsByHeaderID(tx *sqlx.Tx, headerID string) ([]models.UncleModel, error) {
 	log.Debug("retrieving uncle cids for block id ", headerID)
 	headers := make([]models.UncleModel, 0)
-	pgStr := `SELECT CAST(block_number as Text), header_id, block_hash, parent_hash, cid, mh_key, CAST(reward as text)
-				FROM eth.uncle_cids
+	pgStr := `SELECT header_id,block_hash,parent_hash,cid,mh_key, CAST(reward as text) FROM eth.uncle_cids
 				WHERE header_id = $1`
 	return headers, tx.Select(&headers, pgStr, headerID)
 }
@@ -192,15 +190,10 @@ func (ecr *CIDRetriever) RetrieveTxCIDs(tx *sqlx.Tx, txFilter TxFilter, headerID
 	args := make([]interface{}, 0, 3)
 	results := make([]models.TxModel, 0)
 	id := 1
-	pgStr := fmt.Sprintf(`SELECT CAST(transaction_cids.block_number as Text), transaction_cids.tx_hash,
-				transaction_cids.header_id,transaction_cids.cid, transaction_cids.mh_key, transaction_cids.dst,
-				transaction_cids.src, transaction_cids.index, transaction_cids.tx_data
-				FROM eth.transaction_cids
-				INNER JOIN eth.header_cids ON (
-						transaction_cids.header_id = header_cids.block_hash
-						AND transaction_cids.block_number = header_cids.block_number
-					)
-				WHERE header_cids.block_hash = $%d`, id)
+	pgStr := fmt.Sprintf(`SELECT transaction_cids.tx_hash, transaction_cids.header_id,transaction_cids.cid, transaction_cids.mh_key,
+				transaction_cids.dst, transaction_cids.src, transaction_cids.index, transaction_cids.tx_data
+				FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.block_hash)
+			WHERE header_cids.block_hash = $%d`, id)
 	args = append(args, headerID)
 	id++
 	if len(txFilter.Dst) > 0 {
@@ -299,13 +292,11 @@ func receiptFilterConditions(id *int, pgStr string, args []interface{}, rctFilte
 func (ecr *CIDRetriever) RetrieveRctCIDsByHeaderID(tx *sqlx.Tx, rctFilter ReceiptFilter, headerID string, trxHashes []string) ([]models.ReceiptModel, error) {
 	log.Debug("retrieving receipt cids for header id ", headerID)
 	args := make([]interface{}, 0, 4)
-	pgStr := `SELECT CAST(receipt_cids.block_number as Text), receipt_cids.tx_id, receipt_cids.leaf_cid,
-			receipt_cids.leaf_mh_key, receipt_cids.contract, receipt_cids.contract_hash
+	pgStr := `SELECT receipt_cids.tx_id, receipt_cids.leaf_cid, receipt_cids.leaf_mh_key,
+ 			receipt_cids.contract, receipt_cids.contract_hash
  			FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids
 			WHERE receipt_cids.tx_id = transaction_cids.tx_hash
-			AND receipt_cids.block_number = transaction_cids.block_number
 			AND transaction_cids.header_id = header_cids.block_hash
-			AND transaction_cids.block_number = header_cids.block_number
 			AND header_cids.block_hash = $1`
 	id := 2
 	args = append(args, headerID)
@@ -317,25 +308,20 @@ func (ecr *CIDRetriever) RetrieveRctCIDsByHeaderID(tx *sqlx.Tx, rctFilter Receip
 	return receiptCids, tx.Select(&receiptCids, pgStr, args...)
 }
 
-// RetrieveFilteredGQLLogs retrieves and returns all the log CIDs provided blockHash that conform to the provided
+// RetrieveFilteredGQLLogs retrieves and returns all the log cIDs provided blockHash that conform to the provided
 // filter parameters.
 func (ecr *CIDRetriever) RetrieveFilteredGQLLogs(tx *sqlx.Tx, rctFilter ReceiptFilter, blockHash *common.Hash) ([]LogResult, error) {
 	log.Debug("retrieving log cids for receipt ids")
 	args := make([]interface{}, 0, 4)
 	id := 1
-	pgStr := `SELECT CAST(eth.log_cids.block_number as Text), eth.log_cids.leaf_cid, eth.log_cids.index, eth.log_cids.rct_id,
+	pgStr := `SELECT eth.log_cids.leaf_cid, eth.log_cids.index, eth.log_cids.rct_id,
        			eth.log_cids.address, eth.log_cids.topic0, eth.log_cids.topic1, eth.log_cids.topic2, eth.log_cids.topic3,
        			eth.log_cids.log_data, eth.transaction_cids.tx_hash, data, eth.receipt_cids.leaf_cid as cid, eth.receipt_cids.post_status
 				FROM eth.log_cids, eth.receipt_cids, eth.transaction_cids, eth.header_cids, public.blocks
 				WHERE eth.log_cids.rct_id = receipt_cids.tx_id
-				AND eth.log_cids.block_number = eth.receipt_cids.block_number
 				AND receipt_cids.tx_id = transaction_cids.tx_hash
-				AND receipt_cids.block_number = transaction_cids.block_number
-				AND transaction_cids.header_id = header_cids.block_hash
-				AND transaction_cids.block_number = header_cids.block_number
- 				AND log_cids.leaf_mh_key = blocks.key
-				AND log_cids.block_number = blocks.block_number
-				AND header_cids.block_hash = $1`
+ 				AND transaction_cids.header_id = header_cids.block_hash
+ 				AND log_cids.leaf_mh_key = blocks.key AND header_cids.block_hash = $1`
 
 	args = append(args, blockHash.String())
 	id++
@@ -357,17 +343,14 @@ func (ecr *CIDRetriever) RetrieveFilteredGQLLogs(tx *sqlx.Tx, rctFilter ReceiptF
 func (ecr *CIDRetriever) RetrieveFilteredLog(tx *sqlx.Tx, rctFilter ReceiptFilter, blockNumber int64, blockHash *common.Hash) ([]LogResult, error) {
 	log.Debug("retrieving log cids for receipt ids")
 	args := make([]interface{}, 0, 4)
-	pgStr := `SELECT CAST(eth.log_cids.block_number as Text), eth.log_cids.leaf_cid, eth.log_cids.index, eth.log_cids.rct_id,
+	pgStr := `SELECT eth.log_cids.leaf_cid, eth.log_cids.index, eth.log_cids.rct_id,
        			eth.log_cids.address, eth.log_cids.topic0, eth.log_cids.topic1, eth.log_cids.topic2, eth.log_cids.topic3,
        			eth.log_cids.log_data, eth.transaction_cids.tx_hash, eth.transaction_cids.index as txn_index,
        			header_cids.block_hash, CAST(header_cids.block_number as Text)
 							FROM eth.log_cids, eth.receipt_cids, eth.transaction_cids, eth.header_cids
 							WHERE eth.log_cids.rct_id = receipt_cids.tx_id
-							AND eth.log_cids.block_number = eth.receipt_cids.block_number
 							AND receipt_cids.tx_id = transaction_cids.tx_hash
-							AND receipt_cids.block_number = transaction_cids.block_number
-							AND transaction_cids.header_id = header_cids.block_hash
-							AND transaction_cids.block_number = header_cids.block_number`
+							AND transaction_cids.header_id = header_cids.block_hash`
 	id := 1
 	if blockNumber > 0 {
 		pgStr += fmt.Sprintf(` AND header_cids.block_number = $%d`, id)
@@ -397,13 +380,10 @@ func (ecr *CIDRetriever) RetrieveFilteredLog(tx *sqlx.Tx, rctFilter ReceiptFilte
 func (ecr *CIDRetriever) RetrieveRctCIDs(tx *sqlx.Tx, rctFilter ReceiptFilter, blockNumber int64, blockHash *common.Hash, txHashes []string) ([]models.ReceiptModel, error) {
 	log.Debug("retrieving receipt cids for block ", blockNumber)
 	args := make([]interface{}, 0, 5)
-	pgStr := `SELECT CAST(receipt_cids.block_number as Text), receipt_cids.tx_id, receipt_cids.leaf_cid,
-			receipt_cids.leaf_mh_key,
+	pgStr := `SELECT receipt_cids.tx_id, receipt_cids.leaf_cid, receipt_cids.leaf_mh_key, receipt_cids.tx_id
  			FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids
 			WHERE receipt_cids.tx_id = transaction_cids.tx_hash
-			AND receipt_cids.block_number = transaction_cids.block_number
-			AND transaction_cids.header_id = header_cids.block_hash
-			AND transaction_cids.block_number = header_cids.block_number`
+			AND transaction_cids.header_id = header_cids.block_hash`
 	id := 1
 	if blockNumber > 0 {
 		pgStr += fmt.Sprintf(` AND header_cids.block_number = $%d`, id)
@@ -436,13 +416,9 @@ func hasTopics(topics [][]string) bool {
 func (ecr *CIDRetriever) RetrieveStateCIDs(tx *sqlx.Tx, stateFilter StateFilter, headerID string) ([]models.StateNodeModel, error) {
 	log.Debug("retrieving state cids for header id ", headerID)
 	args := make([]interface{}, 0, 2)
-	pgStr := `SELECT CAST(state_cids.block_number as Text), state_cids.header_id,
+	pgStr := `SELECT state_cids.header_id,
 			state_cids.state_leaf_key, state_cids.node_type, state_cids.cid, state_cids.mh_key, state_cids.state_path
-			FROM eth.state_cids
-			INNER JOIN eth.header_cids ON (
-				state_cids.header_id = header_cids.block_hash
-				AND state_cids.block_number = header_cids.block_number
-			)
+			FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
 			WHERE header_cids.block_hash = $1`
 	args = append(args, headerID)
 	addrLen := len(stateFilter.Addresses)
@@ -465,15 +441,11 @@ func (ecr *CIDRetriever) RetrieveStateCIDs(tx *sqlx.Tx, stateFilter StateFilter,
 func (ecr *CIDRetriever) RetrieveStorageCIDs(tx *sqlx.Tx, storageFilter StorageFilter, headerID string) ([]models.StorageNodeWithStateKeyModel, error) {
 	log.Debug("retrieving storage cids for header id ", headerID)
 	args := make([]interface{}, 0, 3)
-	pgStr := `SELECT CAST(storage_cids.block_number as Text), storage_cids.header_id, storage_cids.storage_leaf_key,
-			storage_cids.node_type, storage_cids.cid, storage_cids.mh_key, storage_cids.storage_path, storage_cids.state_path,
-			state_cids.state_leaf_key
+	pgStr := `SELECT storage_cids.header_id, storage_cids.storage_leaf_key, storage_cids.node_type,
+ 			storage_cids.cid, storage_cids.mh_key, storage_cids.storage_path, storage_cids.state_path, state_cids.state_leaf_key
  			FROM eth.storage_cids, eth.state_cids, eth.header_cids
-			WHERE storage_cids.header_id = state_cids.header_id
-			AND storage_cids.state_path = state_cids.state_path
-			AND storage_cids.block_number = state_cids.block_number
+			WHERE storage_cids.header_id = state_cids.header_id AND storage_cids.state_path = state_cids.state_path
 			AND state_cids.header_id = header_cids.block_hash
-			AND state_cids.block_number = header_cids.block_number
 			AND header_cids.block_hash = $1`
 	args = append(args, headerID)
 	id := 2
@@ -524,10 +496,6 @@ func (ecr *CIDRetriever) RetrieveBlockByHash(blockHash common.Hash) (models.Head
 		log.Error("header cid retrieval error")
 		return models.HeaderModel{}, nil, nil, nil, err
 	}
-	blockNumber, err := strconv.ParseInt(headerCID.BlockNumber, 10, 64)
-	if err != nil {
-		return models.HeaderModel{}, nil, nil, nil, err
-	}
 	var uncleCIDs []models.UncleModel
 	uncleCIDs, err = ecr.RetrieveUncleCIDsByHeaderID(tx, headerCID.BlockHash)
 	if err != nil {
@@ -535,7 +503,7 @@ func (ecr *CIDRetriever) RetrieveBlockByHash(blockHash common.Hash) (models.Head
 		return models.HeaderModel{}, nil, nil, nil, err
 	}
 	var txCIDs []models.TxModel
-	txCIDs, err = ecr.RetrieveTxCIDsByHeaderID(tx, headerCID.BlockHash, blockNumber)
+	txCIDs, err = ecr.RetrieveTxCIDsByHeaderID(tx, headerCID.BlockHash)
 	if err != nil {
 		log.Error("tx cid retrieval error")
 		return models.HeaderModel{}, nil, nil, nil, err
@@ -588,7 +556,7 @@ func (ecr *CIDRetriever) RetrieveBlockByNumber(blockNumber int64) (models.Header
 		return models.HeaderModel{}, nil, nil, nil, err
 	}
 	var txCIDs []models.TxModel
-	txCIDs, err = ecr.RetrieveTxCIDsByHeaderID(tx, headerCID[0].BlockHash, blockNumber)
+	txCIDs, err = ecr.RetrieveTxCIDsByHeaderID(tx, headerCID[0].BlockHash)
 	if err != nil {
 		log.Error("tx cid retrieval error")
 		return models.HeaderModel{}, nil, nil, nil, err
@@ -608,33 +576,30 @@ func (ecr *CIDRetriever) RetrieveBlockByNumber(blockNumber int64) (models.Header
 // RetrieveHeaderCIDByHash returns the header for the given block hash
 func (ecr *CIDRetriever) RetrieveHeaderCIDByHash(tx *sqlx.Tx, blockHash common.Hash) (models.HeaderModel, error) {
 	log.Debug("retrieving header cids for block hash ", blockHash.String())
-	pgStr := `SELECT block_hash, CAST(block_number as Text), cid, mh_key FROM eth.header_cids
+	pgStr := `SELECT block_hash,cid,mh_key FROM eth.header_cids
 			WHERE block_hash = $1`
 	var headerCID models.HeaderModel
 	return headerCID, tx.Get(&headerCID, pgStr, blockHash.String())
 }
 
 // RetrieveTxCIDsByHeaderID retrieves all tx CIDs for the given header id
-func (ecr *CIDRetriever) RetrieveTxCIDsByHeaderID(tx *sqlx.Tx, headerID string, blockNumber int64) ([]models.TxModel, error) {
+func (ecr *CIDRetriever) RetrieveTxCIDsByHeaderID(tx *sqlx.Tx, headerID string) ([]models.TxModel, error) {
 	log.Debug("retrieving tx cids for block id ", headerID)
-	pgStr := `SELECT CAST(block_number as Text), header_id, index, tx_hash, cid, mh_key,
-			dst, src, tx_data, tx_type, value
-			FROM eth.transaction_cids
-			WHERE header_id = $1 AND block_number = $2
+	pgStr := `SELECT * FROM eth.transaction_cids
+			WHERE header_id = $1
 			ORDER BY index`
 	var txCIDs []models.TxModel
-	return txCIDs, tx.Select(&txCIDs, pgStr, headerID, blockNumber)
+	return txCIDs, tx.Select(&txCIDs, pgStr, headerID)
 }
 
 // RetrieveReceiptCIDsByTxIDs retrieves receipt CIDs by their associated tx IDs
 func (ecr *CIDRetriever) RetrieveReceiptCIDsByTxIDs(tx *sqlx.Tx, txHashes []string) ([]models.ReceiptModel, error) {
 	log.Debugf("retrieving receipt cids for tx hashes %v", txHashes)
-	pgStr := `SELECT CAST(receipt_cids.block_number as Text), receipt_cids.tx_id, receipt_cids.leaf_cid, receipt_cids.leaf_mh_key,
+	pgStr := `SELECT receipt_cids.tx_id, receipt_cids.leaf_cid, receipt_cids.leaf_mh_key,
  			receipt_cids.contract, receipt_cids.contract_hash
 			FROM eth.receipt_cids, eth.transaction_cids
 			WHERE tx_id = ANY($1)
 			AND receipt_cids.tx_id = transaction_cids.tx_hash
-			AND receipt_cids.block_number = transaction_cids.block_number
 			ORDER BY transaction_cids.index`
 	var rctCIDs []models.ReceiptModel
 	return rctCIDs, tx.Select(&rctCIDs, pgStr, pq.Array(txHashes))

pkg/eth/cid_retriever_test.go

@@ -249,7 +249,6 @@ var _ = Describe("Retriever", func() {
 				AND header_cids.block_number = $1
 				ORDER BY transaction_cids.index`
 			err := db.Select(&expectedRctCIDsAndLeafNodes, pgStr, test_helpers.BlockNumber.Uint64())
-			Expect(err).ToNot(HaveOccurred())
 			cids, empty, err := retriever.Retrieve(openFilter, 1)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(empty).ToNot(BeTrue())
@@ -413,13 +412,12 @@ var _ = Describe("Retriever", func() {
 			Expect(len(cids7[0].StorageNodes)).To(Equal(0))
 			Expect(len(cids7[0].StateNodes)).To(Equal(1))
 			Expect(cids7[0].StateNodes[0]).To(Equal(models.StateNodeModel{
-				BlockNumber: "1",
-				HeaderID:    cids7[0].StateNodes[0].HeaderID,
-				NodeType:    2,
-				StateKey:    common.BytesToHash(test_helpers.AccountLeafKey).Hex(),
-				CID:         test_helpers.State2CID.String(),
-				MhKey:       test_helpers.State2MhKey,
-				Path:        []byte{'\x0c'},
+				HeaderID: cids7[0].StateNodes[0].HeaderID,
+				NodeType: 2,
+				StateKey: common.BytesToHash(test_helpers.AccountLeafKey).Hex(),
+				CID:      test_helpers.State2CID.String(),
+				MhKey:    test_helpers.State2MhKey,
+				Path:     []byte{'\x0c'},
 			}))
 
 			_, empty, err = retriever.Retrieve(rctTopicsAndAddressFilterFail, 1)

pkg/eth/filterer.go

@@ -82,9 +82,8 @@ func (s *ResponseFilterer) filterHeaders(headerFilter HeaderFilter, response *IP
 			return err
 		}
 		response.Header = models.IPLDModel{
-			BlockNumber: payload.Block.Number().String(),
-			Data:        headerRLP,
-			Key:         cid.String(),
+			Data: headerRLP,
+			Key:  cid.String(),
 		}
 		if headerFilter.Uncles {
 			response.Uncles = make([]models.IPLDModel, len(payload.Block.Body().Uncles))
@@ -98,9 +97,8 @@ func (s *ResponseFilterer) filterHeaders(headerFilter HeaderFilter, response *IP
 					return err
 				}
 				response.Uncles[i] = models.IPLDModel{
-					BlockNumber: uncle.Number.String(),
-					Data:        uncleRlp,
-					Key:         cid.String(),
+					Data: uncleRlp,
+					Key:  cid.String(),
 				}
 			}
 		}
@@ -185,9 +183,8 @@ func (s *ResponseFilterer) filerReceipts(receiptFilter ReceiptFilter, response *
 			// TODO: Verify this filter logic.
 			if checkReceipts(receipt, receiptFilter.Topics, topics, receiptFilter.LogAddresses, contracts, trxHashes) {
 				response.Receipts = append(response.Receipts, models.IPLDModel{
-					BlockNumber: payload.Block.Number().String(),
-					Data:        rctIPLDData[idx],
-					Key:         rctLeafCID[idx].String(),
+					Data: rctIPLDData[idx],
+					Key:  rctLeafCID[idx].String(),
 				})
 			}
 		}
@@ -285,9 +282,8 @@ func (s *ResponseFilterer) filterStateAndStorage(stateFilter StateFilter, storag
 					StateLeafKey: common.BytesToHash(stateNode.LeafKey),
 					Path:         stateNode.Path,
 					IPLD: models.IPLDModel{
-						BlockNumber: payload.Block.Number().String(),
-						Data:        stateNode.NodeValue,
-						Key:         cid.String(),
+						Data: stateNode.NodeValue,
+						Key:  cid.String(),
 					},
 					Type: stateNode.NodeType,
 				})
@@ -304,9 +300,8 @@ func (s *ResponseFilterer) filterStateAndStorage(stateFilter StateFilter, storag
 						StateLeafKey:   common.BytesToHash(stateNode.LeafKey),
 						StorageLeafKey: common.BytesToHash(storageNode.LeafKey),
 						IPLD: models.IPLDModel{
-							BlockNumber: payload.Block.Number().String(),
-							Data:        storageNode.NodeValue,
-							Key:         cid.String(),
+							Data: storageNode.NodeValue,
+							Key:  cid.String(),
 						},
 						Type: storageNode.NodeType,
 						Path: storageNode.Path,

pkg/eth/filterer_test.go

@@ -61,16 +61,14 @@ var _ = Describe("Filterer", func() {
 				Expect(stateNode.Type).To(Equal(sdtypes.Leaf))
 				if bytes.Equal(stateNode.StateLeafKey.Bytes(), test_helpers.AccountLeafKey) {
 					Expect(stateNode.IPLD).To(Equal(models.IPLDModel{
-						BlockNumber: test_helpers.BlockNumber.String(),
-						Data:        test_helpers.State2IPLD.RawData(),
-						Key:         test_helpers.State2IPLD.Cid().String(),
+						Data: test_helpers.State2IPLD.RawData(),
+						Key:  test_helpers.State2IPLD.Cid().String(),
 					}))
 				}
 				if bytes.Equal(stateNode.StateLeafKey.Bytes(), test_helpers.ContractLeafKey) {
 					Expect(stateNode.IPLD).To(Equal(models.IPLDModel{
-						BlockNumber: test_helpers.BlockNumber.String(),
-						Data:        test_helpers.State1IPLD.RawData(),
-						Key:         test_helpers.State1IPLD.Cid().String(),
+						Data: test_helpers.State1IPLD.RawData(),
+						Key:  test_helpers.State1IPLD.Cid().String(),
 					}))
 				}
 			}
@@ -89,9 +87,8 @@ var _ = Describe("Filterer", func() {
 			Expect(len(iplds1.StateNodes)).To(Equal(0))
 			Expect(len(iplds1.Receipts)).To(Equal(1))
 			Expect(iplds1.Receipts[0]).To(Equal(models.IPLDModel{
-				BlockNumber: test_helpers.BlockNumber.String(),
-				Data:        test_helpers.Rct1IPLD,
-				Key:         test_helpers.Rct1CID.String(),
+				Data: test_helpers.Rct1IPLD,
+				Key:  test_helpers.Rct1CID.String(),
 			}))
 
 			iplds2, err := filterer.Filter(rctTopicsFilter, test_helpers.MockConvertedPayload)
@@ -105,9 +102,8 @@ var _ = Describe("Filterer", func() {
 			Expect(len(iplds2.StateNodes)).To(Equal(0))
 			Expect(len(iplds2.Receipts)).To(Equal(1))
 			Expect(iplds2.Receipts[0]).To(Equal(models.IPLDModel{
-				BlockNumber: test_helpers.BlockNumber.String(),
-				Data:        test_helpers.Rct1IPLD,
-				Key:         test_helpers.Rct1CID.String(),
+				Data: test_helpers.Rct1IPLD,
+				Key:  test_helpers.Rct1CID.String(),
 			}))
 
 			iplds3, err := filterer.Filter(rctTopicsAndAddressFilter, test_helpers.MockConvertedPayload)
@@ -121,9 +117,8 @@ var _ = Describe("Filterer", func() {
 			Expect(len(iplds3.StateNodes)).To(Equal(0))
 			Expect(len(iplds3.Receipts)).To(Equal(1))
 			Expect(iplds3.Receipts[0]).To(Equal(models.IPLDModel{
-				BlockNumber: test_helpers.BlockNumber.String(),
-				Data:        test_helpers.Rct1IPLD,
-				Key:         test_helpers.Rct1CID.String(),
+				Data: test_helpers.Rct1IPLD,
+				Key:  test_helpers.Rct1CID.String(),
 			}))
 
 			iplds4, err := filterer.Filter(rctAddressesAndTopicFilter, test_helpers.MockConvertedPayload)
@@ -137,9 +132,8 @@ var _ = Describe("Filterer", func() {
 			Expect(len(iplds4.StateNodes)).To(Equal(0))
 			Expect(len(iplds4.Receipts)).To(Equal(1))
 			Expect(iplds4.Receipts[0]).To(Equal(models.IPLDModel{
-				BlockNumber: test_helpers.BlockNumber.String(),
-				Data:        test_helpers.Rct2IPLD,
-				Key:         test_helpers.Rct2CID.String(),
+				Data: test_helpers.Rct2IPLD,
+				Key:  test_helpers.Rct2CID.String(),
 			}))
 
 			iplds5, err := filterer.Filter(rctsForAllCollectedTrxs, test_helpers.MockConvertedPayload)
@@ -171,9 +165,8 @@ var _ = Describe("Filterer", func() {
 			Expect(len(iplds6.StateNodes)).To(Equal(0))
 			Expect(len(iplds6.Receipts)).To(Equal(1))
 			Expect(iplds4.Receipts[0]).To(Equal(models.IPLDModel{
-				BlockNumber: test_helpers.BlockNumber.String(),
-				Data:        test_helpers.Rct2IPLD,
-				Key:         test_helpers.Rct2CID.String(),
+				Data: test_helpers.Rct2IPLD,
+				Key:  test_helpers.Rct2CID.String(),
 			}))
 
 			iplds7, err := filterer.Filter(stateFilter, test_helpers.MockConvertedPayload)
@@ -188,9 +181,8 @@ var _ = Describe("Filterer", func() {
 			Expect(len(iplds7.StateNodes)).To(Equal(1))
 			Expect(iplds7.StateNodes[0].StateLeafKey.Bytes()).To(Equal(test_helpers.AccountLeafKey))
 			Expect(iplds7.StateNodes[0].IPLD).To(Equal(models.IPLDModel{
-				BlockNumber: test_helpers.BlockNumber.String(),
-				Data:        test_helpers.State2IPLD.RawData(),
-				Key:         test_helpers.State2IPLD.Cid().String(),
+				Data: test_helpers.State2IPLD.RawData(),
+				Key:  test_helpers.State2IPLD.Cid().String(),
 			}))
 
 			iplds8, err := filterer.Filter(rctTopicsAndAddressFilterFail, test_helpers.MockConvertedPayload)

pkg/eth/ipld_fetcher.go

@@ -20,7 +20,6 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
-	"strconv"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/statediff/indexer/models"
@@ -103,19 +102,13 @@ func (f *IPLDFetcher) Fetch(cids CIDWrapper) (*IPLDs, error) {
 // FetchHeaders fetches headers
 func (f *IPLDFetcher) FetchHeader(tx *sqlx.Tx, c models.HeaderModel) (models.IPLDModel, error) {
 	log.Debug("fetching header ipld")
-	blockNumber, err := strconv.ParseUint(c.BlockNumber, 10, 64)
-	if err != nil {
-		return models.IPLDModel{}, err
-	}
-
-	headerBytes, err := shared.FetchIPLDByMhKeyAndBlockNumber(tx, c.MhKey, blockNumber)
+	headerBytes, err := shared.FetchIPLDByMhKey(tx, c.MhKey)
 	if err != nil {
 		return models.IPLDModel{}, err
 	}
 	return models.IPLDModel{
-		BlockNumber: c.BlockNumber,
-		Data:        headerBytes,
-		Key:         c.CID,
+		Data: headerBytes,
+		Key:  c.CID,
 	}, nil
 }
 
@@ -124,18 +117,13 @@ func (f *IPLDFetcher) FetchUncles(tx *sqlx.Tx, cids []models.UncleModel) ([]mode
 	log.Debug("fetching uncle iplds")
 	uncleIPLDs := make([]models.IPLDModel, len(cids))
 	for i, c := range cids {
-		blockNumber, err := strconv.ParseUint(c.BlockNumber, 10, 64)
-		if err != nil {
-			return nil, err
-		}
-		uncleBytes, err := shared.FetchIPLDByMhKeyAndBlockNumber(tx, c.MhKey, blockNumber)
+		uncleBytes, err := shared.FetchIPLDByMhKey(tx, c.MhKey)
 		if err != nil {
 			return nil, err
 		}
 		uncleIPLDs[i] = models.IPLDModel{
-			BlockNumber: c.BlockNumber,
-			Data:        uncleBytes,
-			Key:         c.CID,
+			Data: uncleBytes,
+			Key:  c.CID,
 		}
 	}
 	return uncleIPLDs, nil
@@ -146,18 +134,13 @@ func (f *IPLDFetcher) FetchTrxs(tx *sqlx.Tx, cids []models.TxModel) ([]models.IP
 	log.Debug("fetching transaction iplds")
 	trxIPLDs := make([]models.IPLDModel, len(cids))
 	for i, c := range cids {
-		blockNumber, err := strconv.ParseUint(c.BlockNumber, 10, 64)
-		if err != nil {
-			return nil, err
-		}
-		txBytes, err := shared.FetchIPLDByMhKeyAndBlockNumber(tx, c.MhKey, blockNumber)
+		txBytes, err := shared.FetchIPLDByMhKey(tx, c.MhKey)
 		if err != nil {
 			return nil, err
 		}
 		trxIPLDs[i] = models.IPLDModel{
-			BlockNumber: c.BlockNumber,
-			Data:        txBytes,
-			Key:         c.CID,
+			Data: txBytes,
+			Key:  c.CID,
 		}
 	}
 	return trxIPLDs, nil
@@ -168,19 +151,14 @@ func (f *IPLDFetcher) FetchRcts(tx *sqlx.Tx, cids []models.ReceiptModel) ([]mode
 	log.Debug("fetching receipt iplds")
 	rctIPLDs := make([]models.IPLDModel, len(cids))
 	for i, c := range cids {
-		blockNumber, err := strconv.ParseUint(c.BlockNumber, 10, 64)
-		if err != nil {
-			return nil, err
-		}
-		rctBytes, err := shared.FetchIPLDByMhKeyAndBlockNumber(tx, c.LeafMhKey, blockNumber)
+		rctBytes, err := shared.FetchIPLDByMhKey(tx, c.LeafMhKey)
 		if err != nil {
 			return nil, err
 		}
 		//nodeVal, err := DecodeLeafNode(rctBytes)
 		rctIPLDs[i] = models.IPLDModel{
-			BlockNumber: c.BlockNumber,
-			Data:        rctBytes,
-			Key:         c.LeafCID,
+			Data: rctBytes,
+			Key:  c.LeafCID,
 		}
 	}
 	return rctIPLDs, nil
@@ -194,19 +172,14 @@ func (f *IPLDFetcher) FetchState(tx *sqlx.Tx, cids []models.StateNodeModel) ([]S
 		if stateNode.CID == "" {
 			continue
 		}
-		blockNumber, err := strconv.ParseUint(stateNode.BlockNumber, 10, 64)
-		if err != nil {
-			return nil, err
-		}
-		stateBytes, err := shared.FetchIPLDByMhKeyAndBlockNumber(tx, stateNode.MhKey, blockNumber)
+		stateBytes, err := shared.FetchIPLDByMhKey(tx, stateNode.MhKey)
 		if err != nil {
 			return nil, err
 		}
 		stateNodes = append(stateNodes, StateNode{
 			IPLD: models.IPLDModel{
-				BlockNumber: stateNode.BlockNumber,
-				Data:        stateBytes,
-				Key:         stateNode.CID,
+				Data: stateBytes,
+				Key:  stateNode.CID,
 			},
 			StateLeafKey: common.HexToHash(stateNode.StateKey),
 			Type:         ResolveToNodeType(stateNode.NodeType),
@@ -224,19 +197,14 @@ func (f *IPLDFetcher) FetchStorage(tx *sqlx.Tx, cids []models.StorageNodeWithSta
 		if storageNode.CID == "" || storageNode.StateKey == "" {
 			continue
 		}
-		blockNumber, err := strconv.ParseUint(storageNode.BlockNumber, 10, 64)
-		if err != nil {
-			return nil, err
-		}
-		storageBytes, err := shared.FetchIPLDByMhKeyAndBlockNumber(tx, storageNode.MhKey, blockNumber)
+		storageBytes, err := shared.FetchIPLDByMhKey(tx, storageNode.MhKey)
 		if err != nil {
 			return nil, err
 		}
 		storageNodes = append(storageNodes, StorageNode{
 			IPLD: models.IPLDModel{
-				BlockNumber: storageNode.BlockNumber,
-				Data:        storageBytes,
-				Key:         storageNode.CID,
+				Data: storageBytes,
+				Key:  storageNode.CID,
 			},
 			StateLeafKey:   common.HexToHash(storageNode.StateKey),
 			StorageLeafKey: common.HexToHash(storageNode.StorageKey),

pkg/eth/ipld_retriever.go

@@ -36,180 +36,93 @@ const (
 
 	RetrieveHeadersByHashesPgStr = `SELECT cid, data
 								FROM eth.header_cids
-									INNER JOIN public.blocks ON (
-										header_cids.mh_key = blocks.key
-										AND header_cids.block_number = blocks.block_number
-									)
+									INNER JOIN public.blocks ON (header_cids.mh_key = blocks.key)
 								WHERE block_hash = ANY($1::VARCHAR(66)[])`
 	RetrieveHeadersByBlockNumberPgStr = `SELECT cid, data
 								FROM eth.header_cids
-									INNER JOIN public.blocks ON (
-										header_cids.mh_key = blocks.key
-										AND header_cids.block_number = blocks.block_number
-									)
+									INNER JOIN public.blocks ON (header_cids.mh_key = blocks.key)
 								WHERE block_number = $1`
 	RetrieveHeaderByHashPgStr = `SELECT cid, data
 								FROM eth.header_cids
-									INNER JOIN public.blocks ON (
-										header_cids.mh_key = blocks.key
-										AND header_cids.block_number = blocks.block_number
-									)
+									INNER JOIN public.blocks ON (header_cids.mh_key = blocks.key)
 								WHERE block_hash = $1`
 	RetrieveUnclesByHashesPgStr = `SELECT cid, data
 								FROM eth.uncle_cids
-									INNER JOIN public.blocks ON (
-										uncle_cids.mh_key = blocks.key
-										AND uncle_cids.block_number = blocks.block_number
-									)
+									INNER JOIN public.blocks ON (uncle_cids.mh_key = blocks.key)
 								WHERE block_hash = ANY($1::VARCHAR(66)[])`
 	RetrieveUnclesByBlockHashPgStr = `SELECT uncle_cids.cid, data
 										FROM eth.uncle_cids
-											INNER JOIN eth.header_cids ON (
-												uncle_cids.header_id = header_cids.block_hash
-												AND uncle_cids.block_number = header_cids.block_number
-											)
-											INNER JOIN public.blocks ON (
-												uncle_cids.mh_key = blocks.key
-												AND uncle_cids.block_number = blocks.block_number
-											)
+											INNER JOIN eth.header_cids ON (uncle_cids.header_id = header_cids.block_hash)
+											INNER JOIN public.blocks ON (uncle_cids.mh_key = blocks.key)
 										WHERE block_hash = $1`
 	RetrieveUnclesByBlockNumberPgStr = `SELECT uncle_cids.cid, data
 										FROM eth.uncle_cids
-											INNER JOIN eth.header_cids ON (
-												uncle_cids.header_id = header_cids.block_hash
-												AND uncle_cids.block_number = header_cids.block_number
-											)
-											INNER JOIN public.blocks ON (
-												uncle_cids.mh_key = blocks.key
-												AND uncle_cids.block_number = blocks.block_number
-											)
+											INNER JOIN eth.header_cids ON (uncle_cids.header_id = header_cids.block_hash)
+											INNER JOIN public.blocks ON (uncle_cids.mh_key = blocks.key)
 										WHERE block_number = $1`
 	RetrieveUncleByHashPgStr = `SELECT cid, data
 								FROM eth.uncle_cids
-									INNER JOIN public.blocks ON (
-										uncle_cids.mh_key = blocks.key
-										AND uncle_cids.block_number = blocks.block_number
-									)
+									INNER JOIN public.blocks ON (uncle_cids.mh_key = blocks.key)
 								WHERE block_hash = $1`
 	RetrieveTransactionsByHashesPgStr = `SELECT cid, data
 									FROM eth.transaction_cids
-										INNER JOIN public.blocks ON (
-											transaction_cids.mh_key = blocks.key
-											AND transaction_cids.block_number = blocks.block_number
-										)
+										INNER JOIN public.blocks ON (transaction_cids.mh_key = blocks.key)
 									WHERE tx_hash = ANY($1::VARCHAR(66)[])`
 	RetrieveTransactionsByBlockHashPgStr = `SELECT transaction_cids.cid, data
 											FROM eth.transaction_cids
-												INNER JOIN eth.header_cids ON (
-													transaction_cids.header_id = header_cids.block_hash
-													AND transaction_cids.block_number = header_cids.block_number
-												)
-												INNER JOIN public.blocks ON (
-													transaction_cids.mh_key = blocks.key
-													AND transaction_cids.block_number = blocks.block_number
-												)
+												INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.block_hash)
+												INNER JOIN public.blocks ON (transaction_cids.mh_key = blocks.key)
 											WHERE block_hash = $1
 											ORDER BY eth.transaction_cids.index ASC`
 	RetrieveTransactionsByBlockNumberPgStr = `SELECT transaction_cids.cid, data
 											FROM eth.transaction_cids
-												INNER JOIN eth.header_cids ON (
-													transaction_cids.header_id = header_cids.block_hash
-													AND transaction_cids.block_number = header_cids.block_number
-												)
-												INNER JOIN public.blocks ON (
-													transaction_cids.mh_key = blocks.key
-													AND transaction_cids.block_number = blocks.block_number
-												)
+												INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.block_hash)
+												INNER JOIN public.blocks ON (transaction_cids.mh_key = blocks.key)
 											WHERE block_number = $1
 											ORDER BY eth.transaction_cids.index ASC`
 	RetrieveTransactionByHashPgStr = `SELECT cid, data
 									FROM eth.transaction_cids
-										INNER JOIN public.blocks ON (
-											transaction_cids.mh_key = blocks.key
-											AND transaction_cids.block_number = blocks.block_number
-										)
+										INNER JOIN public.blocks ON (transaction_cids.mh_key = blocks.key)
 									WHERE tx_hash = $1`
 	RetrieveReceiptsByTxHashesPgStr = `SELECT receipt_cids.leaf_cid, data
 									FROM eth.receipt_cids
-										INNER JOIN eth.transaction_cids ON (
-											receipt_cids.tx_id = transaction_cids.tx_hash
-											AND receipt_cids.block_number = transaction_cids.block_number
-										)
-										INNER JOIN public.blocks ON (
-											receipt_cids.leaf_mh_key = blocks.key
-											AND receipt_cids.block_number = blocks.block_number
-										)
+										INNER JOIN eth.transaction_cids ON (receipt_cids.tx_id = transaction_cids.tx_hash)
+										INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = blocks.key)
 									WHERE tx_hash = ANY($1::VARCHAR(66)[])`
 	RetrieveReceiptsByBlockHashPgStr = `SELECT receipt_cids.leaf_cid, data, eth.transaction_cids.tx_hash
 										FROM eth.receipt_cids
-											INNER JOIN eth.transaction_cids ON (
-												receipt_cids.tx_id = transaction_cids.tx_hash
-												AND receipt_cids.block_number = transaction_cids.block_number
-											)
-											INNER JOIN eth.header_cids ON (
-												transaction_cids.header_id = header_cids.block_hash
-												AND transaction_cids.block_number = header_cids.block_number
-											)
-											INNER JOIN public.blocks ON (
-												receipt_cids.leaf_mh_key = blocks.key
-												AND receipt_cids.block_number = blocks.block_number
-											)
+											INNER JOIN eth.transaction_cids ON (receipt_cids.tx_id = transaction_cids.tx_hash)
+											INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.block_hash)
+											INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = blocks.key)
 										WHERE block_hash = $1
 										ORDER BY eth.transaction_cids.index ASC`
 	RetrieveReceiptsByBlockNumberPgStr = `SELECT receipt_cids.leaf_cid, data
 										FROM eth.receipt_cids
-											INNER JOIN eth.transaction_cids ON (
-												receipt_cids.tx_id = transaction_cids.tx_hash
-												AND receipt_cids.block_number = transaction_cids.block_number
-											)
-											INNER JOIN eth.header_cids ON (
-												transaction_cids.header_id = header_cids.block_hash
-												AND transaction_cids.block_number = header_cids.block_number
-											)
-											INNER JOIN public.blocks ON (
-												receipt_cids.leaf_mh_key = blocks.key
-												AND receipt_cids.block_number = blocks.block_number
-											)
+											INNER JOIN eth.transaction_cids ON (receipt_cids.tx_id = transaction_cids.tx_hash)
+											INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.block_hash)
+											INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = blocks.key)
 										WHERE block_number = $1
 										ORDER BY eth.transaction_cids.index ASC`
 	RetrieveReceiptByTxHashPgStr = `SELECT receipt_cids.leaf_cid, data
 									FROM eth.receipt_cids
-										INNER JOIN eth.transaction_cids ON (
-											receipt_cids.tx_id = transaction_cids.tx_hash
-											AND receipt_cids.block_number = transaction_cids.block_number
-										)
-										INNER JOIN public.blocks ON (
-											receipt_cids.leaf_mh_key = blocks.key
-											AND receipt_cids.block_number = blocks.block_number
-										)
+										INNER JOIN eth.transaction_cids ON (receipt_cids.tx_id = transaction_cids.tx_hash)
+										INNER JOIN public.blocks ON (receipt_cids.leaf_mh_key = blocks.key)
 									WHERE tx_hash = $1`
 	RetrieveAccountByLeafKeyAndBlockHashPgStr = `SELECT state_cids.cid, data, state_cids.node_type
 												FROM eth.state_cids
-													INNER JOIN eth.header_cids ON (
-														state_cids.header_id = header_cids.block_hash
-														AND state_cids.block_number = header_cids.block_number
-													)
-													INNER JOIN public.blocks ON (
-														state_cids.mh_key = blocks.key
-														AND state_cids.block_number = blocks.block_number
-													)
+													INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
+													INNER JOIN public.blocks ON (state_cids.mh_key = blocks.key)
 												WHERE state_leaf_key = $1
-												AND header_cids.block_number <= (SELECT block_number
+												AND block_number <= (SELECT block_number
 																	FROM eth.header_cids
 																	WHERE block_hash = $2)
-												AND header_cids.block_hash = (SELECT canonical_header_hash(header_cids.block_number))
-												ORDER BY header_cids.block_number DESC
+												AND header_cids.block_hash = (SELECT canonical_header_hash(block_number))
+												ORDER BY block_number DESC
 												LIMIT 1`
 	RetrieveAccountByLeafKeyAndBlockNumberPgStr = `SELECT state_cids.cid, data, state_cids.node_type
 													FROM eth.state_cids
-														INNER JOIN eth.header_cids ON (
-															state_cids.header_id = header_cids.block_hash
-															AND state_cids.block_number = header_cids.block_number
-														)
-														INNER JOIN public.blocks ON (
-															state_cids.mh_key = blocks.key
-															AND state_cids.block_number = blocks.block_number
-														)
+														INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
+														INNER JOIN public.blocks ON (state_cids.mh_key = blocks.key)
 													WHERE state_leaf_key = $1
 													AND block_number <= $2
 													ORDER BY block_number DESC
@@ -219,16 +132,9 @@ const (
 																		INNER JOIN eth.state_cids ON (
 																			storage_cids.header_id = state_cids.header_id
 																			AND storage_cids.state_path = state_cids.state_path
-																			AND storage_cids.block_number = state_cids.block_number
-																		)
-																		INNER JOIN eth.header_cids ON (
-																			state_cids.header_id = header_cids.block_hash
-																			AND state_cids.block_number = header_cids.block_number
-																		)
-																		INNER JOIN public.blocks ON (
-																			storage_cids.mh_key = blocks.key
-																			AND storage_cids.block_number = blocks.block_number
 																		)
+																		INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
+																		INNER JOIN public.blocks ON (storage_cids.mh_key = blocks.key)
 																	WHERE state_leaf_key = $1
 																	AND storage_leaf_key = $2
 																	AND block_number <= $3
@@ -239,23 +145,16 @@ const (
 																		INNER JOIN eth.state_cids ON (
 																			storage_cids.header_id = state_cids.header_id
 																			AND storage_cids.state_path = state_cids.state_path
-																			AND storage_cids.block_number = state_cids.block_number
-																		)
-																		INNER JOIN eth.header_cids ON (
-																			state_cids.header_id = header_cids.block_hash
-																			AND state_cids.block_number = header_cids.block_number
-																		)
-																		INNER JOIN public.blocks ON (
-																			storage_cids.mh_key = blocks.key
-																			AND storage_cids.block_number = blocks.block_number
 																		)
+																		INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
+																		INNER JOIN public.blocks ON (storage_cids.mh_key = blocks.key)
 																	WHERE state_leaf_key = $1
 																	AND storage_leaf_key = $2
-																	AND header_cids.block_number <= (SELECT block_number
+																	AND block_number <= (SELECT block_number
 																						FROM eth.header_cids
 																						WHERE block_hash = $3)
-																	AND header_cids.block_hash = (SELECT canonical_header_hash(header_cids.block_number))
-																	ORDER BY header_cids.block_number DESC
+																	AND header_cids.block_hash = (SELECT canonical_header_hash(block_number))
+																	ORDER BY block_number DESC
 																	LIMIT 1`
 )
 

pkg/eth/test_helpers/test_data.go

@@ -231,44 +231,40 @@ var (
 	}
 	MockTrxMetaPostPublsh = []models.TxModel{
 		{
-			BlockNumber: "1",
-			CID:         Trx1CID.String(), // This is empty until we go to publish to ipfs
-			MhKey:       Trx1MhKey,
-			Src:         SenderAddr.Hex(),
-			Dst:         Address.String(),
-			Index:       0,
-			TxHash:      MockTransactions[0].Hash().String(),
-			Data:        []byte{},
+			CID:    Trx1CID.String(), // This is empty until we go to publish to ipfs
+			MhKey:  Trx1MhKey,
+			Src:    SenderAddr.Hex(),
+			Dst:    Address.String(),
+			Index:  0,
+			TxHash: MockTransactions[0].Hash().String(),
+			Data:   []byte{},
 		},
 		{
-			BlockNumber: "1",
-			CID:         Trx2CID.String(),
-			MhKey:       Trx2MhKey,
-			Src:         SenderAddr.Hex(),
-			Dst:         AnotherAddress.String(),
-			Index:       1,
-			TxHash:      MockTransactions[1].Hash().String(),
-			Data:        []byte{},
+			CID:    Trx2CID.String(),
+			MhKey:  Trx2MhKey,
+			Src:    SenderAddr.Hex(),
+			Dst:    AnotherAddress.String(),
+			Index:  1,
+			TxHash: MockTransactions[1].Hash().String(),
+			Data:   []byte{},
 		},
 		{
-			BlockNumber: "1",
-			CID:         Trx3CID.String(),
-			MhKey:       Trx3MhKey,
-			Src:         SenderAddr.Hex(),
-			Dst:         "",
-			Index:       2,
-			TxHash:      MockTransactions[2].Hash().String(),
-			Data:        MockContractByteCode,
+			CID:    Trx3CID.String(),
+			MhKey:  Trx3MhKey,
+			Src:    SenderAddr.Hex(),
+			Dst:    "",
+			Index:  2,
+			TxHash: MockTransactions[2].Hash().String(),
+			Data:   MockContractByteCode,
 		},
 		{
-			BlockNumber: "1",
-			CID:         Trx4CID.String(),
-			MhKey:       Trx4MhKey,
-			Src:         SenderAddr.Hex(),
-			Dst:         AnotherAddress1.String(),
-			Index:       3,
-			TxHash:      MockTransactions[3].Hash().String(),
-			Data:        []byte{},
+			CID:    Trx4CID.String(),
+			MhKey:  Trx4MhKey,
+			Src:    SenderAddr.Hex(),
+			Dst:    AnotherAddress1.String(),
+			Index:  3,
+			TxHash: MockTransactions[3].Hash().String(),
+			Data:   []byte{},
 		},
 	}
 	MockRctMeta = []models.ReceiptModel{
@@ -300,28 +296,24 @@ var (
 
 	MockRctMetaPostPublish = []models.ReceiptModel{
 		{
-			BlockNumber:  "1",
 			LeafCID:      Rct1CID.String(),
 			LeafMhKey:    Rct1MhKey,
 			Contract:     "",
 			ContractHash: "",
 		},
 		{
-			BlockNumber:  "1",
 			LeafCID:      Rct2CID.String(),
 			LeafMhKey:    Rct2MhKey,
 			Contract:     "",
 			ContractHash: "",
 		},
 		{
-			BlockNumber:  "1",
 			LeafCID:      Rct3CID.String(),
 			LeafMhKey:    Rct3MhKey,
 			Contract:     ContractAddress.String(),
 			ContractHash: ContractHash,
 		},
 		{
-			BlockNumber:  "1",
 			LeafCID:      Rct4CID.String(),
 			LeafMhKey:    Rct4MhKey,
 			Contract:     "",
@@ -399,20 +391,18 @@ var (
 	}
 	MockStateMetaPostPublish = []models.StateNodeModel{
 		{
-			BlockNumber: "1",
-			CID:         State1CID.String(),
-			MhKey:       State1MhKey,
-			Path:        []byte{'\x06'},
-			NodeType:    2,
-			StateKey:    common.BytesToHash(ContractLeafKey).Hex(),
+			CID:      State1CID.String(),
+			MhKey:    State1MhKey,
+			Path:     []byte{'\x06'},
+			NodeType: 2,
+			StateKey: common.BytesToHash(ContractLeafKey).Hex(),
 		},
 		{
-			BlockNumber: "1",
-			CID:         State2CID.String(),
-			MhKey:       State2MhKey,
-			Path:        []byte{'\x0c'},
-			NodeType:    2,
-			StateKey:    common.BytesToHash(AccountLeafKey).Hex(),
+			CID:      State2CID.String(),
+			MhKey:    State2MhKey,
+			Path:     []byte{'\x0c'},
+			NodeType: 2,
+			StateKey: common.BytesToHash(AccountLeafKey).Hex(),
 		},
 	}
 	MockStorageNodes = map[string][]sdtypes.StorageNode{
@@ -471,13 +461,12 @@ var (
 		StateNodes:   MockStateMetaPostPublish,
 		StorageNodes: []models.StorageNodeWithStateKeyModel{
 			{
-				BlockNumber: "1",
-				Path:        []byte{},
-				CID:         StorageCID.String(),
-				MhKey:       StorageMhKey,
-				NodeType:    2,
-				StateKey:    common.BytesToHash(ContractLeafKey).Hex(),
-				StorageKey:  common.BytesToHash(StorageLeafKey).Hex(),
+				Path:       []byte{},
+				CID:        StorageCID.String(),
+				MhKey:      StorageMhKey,
+				NodeType:   2,
+				StateKey:   common.BytesToHash(ContractLeafKey).Hex(),
+				StorageKey: common.BytesToHash(StorageLeafKey).Hex(),
 			},
 		},
 	}
@@ -494,52 +483,43 @@ var (
 	MockIPLDs = eth.IPLDs{
 		BlockNumber: new(big.Int).Set(BlockNumber),
 		Header: models.IPLDModel{
-			BlockNumber: BlockNumber.String(),
-			Data:        HeaderIPLD.RawData(),
-			Key:         HeaderIPLD.Cid().String(),
+			Data: HeaderIPLD.RawData(),
+			Key:  HeaderIPLD.Cid().String(),
 		},
 		Transactions: []models.IPLDModel{
 			{
-				BlockNumber: BlockNumber.String(),
-				Data:        Trx1IPLD.RawData(),
-				Key:         Trx1IPLD.Cid().String(),
+				Data: Trx1IPLD.RawData(),
+				Key:  Trx1IPLD.Cid().String(),
 			},
 			{
-				BlockNumber: BlockNumber.String(),
-				Data:        Trx2IPLD.RawData(),
-				Key:         Trx2IPLD.Cid().String(),
+				Data: Trx2IPLD.RawData(),
+				Key:  Trx2IPLD.Cid().String(),
 			},
 			{
-				BlockNumber: BlockNumber.String(),
-				Data:        Trx3IPLD.RawData(),
-				Key:         Trx3IPLD.Cid().String(),
+				Data: Trx3IPLD.RawData(),
+				Key:  Trx3IPLD.Cid().String(),
 			},
 			{
-				BlockNumber: BlockNumber.String(),
-				Data:        Trx4IPLD.RawData(),
-				Key:         Trx4IPLD.Cid().String(),
+				Data: Trx4IPLD.RawData(),
+				Key:  Trx4IPLD.Cid().String(),
 			},
 		},
 		Receipts: []models.IPLDModel{
 			{
-				BlockNumber: BlockNumber.String(),
-				Data:        Rct1IPLD,
-				Key:         Rct1CID.String(),
+				Data: Rct1IPLD,
+				Key:  Rct1CID.String(),
 			},
 			{
-				BlockNumber: BlockNumber.String(),
-				Data:        Rct2IPLD,
-				Key:         Rct2CID.String(),
+				Data: Rct2IPLD,
+				Key:  Rct2CID.String(),
 			},
 			{
-				BlockNumber: BlockNumber.String(),
-				Data:        Rct3IPLD,
-				Key:         Rct3CID.String(),
+				Data: Rct3IPLD,
+				Key:  Rct3CID.String(),
 			},
 			{
-				BlockNumber: BlockNumber.String(),
-				Data:        Rct4IPLD,
-				Key:         Rct4CID.String(),
+				Data: Rct4IPLD,
+				Key:  Rct4CID.String(),
 			},
 		},
 		StateNodes: []eth.StateNode{
@@ -547,9 +527,8 @@ var (
 				StateLeafKey: common.BytesToHash(ContractLeafKey),
 				Type:         sdtypes.Leaf,
 				IPLD: models.IPLDModel{
-					BlockNumber: BlockNumber.String(),
-					Data:        State1IPLD.RawData(),
-					Key:         State1IPLD.Cid().String(),
+					Data: State1IPLD.RawData(),
+					Key:  State1IPLD.Cid().String(),
 				},
 				Path: []byte{'\x06'},
 			},
@@ -557,9 +536,8 @@ var (
 				StateLeafKey: common.BytesToHash(AccountLeafKey),
 				Type:         sdtypes.Leaf,
 				IPLD: models.IPLDModel{
-					BlockNumber: BlockNumber.String(),
-					Data:        State2IPLD.RawData(),
-					Key:         State2IPLD.Cid().String(),
+					Data: State2IPLD.RawData(),
+					Key:  State2IPLD.Cid().String(),
 				},
 				Path: []byte{'\x0c'},
 			},
@@ -570,9 +548,8 @@ var (
 				StorageLeafKey: common.BytesToHash(StorageLeafKey),
 				Type:           sdtypes.Leaf,
 				IPLD: models.IPLDModel{
-					BlockNumber: BlockNumber.String(),
-					Data:        StorageIPLD.RawData(),
-					Key:         StorageIPLD.Cid().String(),
+					Data: StorageIPLD.RawData(),
+					Key:  StorageIPLD.Cid().String(),
 				},
 				Path: []byte{},
 			},

pkg/fill/fill_suite_test.go

@@ -1,29 +0,0 @@
-// VulcanizeDB
-// Copyright © 2022 Vulcanize
-
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU Affero General Public License for more details.
-
-// You should have received a copy of the GNU Affero General Public License
-// along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
-package fill_test
-
-import (
-	"testing"
-
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-)
-
-func TestFill(t *testing.T) {
-	RegisterFailHandler(Fail)
-	RunSpecs(t, "ipld eth server fill test suite")
-}

pkg/fill/service.go

@@ -1,206 +0,0 @@
-// VulcanizeDB
-// Copyright © 2022 Vulcanize
-
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU Affero General Public License for more details.
-
-// You should have received a copy of the GNU Affero General Public License
-// along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
-package fill
-
-import (
-	"math"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/ethereum/go-ethereum/statediff"
-	"github.com/jmoiron/sqlx"
-	log "github.com/sirupsen/logrus"
-
-	"github.com/vulcanize/ipld-eth-server/v3/pkg/serve"
-)
-
-// WatchedAddress type is used to process currently watched addresses
-type WatchedAddress struct {
-	Address      string `db:"address"`
-	CreatedAt    uint64 `db:"created_at"`
-	WatchedAt    uint64 `db:"watched_at"`
-	LastFilledAt uint64 `db:"last_filled_at"`
-
-	StartBlock uint64
-	EndBlock   uint64
-}
-
-// Service is the underlying struct for the watched address gap filling service
-type Service struct {
-	db       *sqlx.DB
-	client   *rpc.Client
-	interval int
-	quitChan chan bool
-}
-
-// NewServer creates a new Service
-func New(config *serve.Config) *Service {
-	return &Service{
-		db:       config.DB,
-		client:   config.Client,
-		interval: config.WatchedAddressGapFillInterval,
-		quitChan: make(chan bool),
-	}
-}
-
-// Start is used to begin the service
-func (s *Service) Start(wg *sync.WaitGroup) {
-	defer wg.Done()
-	for {
-		select {
-		case <-s.quitChan:
-			log.Info("quiting eth ipld server process")
-			return
-		default:
-			s.fill()
-		}
-	}
-}
-
-// Stop is used to gracefully stop the service
-func (s *Service) Stop() {
-	log.Info("stopping watched address gap filler")
-	close(s.quitChan)
-}
-
-// fill performs the filling of indexing gap for watched addresses
-func (s *Service) fill() {
-	// Wait for specified interval duration
-	time.Sleep(time.Duration(s.interval) * time.Second)
-
-	// Get watched addresses from the db
-	rows := s.fetchWatchedAddresses()
-
-	// Get the block number to start fill at
-	// Get the block number to end fill at
-	fillWatchedAddresses, minStartBlock, maxEndBlock := s.GetFillAddresses(rows)
-
-	if len(fillWatchedAddresses) > 0 {
-		log.Infof("running watched address gap filler for block range: (%d, %d)", minStartBlock, maxEndBlock)
-	}
-
-	// Fill the missing diffs
-	for blockNumber := minStartBlock; blockNumber <= maxEndBlock; blockNumber++ {
-		params := statediff.Params{
-			IntermediateStateNodes:   true,
-			IntermediateStorageNodes: true,
-			IncludeBlock:             true,
-			IncludeReceipts:          true,
-			IncludeTD:                true,
-			IncludeCode:              true,
-		}
-
-		fillAddresses := []interface{}{}
-		for _, fillWatchedAddress := range fillWatchedAddresses {
-			if blockNumber >= fillWatchedAddress.StartBlock && blockNumber <= fillWatchedAddress.EndBlock {
-				params.WatchedAddresses = append(params.WatchedAddresses, common.HexToAddress(fillWatchedAddress.Address))
-				fillAddresses = append(fillAddresses, fillWatchedAddress.Address)
-			}
-		}
-
-		if len(fillAddresses) > 0 {
-			s.writeStateDiffAt(blockNumber, params)
-			s.UpdateLastFilledAt(blockNumber, fillAddresses)
-		}
-	}
-}
-
-// GetFillAddresses finds the encompassing range to perform fill for the given watched addresses
-// it also sets the address specific fill range
-func (s *Service) GetFillAddresses(rows []WatchedAddress) ([]WatchedAddress, uint64, uint64) {
-	fillWatchedAddresses := []WatchedAddress{}
-	minStartBlock := uint64(math.MaxUint64)
-	maxEndBlock := uint64(0)
-
-	for _, row := range rows {
-		// Check for a gap between created_at and watched_at
-		// CreatedAt and WatchedAt being equal is considered a gap of one block
-		if row.CreatedAt > row.WatchedAt {
-			continue
-		}
-
-		startBlock := uint64(0)
-		endBlock := uint64(0)
-
-		// Check if some of the gap was filled earlier
-		if row.LastFilledAt > 0 {
-			if row.LastFilledAt < row.WatchedAt {
-				startBlock = row.LastFilledAt + 1
-			}
-		} else {
-			startBlock = row.CreatedAt
-		}
-
-		// Add the address for filling
-		if startBlock > 0 {
-			row.StartBlock = startBlock
-			if startBlock < minStartBlock {
-				minStartBlock = startBlock
-			}
-
-			endBlock = row.WatchedAt
-			row.EndBlock = endBlock
-			if endBlock > maxEndBlock {
-				maxEndBlock = endBlock
-			}
-
-			fillWatchedAddresses = append(fillWatchedAddresses, row)
-		}
-	}
-
-	return fillWatchedAddresses, minStartBlock, maxEndBlock
-}
-
-// UpdateLastFilledAt updates the fill status for the provided addresses in the db
-func (s *Service) UpdateLastFilledAt(blockNumber uint64, fillAddresses []interface{}) {
-	// Prepare the query
-	query := "UPDATE eth_meta.watched_addresses SET last_filled_at=? WHERE address IN (?" + strings.Repeat(",?", len(fillAddresses)-1) + ")"
-	query = s.db.Rebind(query)
-
-	args := []interface{}{blockNumber}
-	args = append(args, fillAddresses...)
-
-	// Execute the update query
-	_, err := s.db.Exec(query, args...)
-	if err != nil {
-		log.Fatalf(err.Error())
-	}
-}
-
-// fetchWatchedAddresses fetches watched addresses from the db
-func (s *Service) fetchWatchedAddresses() []WatchedAddress {
-	rows := []WatchedAddress{}
-	pgStr := "SELECT * FROM eth_meta.watched_addresses"
-
-	err := s.db.Select(&rows, pgStr)
-	if err != nil {
-		log.Fatalf("Error fetching watched addreesses: %s", err.Error())
-	}
-
-	return rows
-}
-
-// writeStateDiffAt makes a RPC call to writeout statediffs at a blocknumber with the given params
-func (s *Service) writeStateDiffAt(blockNumber uint64, params statediff.Params) {
-	err := s.client.Call(nil, "statediff_writeStateDiffAt", blockNumber, params)
-	if err != nil {
-		log.Fatalf("Error making a RPC call to write statediff at block number %d: %s", blockNumber, err.Error())
-	}
-}

pkg/fill/service_test.go

@@ -1,411 +0,0 @@
-// VulcanizeDB
-// Copyright © 2022 Vulcanize
-
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU Affero General Public License for more details.
-
-// You should have received a copy of the GNU Affero General Public License
-// along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
-package fill_test
-
-import (
-	"context"
-	"math/big"
-
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-
-	"github.com/ethereum/go-ethereum/params"
-	"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
-	sdtypes "github.com/ethereum/go-ethereum/statediff/types"
-	"github.com/jmoiron/sqlx"
-
-	"github.com/vulcanize/ipld-eth-server/v3/pkg/eth/test_helpers"
-	fill "github.com/vulcanize/ipld-eth-server/v3/pkg/fill"
-	"github.com/vulcanize/ipld-eth-server/v3/pkg/serve"
-	"github.com/vulcanize/ipld-eth-server/v3/pkg/shared"
-)
-
-var _ = Describe("Service", func() {
-
-	var (
-		db                      *sqlx.DB
-		watchedAddressGapFiller *fill.Service
-		statediffIndexer        interfaces.StateDiffIndexer
-		err                     error
-
-		contract1Address = "0x5d663F5269090bD2A7DC2390c911dF6083D7b28F"
-		contract2Address = "0x6Eb7e5C66DB8af2E96159AC440cbc8CDB7fbD26B"
-		contract3Address = "0xcfeB164C328CA13EFd3C77E1980d94975aDfedfc"
-
-		chainConfig = params.TestChainConfig
-		ctx         = context.Background()
-	)
-
-	It("test init", func() {
-		// db initialization
-		db = shared.SetupDB()
-
-		// indexer initialization
-		// statediffIndexer, err = indexer.NewStateDiffIndexer(nil, db)
-		statediffIndexer = shared.SetupTestStateDiffIndexer(ctx, chainConfig, test_helpers.Genesis.Hash())
-		Expect(err).ToNot(HaveOccurred())
-
-		// fill service intialization
-		watchedAddressGapFiller = fill.New(&serve.Config{
-			DB: db,
-		})
-	})
-
-	defer It("test teardown", func() {
-		shared.TearDownDB(db)
-	})
-
-	Describe("GetFillAddresses", func() {
-		Context("overlapping fill ranges", func() {
-			It("gives the range to run fill for each address", func() {
-				// input data
-				rows := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-					},
-					{
-						Address:      contract2Address,
-						CreatedAt:    40,
-						WatchedAt:    70,
-						LastFilledAt: 0,
-					},
-					{
-						Address:      contract3Address,
-						CreatedAt:    20,
-						WatchedAt:    30,
-						LastFilledAt: 0,
-					},
-				}
-
-				// expected output data
-				expectedOutputAddresses := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-						StartBlock:   10,
-						EndBlock:     50,
-					},
-					{
-						Address:      contract2Address,
-						CreatedAt:    40,
-						WatchedAt:    70,
-						LastFilledAt: 0,
-						StartBlock:   40,
-						EndBlock:     70,
-					},
-					{
-						Address:      contract3Address,
-						CreatedAt:    20,
-						WatchedAt:    30,
-						LastFilledAt: 0,
-						StartBlock:   20,
-						EndBlock:     30,
-					},
-				}
-				expectedOutputStartBlock := uint64(10)
-				expectedOutputEndBlock := uint64(70)
-
-				fillWatchedAddresses, minStartBlock, maxEndBlock := watchedAddressGapFiller.GetFillAddresses(rows)
-
-				Expect(fillWatchedAddresses).To(Equal(expectedOutputAddresses))
-				Expect(minStartBlock).To(Equal(expectedOutputStartBlock))
-				Expect(maxEndBlock).To(Equal(expectedOutputEndBlock))
-			})
-		})
-
-		Context("non-overlapping fill ranges", func() {
-			It("gives the range to run fill for each address", func() {
-				// input data
-				rows := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-					},
-					{
-						Address:      contract2Address,
-						CreatedAt:    70,
-						WatchedAt:    90,
-						LastFilledAt: 0,
-					},
-				}
-
-				// expected output data
-				expectedOutputAddresses := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-						StartBlock:   10,
-						EndBlock:     50,
-					},
-					{
-						Address:      contract2Address,
-						CreatedAt:    70,
-						WatchedAt:    90,
-						LastFilledAt: 0,
-						StartBlock:   70,
-						EndBlock:     90,
-					},
-				}
-				expectedOutputStartBlock := uint64(10)
-				expectedOutputEndBlock := uint64(90)
-
-				fillWatchedAddresses, minStartBlock, maxEndBlock := watchedAddressGapFiller.GetFillAddresses(rows)
-
-				Expect(fillWatchedAddresses).To(Equal(expectedOutputAddresses))
-				Expect(minStartBlock).To(Equal(expectedOutputStartBlock))
-				Expect(maxEndBlock).To(Equal(expectedOutputEndBlock))
-			})
-		})
-
-		Context("a contract watched before it was created", func() {
-			It("gives no range for an address when it is watched before it's created", func() {
-				// input data
-				rows := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-					},
-					{
-						Address:      contract2Address,
-						CreatedAt:    90,
-						WatchedAt:    70,
-						LastFilledAt: 0,
-					},
-				}
-
-				// expected output data
-				expectedOutputAddresses := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-						StartBlock:   10,
-						EndBlock:     50,
-					},
-				}
-				expectedOutputStartBlock := uint64(10)
-				expectedOutputEndBlock := uint64(50)
-
-				fillWatchedAddresses, minStartBlock, maxEndBlock := watchedAddressGapFiller.GetFillAddresses(rows)
-
-				Expect(fillWatchedAddresses).To(Equal(expectedOutputAddresses))
-				Expect(minStartBlock).To(Equal(expectedOutputStartBlock))
-				Expect(maxEndBlock).To(Equal(expectedOutputEndBlock))
-			})
-		})
-
-		Context("a contract having some of the gap filled earlier", func() {
-			It("gives the remaining range for an address to run fill for", func() {
-				// input data
-				rows := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-					},
-					{
-						Address:      contract2Address,
-						CreatedAt:    40,
-						WatchedAt:    70,
-						LastFilledAt: 50,
-					},
-				}
-
-				// expected output data
-				expectedOutputAddresses := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-						StartBlock:   10,
-						EndBlock:     50,
-					},
-					{
-						Address:      contract2Address,
-						CreatedAt:    40,
-						WatchedAt:    70,
-						LastFilledAt: 50,
-						StartBlock:   51,
-						EndBlock:     70,
-					},
-				}
-				expectedOutputStartBlock := uint64(10)
-				expectedOutputEndBlock := uint64(70)
-
-				fillWatchedAddresses, minStartBlock, maxEndBlock := watchedAddressGapFiller.GetFillAddresses(rows)
-
-				Expect(fillWatchedAddresses).To(Equal(expectedOutputAddresses))
-				Expect(minStartBlock).To(Equal(expectedOutputStartBlock))
-				Expect(maxEndBlock).To(Equal(expectedOutputEndBlock))
-			})
-
-			It("gives no range for an address when the gap is already filled", func() {
-				// input data
-				rows := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-					},
-					{
-						Address:      contract2Address,
-						CreatedAt:    40,
-						WatchedAt:    70,
-						LastFilledAt: 70,
-					},
-				}
-
-				// expected output data
-				expectedOutputAddresses := []fill.WatchedAddress{
-					{
-						Address:      contract1Address,
-						CreatedAt:    10,
-						WatchedAt:    50,
-						LastFilledAt: 0,
-						StartBlock:   10,
-						EndBlock:     50,
-					},
-				}
-				expectedOutputStartBlock := uint64(10)
-				expectedOutputEndBlock := uint64(50)
-
-				fillWatchedAddresses, minStartBlock, maxEndBlock := watchedAddressGapFiller.GetFillAddresses(rows)
-
-				Expect(fillWatchedAddresses).To(Equal(expectedOutputAddresses))
-				Expect(minStartBlock).To(Equal(expectedOutputStartBlock))
-				Expect(maxEndBlock).To(Equal(expectedOutputEndBlock))
-			})
-		})
-	})
-
-	Describe("UpdateLastFilledAt", func() {
-		pgStr := "SELECT * FROM eth_meta.watched_addresses"
-
-		BeforeEach(func() {
-			shared.TearDownDB(db)
-		})
-
-		It("updates last filled at for a single address", func() {
-			// fill db with watched addresses
-			watchedAddresses := []sdtypes.WatchAddressArg{
-				{
-					Address:   contract1Address,
-					CreatedAt: 10,
-				},
-			}
-			watchedAt := uint64(50)
-			err = statediffIndexer.InsertWatchedAddresses(watchedAddresses, big.NewInt(int64(watchedAt)))
-			Expect(err).ToNot(HaveOccurred())
-
-			// update last filled at block in the db
-			fillAddresses := []interface{}{
-				contract1Address,
-			}
-			fillAt := uint64(12)
-			watchedAddressGapFiller.UpdateLastFilledAt(fillAt, fillAddresses)
-
-			// expected data
-			expectedData := []fill.WatchedAddress{
-				{
-					Address:      contract1Address,
-					CreatedAt:    10,
-					WatchedAt:    watchedAt,
-					LastFilledAt: fillAt,
-				},
-			}
-
-			rows := []fill.WatchedAddress{}
-			err = db.Select(&rows, pgStr)
-			Expect(err).ToNot(HaveOccurred())
-
-			Expect(rows).To(Equal(expectedData))
-		})
-
-		It("updates last filled at for multiple address", func() {
-			// fill db with watched addresses
-			watchedAddresses := []sdtypes.WatchAddressArg{
-				{
-					Address:   contract1Address,
-					CreatedAt: 10,
-				},
-				{
-					Address:   contract2Address,
-					CreatedAt: 20,
-				},
-				{
-					Address:   contract3Address,
-					CreatedAt: 30,
-				},
-			}
-			watchedAt := uint64(50)
-			err = statediffIndexer.InsertWatchedAddresses(watchedAddresses, big.NewInt(int64(watchedAt)))
-			Expect(err).ToNot(HaveOccurred())
-
-			// update last filled at block in the db
-			fillAddresses := []interface{}{
-				contract1Address,
-				contract2Address,
-				contract3Address,
-			}
-			fillAt := uint64(50)
-			watchedAddressGapFiller.UpdateLastFilledAt(fillAt, fillAddresses)
-
-			// expected data
-			expectedData := []fill.WatchedAddress{
-				{
-					Address:      contract1Address,
-					CreatedAt:    10,
-					WatchedAt:    watchedAt,
-					LastFilledAt: fillAt,
-				},
-				{
-					Address:      contract2Address,
-					CreatedAt:    20,
-					WatchedAt:    watchedAt,
-					LastFilledAt: fillAt,
-				},
-				{
-					Address:      contract3Address,
-					CreatedAt:    30,
-					WatchedAt:    watchedAt,
-					LastFilledAt: fillAt,
-				},
-			}
-
-			rows := []fill.WatchedAddress{}
-			err = db.Select(&rows, pgStr)
-			Expect(err).ToNot(HaveOccurred())
-
-			Expect(rows).To(Equal(expectedData))
-		})
-	})
-})

pkg/serve/config.go

@@ -55,9 +55,6 @@ const (
 
 	VALIDATOR_ENABLED         = "VALIDATOR_ENABLED"
 	VALIDATOR_EVERY_NTH_BLOCK = "VALIDATOR_EVERY_NTH_BLOCK"
-
-	WATCHED_ADDRESS_GAP_FILLER_ENABLED  = "WATCHED_ADDRESS_GAP_FILLER_ENABLED"
-	WATCHED_ADDRESS_GAP_FILLER_INTERVAL = "WATCHED_ADDRESS_GAP_FILLER_INTERVAL"
 )
 
 // Config struct
@@ -98,9 +95,6 @@ type Config struct {
 
 	StateValidationEnabled       bool
 	StateValidationEveryNthBlock uint64
-
-	WatchedAddressGapFillerEnabled bool
-	WatchedAddressGapFillInterval  int
 }
 
 // NewConfig is used to initialize a watcher config from a .toml file
@@ -237,8 +231,6 @@ func NewConfig() (*Config, error) {
 
 	c.loadValidatorConfig()
 
-	c.loadWatchedAddressGapFillerConfig()
-
 	return c, err
 }
 
@@ -301,11 +293,3 @@ func (c *Config) loadValidatorConfig() {
 	c.StateValidationEnabled = viper.GetBool("validator.enabled")
 	c.StateValidationEveryNthBlock = viper.GetUint64("validator.everyNthBlock")
 }
-
-func (c *Config) loadWatchedAddressGapFillerConfig() {
-	viper.BindEnv("watch.fill.enabled", WATCHED_ADDRESS_GAP_FILLER_ENABLED)
-	viper.BindEnv("watch.fill.interval", WATCHED_ADDRESS_GAP_FILLER_INTERVAL)
-
-	c.WatchedAddressGapFillerEnabled = viper.GetBool("watch.fill.enabled")
-	c.WatchedAddressGapFillInterval = viper.GetInt("watch.fill.interval")
-}

pkg/shared/functions.go

@@ -18,9 +18,11 @@ package shared
 
 import (
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
 	"github.com/ipfs/go-cid"
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	dshelp "github.com/ipfs/go-ipfs-ds-help"
+	node "github.com/ipfs/go-ipld-format"
 	"github.com/jmoiron/sqlx"
 	"github.com/sirupsen/logrus"
 )
@@ -48,11 +50,31 @@ func Rollback(tx *sqlx.Tx) {
 	}
 }
 
-// FetchIPLDByMhKeyAndBlockNumber is used to retrieve an ipld from Postgres blockstore with the provided tx, mhkey string and blockNumber
-func FetchIPLDByMhKeyAndBlockNumber(tx *sqlx.Tx, mhKey string, blockNumber uint64) ([]byte, error) {
-	pgStr := `SELECT data FROM public.blocks WHERE key = $1 AND block_number = $2`
+// PublishIPLD is used to insert an ipld into Postgres blockstore with the provided tx
+func PublishIPLD(tx *sqlx.Tx, i node.Node) error {
+	dbKey := dshelp.MultihashToDsKey(i.Cid().Hash())
+	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
+	raw := i.RawData()
+	_, err := tx.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, prefixedKey, raw)
+	return err
+}
+
+// FetchIPLD is used to retrieve an ipld from Postgres blockstore with the provided tx and cid string
+func FetchIPLD(tx *sqlx.Tx, cid string) ([]byte, error) {
+	mhKey, err := MultihashKeyFromCIDString(cid)
+	if err != nil {
+		return nil, err
+	}
+	pgStr := `SELECT data FROM public.blocks WHERE key = $1`
 	var block []byte
-	return block, tx.Get(&block, pgStr, mhKey, blockNumber)
+	return block, tx.Get(&block, pgStr, mhKey)
+}
+
+// FetchIPLDByMhKey is used to retrieve an ipld from Postgres blockstore with the provided tx and mhkey string
+func FetchIPLDByMhKey(tx *sqlx.Tx, mhKey string) ([]byte, error) {
+	pgStr := `SELECT data FROM public.blocks WHERE key = $1`
+	var block []byte
+	return block, tx.Get(&block, pgStr, mhKey)
 }
 
 // MultihashKeyFromCID converts a cid into a blockstore-prefixed multihash db key string
@@ -70,3 +92,15 @@ func MultihashKeyFromCIDString(c string) (string, error) {
 	dbKey := dshelp.MultihashToDsKey(dc.Hash())
 	return blockstore.BlockPrefix.String() + dbKey.String(), nil
 }
+
+// PublishRaw derives a cid from raw bytes and provided codec and multihash type, and writes it to the db tx
+func PublishRaw(tx *sqlx.Tx, codec, mh uint64, raw []byte) (string, error) {
+	c, err := ipld.RawdataToCid(codec, raw, mh)
+	if err != nil {
+		return "", err
+	}
+	dbKey := dshelp.MultihashToDsKey(c.Hash())
+	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
+	_, err = tx.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, prefixedKey, raw)
+	return c.String(), err
+}

pkg/shared/test_helpers.go

@@ -58,12 +58,8 @@ func SetupDB() *sqlx.DB {
 func TearDownDB(db *sqlx.DB) {
 	tx, err := db.Beginx()
 	Expect(err).NotTo(HaveOccurred())
-	_, err = tx.Exec(`DELETE FROM nodes`)
-	Expect(err).NotTo(HaveOccurred())
 	_, err = tx.Exec(`DELETE FROM eth.header_cids`)
 	Expect(err).NotTo(HaveOccurred())
-	_, err = tx.Exec(`DELETE FROM eth.uncle_cids`)
-	Expect(err).NotTo(HaveOccurred())
 	_, err = tx.Exec(`DELETE FROM eth.transaction_cids`)
 	Expect(err).NotTo(HaveOccurred())
 	_, err = tx.Exec(`DELETE FROM eth.receipt_cids`)
@@ -72,10 +68,6 @@ func TearDownDB(db *sqlx.DB) {
 	Expect(err).NotTo(HaveOccurred())
 	_, err = tx.Exec(`DELETE FROM eth.storage_cids`)
 	Expect(err).NotTo(HaveOccurred())
-	_, err = tx.Exec(`DELETE FROM eth.state_accounts`)
-	Expect(err).NotTo(HaveOccurred())
-	_, err = tx.Exec(`DELETE FROM eth.access_list_elements`)
-	Expect(err).NotTo(HaveOccurred())
 	_, err = tx.Exec(`DELETE FROM blocks`)
 	Expect(err).NotTo(HaveOccurred())
 	_, err = tx.Exec(`DELETE FROM eth.log_cids`)

scripts/run_integration_test.sh

@@ -4,8 +4,6 @@ set -o xtrace
 export ETH_FORWARD_ETH_CALLS=false
 export DB_WRITE=true
 export ETH_PROXY_ON_ERROR=false
-export WATCHED_ADDRESS_GAP_FILLER_ENABLED=false
-export WATCHED_ADDRESS_GAP_FILLER_INTERVAL=5
 
 export PGPASSWORD=password
 export DATABASE_USER=vdbm

scripts/run_integration_test_forward_eth_calls.sh

@@ -4,8 +4,6 @@ set -o xtrace
 export ETH_FORWARD_ETH_CALLS=true
 export DB_WRITE=false
 export ETH_PROXY_ON_ERROR=false
-export WATCHED_ADDRESS_GAP_FILLER_ENABLED=false
-export WATCHED_ADDRESS_GAP_FILLER_INTERVAL=5
 
 export PGPASSWORD=password
 export DATABASE_USER=vdbm

scripts/run_integration_test_watched_address_gap_filler.sh

@@ -1,19 +0,0 @@
-set -e
-set -o xtrace
-
-export ETH_FORWARD_ETH_CALLS=false
-export DB_WRITE=true
-export ETH_PROXY_ON_ERROR=false
-export WATCHED_ADDRESS_GAP_FILLER_ENABLED=true
-export WATCHED_ADDRESS_GAP_FILLER_INTERVAL=5
-
-export PGPASSWORD=password
-export DATABASE_USER=vdbm
-export DATABASE_PORT=8077
-export DATABASE_PASSWORD=password
-export DATABASE_HOSTNAME=127.0.0.1
-
-# Wait for containers to be up and execute the integration test.
-while [ "$(curl -s -o /dev/null -w ''%{http_code}'' localhost:8081)" != "200" ]; do echo "waiting for ipld-eth-server..." && sleep 5; done && \
-          while [ "$(curl -s -o /dev/null -w ''%{http_code}'' localhost:8545)" != "200" ]; do echo "waiting for geth-statediff..." && sleep 5; done && \
-          make integrationtest

scripts/run_unit_test.sh

@@ -1,26 +1,8 @@
-#!/bin/bash
+# Clear up existing docker images and volume.
+docker-compose down --remove-orphans --volumes
 
-set -e
+docker-compose -f docker-compose.yml up -d ipld-eth-db
+sleep 10
+PGPASSWORD=password DATABASE_USER=vdbm DATABASE_PORT=8077 DATABASE_PASSWORD=password DATABASE_HOSTNAME=127.0.0.1 DATABASE_NAME=vulcanize_testing make test
 
-mkdir -p out
-
-# Remove existing docker-tsdb directory
-rm -rf out/docker-tsdb/
-
-# Copy over files to setup TimescaleDB
-ID=$(docker create vulcanize/ipld-eth-db:v4.1.1-alpha)
-docker cp $ID:/app/docker-tsdb out/docker-tsdb/
-docker rm -v $ID
-
-# Spin up TimescaleDB
-docker-compose -f out/docker-tsdb/docker-compose.test.yml -f docker-compose.yml up ipld-eth-db
-trap "docker-compose -f out/docker-tsdb/docker-compose.test.yml -f docker-compose.yml down --remove-orphans --volumes; rm -rf out/" SIGINT SIGTERM ERR
-sleep 45
-
-# Run unit tests
-go clean -testcache
-PGPASSWORD=password DATABASE_USER=vdbm DATABASE_PORT=8066 DATABASE_PASSWORD=password DATABASE_HOSTNAME=127.0.0.1 DATABASE_NAME=vulcanize_testing_v4 make test
-
-# Clean up
-docker-compose -f out/docker-tsdb/docker-compose.test.yml -f docker-compose.yml down --remove-orphans --volumes
-rm -rf out/
+docker-compose down --remove-orphans --volumes

test/README.md

@@ -4,10 +4,10 @@
 
 - Clone [stack-orchestrator](https://github.com/vulcanize/stack-orchestrator) and [go-ethereum](https://github.com/vulcanize/go-ethereum) repositories.
 
-- Checkout [v3 release](https://github.com/vulcanize/go-ethereum/releases/tag/v1.10.17-statediff-3.2.0) in go-ethereum repo.
+- Checkout [v3 release](https://github.com/vulcanize/go-ethereum/releases/tag/v1.10.17-statediff-3.2.1) in go-ethereum repo.
   ```bash
   # In go-ethereum repo.
-  git checkout v1.10.17-statediff-3.2.0
+  git checkout v1.10.17-statediff-3.2.1
   ```
 
 - Checkout working commit in stack-orchestrator repo.
@@ -85,28 +85,3 @@
     ```bash
     ./scripts/run_integration_test_forward_eth_calls.sh
     ```
-
-  - Update `config.sh` file:
-    ```bash
-    #!/bin/bash
-
-    # Path to go-ethereum repo.
-    vulcanize_go_ethereum=~/go-ethereum/
-
-    # Path to ipld-eth-server repo.
-    vulcanize_ipld_eth_server=~/ipld-eth-server/
-
-    db_write=true
-    eth_forward_eth_calls=false
-    eth_proxy_on_error=false
-    eth_http_path="go-ethereum:8545"
-    watched_addres_gap_filler_enabled=true
-    watched_addres_gap_filler_interval=5
-    ```
-
-  - Stop the stack-orchestrator and start again using the same command
-
-  - Run integration tests for watched addresses with gap filling service enabled:
-    ```bash
-    ./scripts/run_integration_test_watched_address_gap_filler.sh
-    ```

test/direct_proxy_integration_test.go

@@ -20,8 +20,6 @@ import (
 var _ = Describe("Integration test", func() {
 	directProxyEthCalls, err := strconv.ParseBool(os.Getenv("ETH_FORWARD_ETH_CALLS"))
 	Expect(err).To(BeNil())
-	watchedAddressServiceEnabled, err := strconv.ParseBool(os.Getenv("WATCHED_ADDRESS_GAP_FILLER_ENABLED"))
-	Expect(err).To(BeNil())
 	gethHttpPath := "http://127.0.0.1:8545"
 	gethClient, err := ethclient.Dial(gethHttpPath)
 	Expect(err).ToNot(HaveOccurred())
@@ -41,7 +39,7 @@ var _ = Describe("Integration test", func() {
 	sleepInterval := 2 * time.Second
 
 	BeforeEach(func() {
-		if !directProxyEthCalls || watchedAddressServiceEnabled {
+		if !directProxyEthCalls {
 			Skip("skipping direct-proxy-forwarding integration tests")
 		}
 	})

test/integration_test.go

@@ -31,8 +31,6 @@ var (
 var _ = Describe("Integration test", func() {
 	directProxyEthCalls, err := strconv.ParseBool(os.Getenv("ETH_FORWARD_ETH_CALLS"))
 	Expect(err).To(BeNil())
-	watchedAddressServiceEnabled, err := strconv.ParseBool(os.Getenv("WATCHED_ADDRESS_GAP_FILLER_ENABLED"))
-	Expect(err).To(BeNil())
 	gethHttpPath := "http://127.0.0.1:8545"
 	gethClient, err := ethclient.Dial(gethHttpPath)
 	Expect(err).ToNot(HaveOccurred())
@@ -52,7 +50,7 @@ var _ = Describe("Integration test", func() {
 	sleepInterval := 2 * time.Second
 
 	BeforeEach(func() {
-		if directProxyEthCalls || watchedAddressServiceEnabled {
+		if directProxyEthCalls {
 			Skip("skipping no-direct-proxy-forwarding integration tests")
 		}
 	})

test/watch_address_integration_test.go

@@ -28,9 +28,6 @@ var _ = Describe("WatchAddress integration test", func() {
 	dbWrite, err := strconv.ParseBool(os.Getenv("DB_WRITE"))
 	Expect(err).To(BeNil())
 
-	watchedAddressServiceEnabled, err := strconv.ParseBool(os.Getenv("WATCHED_ADDRESS_GAP_FILLER_ENABLED"))
-	Expect(err).To(BeNil())
-
 	gethHttpPath := "http://127.0.0.1:8545"
 	gethRPCClient, err := rpc.Dial(gethHttpPath)
 	Expect(err).ToNot(HaveOccurred())
@@ -72,7 +69,7 @@ var _ = Describe("WatchAddress integration test", func() {
 	)
 
 	BeforeEach(func() {
-		if !dbWrite || watchedAddressServiceEnabled {
+		if !dbWrite {
 			Skip("skipping WatchAddress API integration tests")
 		}
 	})

test/watched_address_gap_filling_service_integration_test.go

@@ -1,318 +0,0 @@
-package integration_test
-
-import (
-	"context"
-	"math/big"
-	"os"
-	"strconv"
-	"time"
-
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/ethclient"
-	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/ethereum/go-ethereum/statediff/types"
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-
-	integration "github.com/vulcanize/ipld-eth-server/v3/test"
-)
-
-var _ = Describe("Watched address gap filling service integration test", func() {
-	dbWrite, err := strconv.ParseBool(os.Getenv("DB_WRITE"))
-	Expect(err).To(BeNil())
-
-	watchedAddressServiceEnabled, err := strconv.ParseBool(os.Getenv("WATCHED_ADDRESS_GAP_FILLER_ENABLED"))
-	Expect(err).To(BeNil())
-
-	serviceInterval, err := strconv.ParseInt(os.Getenv("WATCHED_ADDRESS_GAP_FILLER_INTERVAL"), 10, 0)
-	Expect(err).To(BeNil())
-
-	gethHttpPath := "http://127.0.0.1:8545"
-	gethRPCClient, err := rpc.Dial(gethHttpPath)
-	Expect(err).ToNot(HaveOccurred())
-
-	ipldEthHttpPath := "http://127.0.0.1:8081"
-	ipldClient, err := ethclient.Dial(ipldEthHttpPath)
-	Expect(err).ToNot(HaveOccurred())
-	ipldRPCClient, err := rpc.Dial(ipldEthHttpPath)
-	Expect(err).ToNot(HaveOccurred())
-
-	var (
-		ctx = context.Background()
-
-		contractErr error
-		txErr       error
-
-		GLD1 *integration.ContractDeployed
-		SLV1 *integration.ContractDeployed
-		SLV2 *integration.ContractDeployed
-
-		countAIndex string
-		countBIndex string
-
-		oldCountA1 = big.NewInt(0)
-		oldCountA2 = big.NewInt(0)
-		oldCountB2 = big.NewInt(0)
-
-		updatedCountA1 = big.NewInt(1)
-		updatedCountA2 = big.NewInt(1)
-		updatedCountB2 = big.NewInt(1)
-
-		SLV2CountBIncrementedAt *integration.CountIncremented
-
-		contract1Salt = "contract1SLV"
-	)
-
-	BeforeEach(func() {
-		if !dbWrite || !watchedAddressServiceEnabled {
-			Skip("skipping watched address gap filling service integration tests")
-		}
-	})
-
-	It("test init", func() {
-		// Clear out watched addresses
-		err := integration.ClearWatchedAddresses(gethRPCClient)
-		Expect(err).ToNot(HaveOccurred())
-
-		// Deploy a GLD contract
-		GLD1, contractErr = integration.DeployContract()
-		Expect(contractErr).ToNot(HaveOccurred())
-
-		// Watch GLD1 contract
-		operation := types.Add
-		args := []types.WatchAddressArg{
-			{
-				Address:   GLD1.Address,
-				CreatedAt: uint64(GLD1.BlockNumber),
-			},
-		}
-		ipldErr := ipldRPCClient.Call(nil, ipldMethod, operation, args)
-		Expect(ipldErr).ToNot(HaveOccurred())
-
-		// Deploy two SLV contracts and update storage slots
-		SLV1, contractErr = integration.Create2Contract("SLVToken", contract1Salt)
-		Expect(contractErr).ToNot(HaveOccurred())
-
-		_, txErr = integration.IncrementCount(SLV1.Address, "A")
-		Expect(txErr).ToNot(HaveOccurred())
-
-		SLV2, contractErr = integration.DeploySLVContract()
-		Expect(contractErr).ToNot(HaveOccurred())
-
-		_, txErr = integration.IncrementCount(SLV2.Address, "A")
-		Expect(txErr).ToNot(HaveOccurred())
-		SLV2CountBIncrementedAt, txErr = integration.IncrementCount(SLV2.Address, "B")
-		Expect(txErr).ToNot(HaveOccurred())
-
-		// Get storage slot keys
-		storageSlotAKey, err := integration.GetStorageSlotKey("SLVToken", "countA")
-		Expect(err).ToNot(HaveOccurred())
-		countAIndex = storageSlotAKey.Key
-
-		storageSlotBKey, err := integration.GetStorageSlotKey("SLVToken", "countB")
-		Expect(err).ToNot(HaveOccurred())
-		countBIndex = storageSlotBKey.Key
-	})
-
-	defer It("test cleanup", func() {
-		// Destroy create2 contract
-		_, err := integration.DestroySLVContract(SLV1.Address)
-		Expect(err).ToNot(HaveOccurred())
-
-		// Clear out watched addresses
-		err = integration.ClearWatchedAddresses(gethRPCClient)
-		Expect(err).ToNot(HaveOccurred())
-	})
-
-	Context("previously unwatched contract watched", func() {
-		It("indexes state only for watched contract", func() {
-			// WatchedAddresses = [GLD1]
-			// SLV1, countA
-			countA1Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV1.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA1 := new(big.Int).SetBytes(countA1Storage)
-			Expect(countA1.String()).To(Equal(oldCountA1.String()))
-
-			// SLV2, countA
-			countA2Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV2.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA2 := new(big.Int).SetBytes(countA2Storage)
-			Expect(countA2.String()).To(Equal(oldCountA2.String()))
-
-			// SLV2, countB
-			countB2Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV2.Address), common.HexToHash(countBIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countB2 := new(big.Int).SetBytes(countB2Storage)
-			Expect(countB2.String()).To(Equal(oldCountB2.String()))
-		})
-
-		It("indexes past state on watching a contract", func() {
-			// Watch SLV1 contract
-			args := []types.WatchAddressArg{
-				{
-					Address:   SLV1.Address,
-					CreatedAt: uint64(SLV1.BlockNumber),
-				},
-			}
-			ipldErr := ipldRPCClient.Call(nil, ipldMethod, types.Add, args)
-			Expect(ipldErr).ToNot(HaveOccurred())
-
-			// Sleep for service interval + few extra seconds
-			time.Sleep(time.Duration(serviceInterval+2) * time.Second)
-
-			// WatchedAddresses = [GLD1, SLV1]
-			// SLV1, countA
-			countA1Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV1.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA1 := new(big.Int).SetBytes(countA1Storage)
-			Expect(countA1.String()).To(Equal(updatedCountA1.String()))
-
-			// SLV2, countA
-			countA2Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV2.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA2 := new(big.Int).SetBytes(countA2Storage)
-			Expect(countA2.String()).To(Equal(oldCountA2.String()))
-
-			// SLV2, countB
-			countB2Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV2.Address), common.HexToHash(countBIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countB2 := new(big.Int).SetBytes(countB2Storage)
-			Expect(countB2.String()).To(Equal(oldCountB2.String()))
-		})
-	})
-
-	Context("previously unwatched contract watched (different 'created_at')", func() {
-		It("indexes past state from 'created_at' onwards on watching a contract", func() {
-			// Watch SLV2 (created_at -> countB incremented) contract
-			args := []types.WatchAddressArg{
-				{
-					Address:   SLV2.Address,
-					CreatedAt: uint64(SLV2CountBIncrementedAt.BlockNumber),
-				},
-			}
-			ipldErr := ipldRPCClient.Call(nil, ipldMethod, types.Add, args)
-			Expect(ipldErr).ToNot(HaveOccurred())
-
-			// Sleep for service interval + few extra seconds
-			time.Sleep(time.Duration(serviceInterval+2) * time.Second)
-
-			// WatchedAddresses = [GLD1, SLV1, SLV2]
-			// SLV2, countA
-			countA2Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV2.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA2 := new(big.Int).SetBytes(countA2Storage)
-			Expect(countA2.String()).To(Equal(oldCountA2.String()))
-
-			// SLV2, countB
-			countB2Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV2.Address), common.HexToHash(countBIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countB2 := new(big.Int).SetBytes(countB2Storage)
-			Expect(countB2.String()).To(Equal(updatedCountB2.String()))
-		})
-
-		It("indexes missing past state on watching a contract from an earlier 'created_at'", func() {
-			// Remove SLV2 (created_at -> countB incremented) watched contract
-			args := []types.WatchAddressArg{
-				{
-					Address:   SLV2.Address,
-					CreatedAt: uint64(SLV2CountBIncrementedAt.BlockNumber),
-				},
-			}
-			ipldErr := ipldRPCClient.Call(nil, ipldMethod, types.Remove, args)
-			Expect(ipldErr).ToNot(HaveOccurred())
-
-			// Watch SLV2 (created_at -> deployment) contract
-			args = []types.WatchAddressArg{
-				{
-					Address:   SLV2.Address,
-					CreatedAt: uint64(SLV2.BlockNumber),
-				},
-			}
-			ipldErr = ipldRPCClient.Call(nil, ipldMethod, types.Add, args)
-			Expect(ipldErr).ToNot(HaveOccurred())
-
-			// Sleep for service interval + few extra seconds
-			time.Sleep(time.Duration(serviceInterval+2) * time.Second)
-
-			// WatchedAddresses = [GLD1, SLV1, SLV2]
-			// SLV2, countA
-			countA2Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV2.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA2 := new(big.Int).SetBytes(countA2Storage)
-			Expect(countA2.String()).To(Equal(updatedCountA2.String()))
-
-			// SLV2, countB
-			countB2Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV2.Address), common.HexToHash(countBIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countB2 := new(big.Int).SetBytes(countB2Storage)
-			Expect(countB2.String()).To(Equal(updatedCountB2.String()))
-		})
-	})
-
-	Context("destroyed contract redeployed and watched", func() {
-		It("returns zero value for destroyed contract", func() {
-			_, err := integration.DestroySLVContract(SLV1.Address)
-			Expect(err).ToNot(HaveOccurred())
-
-			updatedCountA1 = big.NewInt(0)
-
-			operation := types.Remove
-			args := []types.WatchAddressArg{
-				{
-					Address:   SLV1.Address,
-					CreatedAt: uint64(SLV1.BlockNumber),
-				},
-			}
-			ipldErr := ipldRPCClient.Call(nil, ipldMethod, operation, args)
-			Expect(ipldErr).ToNot(HaveOccurred())
-
-			// WatchedAddresses = [GLD1, SLV2]
-			// SLV1, countA
-			countA1Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV1.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA1 := new(big.Int).SetBytes(countA1Storage)
-			Expect(countA1.String()).To(Equal(updatedCountA1.String()))
-			oldCountA1.Set(updatedCountA1)
-		})
-
-		It("indexes state for redeployed contract", func() {
-			// Redeploy contract
-			SLV1, contractErr = integration.Create2Contract("SLVToken", contract1Salt)
-			Expect(contractErr).ToNot(HaveOccurred())
-
-			// Add to watched address
-			operation := types.Add
-			args := []types.WatchAddressArg{
-				{
-					Address:   SLV1.Address,
-					CreatedAt: uint64(SLV1.BlockNumber),
-				},
-			}
-			ipldErr := ipldRPCClient.Call(nil, ipldMethod, operation, args)
-			Expect(ipldErr).ToNot(HaveOccurred())
-
-			// Sleep for service interval + few extra seconds
-			time.Sleep(time.Duration(serviceInterval+2) * time.Second)
-
-			// WatchedAddresses = [GLD1, SLV2, SLV1]
-			// SLV1, countA
-			countA1Storage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV1.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA1 := new(big.Int).SetBytes(countA1Storage)
-			Expect(countA1.String()).To(Equal(updatedCountA1.String()))
-			oldCountA1.Set(updatedCountA1)
-
-			// Update storage slots
-			_, txErr = integration.IncrementCount(SLV1.Address, "A")
-			time.Sleep(sleepInterval)
-			Expect(txErr).ToNot(HaveOccurred())
-			updatedCountA1.Add(updatedCountA1, big.NewInt(1))
-
-			countA1AfterIncrementStorage, err := ipldClient.StorageAt(ctx, common.HexToAddress(SLV1.Address), common.HexToHash(countAIndex), nil)
-			Expect(err).ToNot(HaveOccurred())
-			countA1AfterIncrement := new(big.Int).SetBytes(countA1AfterIncrementStorage)
-			Expect(countA1AfterIncrement.String()).To(Equal(updatedCountA1.String()))
-			oldCountA1.Set(updatedCountA1)
-		})
-	})
-})