Merge pull request #45 from vulcanize/ng-merge-fix-iterator

Merge changes to fix iterator and update

.github/workflows/issues-notion-sync.yml

@@ -1,29 +0,0 @@
-name: Notion Sync
-
-on:
-  workflow_dispatch:
-  issues:
-    types:
-      [
-        opened,
-        edited,
-        labeled,
-        unlabeled,
-        assigned,
-        unassigned,
-        milestoned,
-        demilestoned,
-        reopened,
-        closed,
-      ]
-
-jobs:
-  notion_job:
-    runs-on: ubuntu-latest
-    name: Add GitHub Issues to Notion
-    steps:
-      - name: Add GitHub Issues to Notion
-        uses: vulcanize/notion-github-action@v1.2.4-issueid
-        with:
-          notion-token: ${{ secrets.NOTION_TOKEN }}
-          notion-db: ${{ secrets.NOTION_DATABASE }}

go.mod

@@ -114,4 +114,4 @@ require (
 	lukechampine.com/blake3 v1.1.7 // indirect
 )
 
-replace github.com/ethereum/go-ethereum v1.10.18 => github.com/vulcanize/go-ethereum v1.10.18-statediff-3.2.2
+replace github.com/ethereum/go-ethereum v1.10.18 => github.com/vulcanize/go-ethereum v1.10.18-statediff-4.0.2-alpha

go.sum

@@ -778,8 +778,8 @@ github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPU
 github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
 github.com/vulcanize/go-eth-state-node-iterator v1.1.0 h1:wbOpP9BnqDcN8OWbuOw/QvR7Mde5JuuI15SGfLil0ZE=
 github.com/vulcanize/go-eth-state-node-iterator v1.1.0/go.mod h1:a1aRW1/fEmcpY8AAU47p9TNL3QNjpeqCJd/hLtFKVL8=
-github.com/vulcanize/go-ethereum v1.10.18-statediff-3.2.2 h1:hAZFOtnfdHtVgiwtcNFZDxCEN01c4W5Xsw+1K/IlHBA=
-github.com/vulcanize/go-ethereum v1.10.18-statediff-3.2.2/go.mod h1:HelXH7UT1uWdb+St6UAj4pPf93GOggjIV7pVbrWIZ3o=
+github.com/vulcanize/go-ethereum v1.10.18-statediff-4.0.2-alpha h1:gTcfBHoaRpsRbP1/dnurg5LFUaqsQMTNqs4R4S2wm9U=
+github.com/vulcanize/go-ethereum v1.10.18-statediff-4.0.2-alpha/go.mod h1:HelXH7UT1uWdb+St6UAj4pPf93GOggjIV7pVbrWIZ3o=
 github.com/whyrusleeping/go-logging v0.0.0-20170515211332-0457bb6b88fc/go.mod h1:bopw91TMyo8J3tvftk8xmU2kPmlrt4nScJQZU2hE5EM=
 github.com/willf/bitset v1.1.3/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=

pkg/snapshot/file/publisher.go

@@ -18,6 +18,7 @@ package publisher
 import (
 	"encoding/csv"
 	"fmt"
+	"math/big"
 	"os"
 	"path/filepath"
 	"sync/atomic"
@@ -160,20 +161,20 @@ func (p *publisher) BeginTx() (snapt.Tx, error) {
 
 // PublishRaw derives a cid from raw bytes and provided codec and multihash type, and writes it to the db tx
 // returns the CID and blockstore prefixed multihash key
-func (tx fileWriters) publishRaw(codec uint64, raw []byte) (cid, prefixedKey string, err error) {
+func (tx fileWriters) publishRaw(codec uint64, raw []byte, height *big.Int) (cid, prefixedKey string, err error) {
 	c, err := ipld.RawdataToCid(codec, raw, multihash.KECCAK_256)
 	if err != nil {
 		return
 	}
 	cid = c.String()
-	prefixedKey, err = tx.publishIPLD(c, raw)
+	prefixedKey, err = tx.publishIPLD(c, raw, height)
 	return
 }
 
-func (tx fileWriters) publishIPLD(c cid.Cid, raw []byte) (string, error) {
+func (tx fileWriters) publishIPLD(c cid.Cid, raw []byte, height *big.Int) (string, error) {
 	dbKey := dshelp.MultihashToDsKey(c.Hash())
 	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
-	return prefixedKey, tx.write(&snapt.TableIPLDBlock, prefixedKey, raw)
+	return prefixedKey, tx.write(&snapt.TableIPLDBlock, height.String(), prefixedKey, raw)
 }
 
 // PublishHeader writes the header to the ipfs backing pg datastore and adds secondary
@@ -183,7 +184,7 @@ func (p *publisher) PublishHeader(header *types.Header) error {
 	if err != nil {
 		return err
 	}
-	if _, err = p.writers.publishIPLD(headerNode.Cid(), headerNode.RawData()); err != nil {
+	if _, err = p.writers.publishIPLD(headerNode.Cid(), headerNode.RawData(), header.Number); err != nil {
 		return err
 	}
 
@@ -205,19 +206,19 @@ func (p *publisher) PublishHeader(header *types.Header) error {
 
 // PublishStateNode writes the state node to the ipfs backing datastore and adds secondary indexes
 // in the state_cids table
-func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, snapTx snapt.Tx) error {
+func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, height *big.Int, snapTx snapt.Tx) error {
 	var stateKey string
 	if !snapt.IsNullHash(node.Key) {
 		stateKey = node.Key.Hex()
 	}
 
 	tx := snapTx.(fileTx)
-	stateCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStateTrie, node.Value)
+	stateCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStateTrie, node.Value, height)
 	if err != nil {
 		return err
 	}
 
-	err = tx.write(&snapt.TableStateNode, headerID, stateKey, stateCIDStr, node.Path,
+	err = tx.write(&snapt.TableStateNode, height.String(), headerID, stateKey, stateCIDStr, node.Path,
 		node.NodeType, false, mhKey)
 	if err != nil {
 		return err
@@ -233,19 +234,19 @@ func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, snapTx s
 
 // PublishStorageNode writes the storage node to the ipfs backing pg datastore and adds secondary
 // indexes in the storage_cids table
-func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, statePath []byte, snapTx snapt.Tx) error {
+func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, height *big.Int, statePath []byte, snapTx snapt.Tx) error {
 	var storageKey string
 	if !snapt.IsNullHash(node.Key) {
 		storageKey = node.Key.Hex()
 	}
 
 	tx := snapTx.(fileTx)
-	storageCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStorageTrie, node.Value)
+	storageCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStorageTrie, node.Value, height)
 	if err != nil {
 		return err
 	}
 
-	err = tx.write(&snapt.TableStorageNode, headerID, statePath, storageKey, storageCIDStr, node.Path,
+	err = tx.write(&snapt.TableStorageNode, height.String(), headerID, statePath, storageKey, storageCIDStr, node.Path,
 		node.NodeType, false, mhKey)
 	if err != nil {
 		return err
@@ -260,7 +261,7 @@ func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, stateP
 }
 
 // PublishCode writes code to the ipfs backing pg datastore
-func (p *publisher) PublishCode(codeHash common.Hash, codeBytes []byte, snapTx snapt.Tx) error {
+func (p *publisher) PublishCode(height *big.Int, codeHash common.Hash, codeBytes []byte, snapTx snapt.Tx) error {
 	// no codec for code, doesn't matter though since blockstore key is multihash-derived
 	mhKey, err := shared.MultihashKeyFromKeccak256(codeHash)
 	if err != nil {
@@ -268,7 +269,7 @@ func (p *publisher) PublishCode(codeHash common.Hash, codeBytes []byte, snapTx s
 	}
 
 	tx := snapTx.(fileTx)
-	if err = tx.write(&snapt.TableIPLDBlock, mhKey, codeBytes); err != nil {
+	if err = tx.write(&snapt.TableIPLDBlock, height.String(), mhKey, codeBytes); err != nil {
 		return fmt.Errorf("error publishing code IPLD: %v", err)
 	}
 	// increment code node counter.

pkg/snapshot/file/publisher_test.go

@@ -38,7 +38,7 @@ func writeFiles(t *testing.T, dir string) *publisher {
 	test.NoError(t, err)
 
 	headerID := fixt.Block1_Header.Hash().String()
-	test.NoError(t, pub.PublishStateNode(&fixt.Block1_StateNode0, headerID, tx))
+	test.NoError(t, pub.PublishStateNode(&fixt.Block1_StateNode0, headerID, fixt.Block1_Header.Number, tx))
 
 	test.NoError(t, tx.Commit())
 	return pub

pkg/snapshot/pg/publisher.go

@@ -18,6 +18,7 @@ package pg
 import (
 	"context"
 	"fmt"
+	"math/big"
 	"sync/atomic"
 	"time"
 
@@ -88,20 +89,20 @@ func (p *publisher) BeginTx() (snapt.Tx, error) {
 
 // PublishRaw derives a cid from raw bytes and provided codec and multihash type, and writes it to the db tx
 // returns the CID and blockstore prefixed multihash key
-func (tx pubTx) publishRaw(codec uint64, raw []byte) (cid, prefixedKey string, err error) {
+func (tx pubTx) publishRaw(codec uint64, raw []byte, height *big.Int) (cid, prefixedKey string, err error) {
 	c, err := ipld.RawdataToCid(codec, raw, multihash.KECCAK_256)
 	if err != nil {
 		return
 	}
 	cid = c.String()
-	prefixedKey, err = tx.publishIPLD(c, raw)
+	prefixedKey, err = tx.publishIPLD(c, raw, height)
 	return
 }
 
-func (tx pubTx) publishIPLD(c cid.Cid, raw []byte) (string, error) {
+func (tx pubTx) publishIPLD(c cid.Cid, raw []byte, height *big.Int) (string, error) {
 	dbKey := dshelp.MultihashToDsKey(c.Hash())
 	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
-	_, err := tx.Exec(snapt.TableIPLDBlock.ToInsertStatement(), prefixedKey, raw)
+	_, err := tx.Exec(snapt.TableIPLDBlock.ToInsertStatement(), height.Uint64(), prefixedKey, raw)
 	return prefixedKey, err
 }
 
@@ -119,7 +120,7 @@ func (p *publisher) PublishHeader(header *types.Header) (err error) {
 	tx := pubTx{snapTx, nil}
 	defer func() { err = snapt.CommitOrRollback(tx, err) }()
 
-	if _, err = tx.publishIPLD(headerNode.Cid(), headerNode.RawData()); err != nil {
+	if _, err = tx.publishIPLD(headerNode.Cid(), headerNode.RawData(), header.Number); err != nil {
 		return err
 	}
 
@@ -132,20 +133,20 @@ func (p *publisher) PublishHeader(header *types.Header) (err error) {
 }
 
 // PublishStateNode writes the state node to the ipfs backing datastore and adds secondary indexes in the state_cids table
-func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, snapTx snapt.Tx) error {
+func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, height *big.Int, snapTx snapt.Tx) error {
 	var stateKey string
 	if !snapt.IsNullHash(node.Key) {
 		stateKey = node.Key.Hex()
 	}
 
 	tx := snapTx.(pubTx)
-	stateCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStateTrie, node.Value)
+	stateCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStateTrie, node.Value, height)
 	if err != nil {
 		return err
 	}
 
 	_, err = tx.Exec(snapt.TableStateNode.ToInsertStatement(),
-		headerID, stateKey, stateCIDStr, node.Path, node.NodeType, false, mhKey)
+		height.Uint64(), headerID, stateKey, stateCIDStr, node.Path, node.NodeType, false, mhKey)
 	if err != nil {
 		return err
 	}
@@ -159,20 +160,20 @@ func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, snapTx s
 }
 
 // PublishStorageNode writes the storage node to the ipfs backing pg datastore and adds secondary indexes in the storage_cids table
-func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, statePath []byte, snapTx snapt.Tx) error {
+func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, height *big.Int, statePath []byte, snapTx snapt.Tx) error {
 	var storageKey string
 	if !snapt.IsNullHash(node.Key) {
 		storageKey = node.Key.Hex()
 	}
 
 	tx := snapTx.(pubTx)
-	storageCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStorageTrie, node.Value)
+	storageCIDStr, mhKey, err := tx.publishRaw(ipld.MEthStorageTrie, node.Value, height)
 	if err != nil {
 		return err
 	}
 
 	_, err = tx.Exec(snapt.TableStorageNode.ToInsertStatement(),
-		headerID, statePath, storageKey, storageCIDStr, node.Path, node.NodeType, false, mhKey)
+		height.Uint64(), headerID, statePath, storageKey, storageCIDStr, node.Path, node.NodeType, false, mhKey)
 	if err != nil {
 		return err
 	}
@@ -186,7 +187,7 @@ func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, stateP
 }
 
 // PublishCode writes code to the ipfs backing pg datastore
-func (p *publisher) PublishCode(codeHash common.Hash, codeBytes []byte, snapTx snapt.Tx) error {
+func (p *publisher) PublishCode(height *big.Int, codeHash common.Hash, codeBytes []byte, snapTx snapt.Tx) error {
 	// no codec for code, doesn't matter though since blockstore key is multihash-derived
 	mhKey, err := shared.MultihashKeyFromKeccak256(codeHash)
 	if err != nil {
@@ -194,7 +195,7 @@ func (p *publisher) PublishCode(codeHash common.Hash, codeBytes []byte, snapTx s
 	}
 
 	tx := snapTx.(pubTx)
-	if _, err = tx.Exec(snapt.TableIPLDBlock.ToInsertStatement(), mhKey, codeBytes); err != nil {
+	if _, err = tx.Exec(snapt.TableIPLDBlock.ToInsertStatement(), height.Uint64(), mhKey, codeBytes); err != nil {
 		return fmt.Errorf("error publishing code IPLD: %v", err)
 	}
 

pkg/snapshot/pg/publisher_test.go

@@ -36,7 +36,7 @@ func writeData(t *testing.T) *publisher {
 	test.NoError(t, err)
 
 	headerID := fixt.Block1_Header.Hash().String()
-	test.NoError(t, pub.PublishStateNode(&fixt.Block1_StateNode0, headerID, tx))
+	test.NoError(t, pub.PublishStateNode(&fixt.Block1_StateNode0, headerID, fixt.Block1_Header.Number, tx))
 
 	test.NoError(t, tx.Commit())
 	return pub
@@ -73,6 +73,7 @@ func TestBasic(t *testing.T) {
 	test.NoError(t, err)
 
 	headerNode, err := ipld.NewEthHeader(&fixt.Block1_Header)
+	test.NoError(t, err)
 	test.ExpectEqual(t, headerNode.Cid().String(), header.CID)
 	test.ExpectEqual(t, fixt.Block1_Header.Hash().String(), header.BlockHash)
 }

pkg/snapshot/service.go

@@ -19,6 +19,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"math/big"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -142,9 +143,9 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 	}()
 
 	if len(iters) > 0 {
-		return s.createSnapshotAsync(iters, headerID)
+		return s.createSnapshotAsync(iters, headerID, new(big.Int).SetUint64(params.Height))
 	} else {
-		return s.createSnapshot(iters[0], headerID)
+		return s.createSnapshot(iters[0], headerID, new(big.Int).SetUint64(params.Height))
 	}
 }
 
@@ -197,7 +198,7 @@ func resolveNode(it trie.NodeIterator, trieDB *trie.Database) (*nodeResult, erro
 	}, nil
 }
 
-func (s *Service) createSnapshot(it trie.NodeIterator, headerID string) error {
+func (s *Service) createSnapshot(it trie.NodeIterator, headerID string, height *big.Int) error {
 	tx, err := s.ipfsPublisher.BeginTx()
 	if err != nil {
 		return err
@@ -232,7 +233,7 @@ func (s *Service) createSnapshot(it trie.NodeIterator, headerID string) error {
 			encodedPath := trie.HexToCompact(valueNodePath)
 			leafKey := encodedPath[1:]
 			res.node.Key = common.BytesToHash(leafKey)
-			err := s.ipfsPublisher.PublishStateNode(&res.node, headerID, tx)
+			err := s.ipfsPublisher.PublishStateNode(&res.node, headerID, height, tx)
 			if err != nil {
 				return err
 			}
@@ -246,17 +247,17 @@ func (s *Service) createSnapshot(it trie.NodeIterator, headerID string) error {
 					return errors.New("missing code")
 				}
 
-				if err = s.ipfsPublisher.PublishCode(codeHash, codeBytes, tx); err != nil {
+				if err = s.ipfsPublisher.PublishCode(height, codeHash, codeBytes, tx); err != nil {
 					return err
 				}
 			}
 
-			if tx, err = s.storageSnapshot(account.Root, headerID, res.node.Path, tx); err != nil {
+			if tx, err = s.storageSnapshot(account.Root, headerID, height, res.node.Path, tx); err != nil {
 				return fmt.Errorf("failed building storage snapshot for account %+v\r\nerror: %w", account, err)
 			}
 		case Extension, Branch:
 			res.node.Key = common.BytesToHash([]byte{})
-			if err := s.ipfsPublisher.PublishStateNode(&res.node, headerID, tx); err != nil {
+			if err := s.ipfsPublisher.PublishStateNode(&res.node, headerID, height, tx); err != nil {
 				return err
 			}
 		default:
@@ -267,14 +268,14 @@ func (s *Service) createSnapshot(it trie.NodeIterator, headerID string) error {
 }
 
 // Full-trie concurrent snapshot
-func (s *Service) createSnapshotAsync(iters []trie.NodeIterator, headerID string) error {
+func (s *Service) createSnapshotAsync(iters []trie.NodeIterator, headerID string, height *big.Int) error {
 	errors := make(chan error)
 	var wg sync.WaitGroup
 	for _, it := range iters {
 		wg.Add(1)
 		go func(it trie.NodeIterator) {
 			defer wg.Done()
-			if err := s.createSnapshot(it, headerID); err != nil {
+			if err := s.createSnapshot(it, headerID, height); err != nil {
 				errors <- err
 			}
 		}(it)
@@ -295,7 +296,7 @@ func (s *Service) createSnapshotAsync(iters []trie.NodeIterator, headerID string
 	return err
 }
 
-func (s *Service) storageSnapshot(sr common.Hash, headerID string, statePath []byte, tx Tx) (Tx, error) {
+func (s *Service) storageSnapshot(sr common.Hash, headerID string, height *big.Int, statePath []byte, tx Tx) (Tx, error) {
 	if bytes.Equal(sr.Bytes(), emptyContractRoot.Bytes()) {
 		return tx, nil
 	}
@@ -339,7 +340,7 @@ func (s *Service) storageSnapshot(sr common.Hash, headerID string, statePath []b
 		default:
 			return nil, errors.New("unexpected node type")
 		}
-		if err = s.ipfsPublisher.PublishStorageNode(&res.node, headerID, statePath, tx); err != nil {
+		if err = s.ipfsPublisher.PublishStorageNode(&res.node, headerID, height, statePath, tx); err != nil {
 			return nil, err
 		}
 	}

pkg/snapshot/service_test.go

@@ -2,6 +2,7 @@ package snapshot
 
 import (
 	"errors"
+	"math/big"
 	"os"
 	"path/filepath"
 	"testing"
@@ -43,7 +44,7 @@ func TestCreateSnapshot(t *testing.T) {
 			Times(workers)
 		pub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).
 			AnyTimes()
-		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any()).
+		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 			// Use MinTimes as duplicate nodes are expected at boundaries
 			MinTimes(len(fixt.Block1_StateNodePaths))
 
@@ -79,7 +80,7 @@ func TestCreateSnapshot(t *testing.T) {
 	}
 }
 
-func failingPublishStateNode(_ *snapt.Node, _ string, _ snapt.Tx) error {
+func failingPublishStateNode(_ *snapt.Node, _ string, _ *big.Int, _ snapt.Tx) error {
 	return errors.New("failingPublishStateNode")
 }
 
@@ -89,7 +90,7 @@ func TestRecovery(t *testing.T) {
 		pub.EXPECT().PublishHeader(gomock.Any()).AnyTimes()
 		pub.EXPECT().BeginTx().Return(tx, nil).AnyTimes()
 		pub.EXPECT().PrepareTxForBatch(gomock.Any(), gomock.Any()).Return(tx, nil).AnyTimes()
-		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any()).
+		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 			Times(workers).
 			DoAndReturn(failingPublishStateNode)
 		tx.EXPECT().Commit().AnyTimes()
@@ -117,7 +118,7 @@ func TestRecovery(t *testing.T) {
 			t.Fatal("cannot stat recovery file:", err)
 		}
 
-		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
+		pub.EXPECT().PublishStateNode(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 		err = service.CreateSnapshot(params)
 		if err != nil {
 			t.Fatal(err)

pkg/types/publisher.go

@@ -1,15 +1,17 @@
 package types
 
 import (
+	"math/big"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
 type Publisher interface {
 	PublishHeader(header *types.Header) error
-	PublishStateNode(node *Node, headerID string, tx Tx) error
-	PublishStorageNode(node *Node, headerID string, statePath []byte, tx Tx) error
-	PublishCode(codeHash common.Hash, codeBytes []byte, tx Tx) error
+	PublishStateNode(node *Node, headerID string, height *big.Int, tx Tx) error
+	PublishStorageNode(node *Node, headerID string, height *big.Int, statePath []byte, tx Tx) error
+	PublishCode(height *big.Int, codeHash common.Hash, codeBytes []byte, tx Tx) error
 	BeginTx() (Tx, error)
 	PrepareTxForBatch(tx Tx, batchSize uint) (Tx, error)
 }

pkg/types/schema.go

@@ -3,10 +3,11 @@ package types
 var TableIPLDBlock = Table{
 	`public.blocks`,
 	[]column{
+		{"block_number", bigint},
 		{"key", text},
 		{"data", bytea},
 	},
-	`ON CONFLICT (key) DO NOTHING`,
+	"ON CONFLICT (key, block_number) DO NOTHING",
 }
 
 var TableNodeInfo = Table{
@@ -40,12 +41,13 @@ var TableHeader = Table{
 		{"times_validated", integer},
 		{"coinbase", varchar},
 	},
-	"ON CONFLICT (block_hash) DO UPDATE SET (parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase) = (EXCLUDED.parent_hash, EXCLUDED.cid, EXCLUDED.td, EXCLUDED.node_id, EXCLUDED.reward, EXCLUDED.state_root, EXCLUDED.tx_root, EXCLUDED.receipt_root, EXCLUDED.uncle_root, EXCLUDED.bloom, EXCLUDED.timestamp, EXCLUDED.mh_key, eth.header_cids.times_validated + 1, EXCLUDED.coinbase)",
+	"ON CONFLICT (block_hash, block_number) DO UPDATE SET (parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase) = (EXCLUDED.parent_hash, EXCLUDED.cid, EXCLUDED.td, EXCLUDED.node_id, EXCLUDED.reward, EXCLUDED.state_root, EXCLUDED.tx_root, EXCLUDED.receipt_root, EXCLUDED.uncle_root, EXCLUDED.bloom, EXCLUDED.timestamp, EXCLUDED.mh_key, eth.header_cids.times_validated + 1, EXCLUDED.coinbase)",
 }
 
 var TableStateNode = Table{
 	"eth.state_cids",
 	[]column{
+		{"block_number", bigint},
 		{"header_id", varchar},
 		{"state_leaf_key", varchar},
 		{"cid", text},
@@ -54,12 +56,13 @@ var TableStateNode = Table{
 		{"diff", boolean},
 		{"mh_key", text},
 	},
-	`ON CONFLICT (header_id, state_path) DO UPDATE SET (state_leaf_key, cid, node_type, diff, mh_key) = (EXCLUDED.state_leaf_key, EXCLUDED.cid, EXCLUDED.node_type, EXCLUDED.diff, EXCLUDED.mh_key)`,
+	"ON CONFLICT (header_id, state_path, block_number) DO UPDATE SET (state_leaf_key, cid, node_type, diff, mh_key) = (EXCLUDED.state_leaf_key, EXCLUDED.cid, EXCLUDED.node_type, EXCLUDED.diff, EXCLUDED.mh_key)",
 }
 
 var TableStorageNode = Table{
 	"eth.storage_cids",
 	[]column{
+		{"block_number", bigint},
 		{"header_id", varchar},
 		{"state_path", bytea},
 		{"storage_leaf_key", varchar},
@@ -69,5 +72,5 @@ var TableStorageNode = Table{
 		{"diff", boolean},
 		{"mh_key", text},
 	},
-	"ON CONFLICT (header_id, state_path, storage_path) DO UPDATE SET (storage_leaf_key, cid, node_type, diff, mh_key) = (EXCLUDED.storage_leaf_key, EXCLUDED.cid, EXCLUDED.node_type, EXCLUDED.diff, EXCLUDED.mh_key)",
+	"ON CONFLICT (header_id, state_path, storage_path, block_number) DO UPDATE SET (storage_leaf_key, cid, node_type, diff, mh_key) = (EXCLUDED.storage_leaf_key, EXCLUDED.cid, EXCLUDED.node_type, EXCLUDED.diff, EXCLUDED.mh_key)",
 }