only write Removed node ipld block (on db init) and reuse constant cid and mhkey

statediff/README.md

@@ -81,6 +81,7 @@ This service introduces a CLI flag namespace `statediff`
 `--statediff.writing` is used to tell the service to write state diff objects it produces from synced ChainEvents directly to a configured Postgres database
 `--statediff.workers` is used to set the number of concurrent workers to process state diff objects and write them into the database
 `--statediff.db` is the connection string for the Postgres database to write to
+`--statediff.db.init` indicates whether we need to initialize a new database; set true if its the first time running the process on a given database
 `--statediff.dbnodeid` is the node id to use in the Postgres database
 `--statediff.dbclientname` is the client name to use in the Postgres database
 
@@ -88,7 +89,7 @@ The service can only operate in full sync mode (`--syncmode=full`), but only the
 
 e.g.
 `
-./build/bin/geth --syncmode=full --gcmode=archive --statediff --statediff.writing --statediff.db=postgres://localhost:5432/vulcanize_testing?sslmode=disable --statediff.dbnodeid={nodeId} --statediff.dbclientname={dbClientName}
+./build/bin/geth --syncmode=full --gcmode=archive --statediff --statediff.writing --statediff.db=postgres://localhost:5432/vulcanize_testing?sslmode=disable --statediff.db.init=true --statediff.dbnodeid={nodeId} --statediff.dbclientname={dbClientName}
 `
 
 ### RPC endpoints

statediff/builder_test.go

@@ -1485,6 +1485,7 @@ func TestBuilderWithRemovedAccountAndStorage(t *testing.T) {
 						NodeType:  sdtypes.Removed,
 						LeafKey:   contractLeafKey,
 						NodeValue: []byte{},
+						StorageNodes: emptyStorage,
 					},
 					{
 						Path:         []byte{'\x0c'},

statediff/indexer/helpers.go

@@ -23,19 +23,9 @@ import (
 	"github.com/ethereum/go-ethereum/statediff/types"
 )
 
+// ResolveFromNodeType wrapper around NodeType.Int() so that we maintain backwards compatability
 func ResolveFromNodeType(nodeType types.NodeType) int {
-	switch nodeType {
-	case types.Branch:
-		return 0
-	case types.Extension:
-		return 1
-	case types.Leaf:
-		return 2
-	case types.Removed:
-		return 3
-	default:
-		return -1
-	}
+	return nodeType.Int()
 }
 
 // ChainConfig returns the appropriate ethereum chain config for the provided chain id

statediff/indexer/indexer.go

@@ -47,6 +47,12 @@ var (
 	dbMetrics      = RegisterDBMetrics(metrics.DefaultRegistry)
 )
 
+const (
+	removedNodeStorageCID = "bagmacgzayxjemamg64rtzet6pwznzrydydsqbnstzkbcoo337lmaixmfurya"
+	removedNodeStateCID = "baglacgzayxjemamg64rtzet6pwznzrydydsqbnstzkbcoo337lmaixmfurya"
+	removedNodeMhKey = "/blocks/DMQMLUSGAGDPOIZ4SJ7H3MW4Y4B4BZIAWZJ4VARHHN57VWAELWC2I4A"
+)
+
 // Indexer interface to allow substitution of mocks for testing
 type Indexer interface {
 	PushBlock(block *types.Block, receipts types.Receipts, totalDifficulty *big.Int) (*BlockTx, error)
@@ -59,14 +65,21 @@ type Indexer interface {
 type StateDiffIndexer struct {
 	chainConfig *params.ChainConfig
 	dbWriter    *PostgresCIDWriter
+	init bool
 }
 
 // NewStateDiffIndexer creates a pointer to a new PayloadConverter which satisfies the PayloadConverter interface
-func NewStateDiffIndexer(chainConfig *params.ChainConfig, db *postgres.DB) *StateDiffIndexer {
+func NewStateDiffIndexer(chainConfig *params.ChainConfig, db *postgres.DB, init bool) (*StateDiffIndexer, error) {
+	// If this is the first time writing to this db, write the public.blocks entry for an empty node (for Removed state and storage node types)
+	if init {
+		if err := shared.PublishDirectWithDB(db, removedNodeMhKey, []byte{}); err != nil {
+			return nil, err
+		}
+	}
 	return &StateDiffIndexer{
 		chainConfig: chainConfig,
 		dbWriter:    NewPostgresCIDWriter(db),
-	}
+	}, nil
 }
 
 type BlockTx struct {
@@ -76,7 +89,7 @@ type BlockTx struct {
 	Close       func(err error) error
 }
 
-// Reporting function to run as goroutine
+// ReportDBMetrics is a reporting function to run as goroutine
 func (sdi *StateDiffIndexer) ReportDBMetrics(delay time.Duration, quit <-chan bool) {
 	if !metrics.Enabled {
 		return
@@ -95,7 +108,7 @@ func (sdi *StateDiffIndexer) ReportDBMetrics(delay time.Duration, quit <-chan bo
 	}()
 }
 
-// Pushes and indexes block data in database, except state & storage nodes (includes header, uncles, transactions & receipts)
+// PushBlock pushes and indexes block data in database, except state & storage nodes (includes header, uncles, transactions & receipts)
 // Returns an initiated DB transaction which must be Closed via defer to commit or rollback
 func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receipts, totalDifficulty *big.Int) (*BlockTx, error) {
 	start, t := time.Now(), time.Now()
@@ -250,6 +263,7 @@ func (sdi *StateDiffIndexer) processHeader(tx *sqlx.Tx, header *types.Header, he
 	})
 }
 
+// processUncles publishes and indexes uncle IPLDs in Postgres
 func (sdi *StateDiffIndexer) processUncles(tx *sqlx.Tx, headerID int64, blockNumber uint64, uncleNodes []*ipld.EthHeader) error {
 	// publish and index uncles
 	for _, uncleNode := range uncleNodes {
@@ -434,19 +448,32 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *sqlx.Tx, args processArgs
 	return nil
 }
 
+// PushStateNode publishes and indexes a state diff node object (including any child storage nodes) in the IPLD database
 func (sdi *StateDiffIndexer) PushStateNode(tx *BlockTx, stateNode sdtypes.StateNode) error {
 	// publish the state node
-	stateCIDStr, err := shared.PublishRaw(tx.dbtx, ipld.MEthStateTrie, multihash.KECCAK_256, stateNode.NodeValue)
+	if stateNode.NodeType == sdtypes.Removed {
+		// short circuit if it is a Removed node
+		// this assumes the db has been initialized and a public.blocks entry for the Removed node is present
+		stateModel := models.StateNodeModel{
+			Path:     stateNode.Path,
+			StateKey: common.BytesToHash(stateNode.LeafKey).String(),
+			CID:      removedNodeStateCID,
+			MhKey:    removedNodeMhKey,
+			NodeType: stateNode.NodeType.Int(),
+		}
+		_, err := sdi.dbWriter.upsertStateCID(tx.dbtx, stateModel, tx.headerID)
+		return err
+	}
+	stateCIDStr, stateMhKey, err := shared.PublishRaw(tx.dbtx, ipld.MEthStateTrie, multihash.KECCAK_256, stateNode.NodeValue)
 	if err != nil {
 		return fmt.Errorf("error publishing state node IPLD: %v", err)
 	}
-	mhKey, _ := shared.MultihashKeyFromCIDString(stateCIDStr)
 	stateModel := models.StateNodeModel{
 		Path:     stateNode.Path,
 		StateKey: common.BytesToHash(stateNode.LeafKey).String(),
 		CID:      stateCIDStr,
-		MhKey:    mhKey,
-		NodeType: ResolveFromNodeType(stateNode.NodeType),
+		MhKey:    stateMhKey,
+		NodeType: stateNode.NodeType.Int(),
 	}
 	// index the state node, collect the stateID to reference by FK
 	stateID, err := sdi.dbWriter.upsertStateCID(tx.dbtx, stateModel, tx.headerID)
@@ -478,17 +505,31 @@ func (sdi *StateDiffIndexer) PushStateNode(tx *BlockTx, stateNode sdtypes.StateN
 	}
 	// if there are any storage nodes associated with this node, publish and index them
 	for _, storageNode := range stateNode.StorageNodes {
-		storageCIDStr, err := shared.PublishRaw(tx.dbtx, ipld.MEthStorageTrie, multihash.KECCAK_256, storageNode.NodeValue)
+		if storageNode.NodeType == sdtypes.Removed {
+			// short circuit if it is a Removed node
+			// this assumes the db has been initialized and a public.blocks entry for the Removed node is present
+			storageModel := models.StorageNodeModel{
+				Path:     storageNode.Path,
+				StorageKey: common.BytesToHash(storageNode.LeafKey).String(),
+				CID:      removedNodeStorageCID,
+				MhKey:    removedNodeMhKey,
+				NodeType: storageNode.NodeType.Int(),
+			}
+			if err := sdi.dbWriter.upsertStorageCID(tx.dbtx, storageModel, stateID); err != nil {
+				return err
+			}
+			continue
+		}
+		storageCIDStr, storageMhKey, err := shared.PublishRaw(tx.dbtx, ipld.MEthStorageTrie, multihash.KECCAK_256, storageNode.NodeValue)
 		if err != nil {
 			return fmt.Errorf("error publishing storage node IPLD: %v", err)
 		}
-		mhKey, _ := shared.MultihashKeyFromCIDString(storageCIDStr)
 		storageModel := models.StorageNodeModel{
 			Path:       storageNode.Path,
 			StorageKey: common.BytesToHash(storageNode.LeafKey).String(),
 			CID:        storageCIDStr,
-			MhKey:      mhKey,
-			NodeType:   ResolveFromNodeType(storageNode.NodeType),
+			MhKey:      storageMhKey,
+			NodeType:   storageNode.NodeType.Int(),
 		}
 		if err := sdi.dbWriter.upsertStorageCID(tx.dbtx, storageModel, stateID); err != nil {
 			return err
@@ -498,7 +539,7 @@ func (sdi *StateDiffIndexer) PushStateNode(tx *BlockTx, stateNode sdtypes.StateN
 	return nil
 }
 
-// Publishes code and codehash pairs to the ipld database
+// PushCodeAndCodeHash publishes code and codehash pairs to the ipld database
 func (sdi *StateDiffIndexer) PushCodeAndCodeHash(tx *BlockTx, codeAndCodeHash sdtypes.CodeAndCodeHash) error {
 	// codec doesn't matter since db key is multihash-based
 	mhKey, err := shared.MultihashKeyFromKeccak256(codeAndCodeHash.Hash)

statediff/indexer/indexer_legacy_test.go

@@ -42,7 +42,8 @@ func setupLegacy(t *testing.T) {
 	db, err = shared.SetupDB()
 	require.NoError(t, err)
 
-	ind = indexer.NewStateDiffIndexer(legacyData.Config, db)
+	ind, err = indexer.NewStateDiffIndexer(legacyData.Config, db, false)
+	require.NoError(t, err)
 	var tx *indexer.BlockTx
 	tx, err = ind.PushBlock(
 		mockLegacyBlock,

statediff/indexer/indexer_test.go

@@ -139,7 +139,8 @@ func setup(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	ind = indexer.NewStateDiffIndexer(mocks.TestConfig, db)
+	ind, err = indexer.NewStateDiffIndexer(mocks.TestConfig, db, false)
+	require.NoError(t, err)
 	var tx *indexer.BlockTx
 	tx, err = ind.PushBlock(
 		mockBlock,

statediff/indexer/shared/functions.go

@@ -20,6 +20,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipfs/ipld"
+	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
 
 	"github.com/ipfs/go-cid"
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
@@ -96,15 +97,16 @@ func MultihashKeyFromCIDString(c string) (string, error) {
 }
 
 // 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) {
+// returns the CID and blockstore prefixed multihash key
+func PublishRaw(tx *sqlx.Tx, codec, mh uint64, raw []byte) (string, string, error) {
 	c, err := ipld.RawdataToCid(codec, raw, mh)
 	if err != nil {
-		return "", err
+		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
+	return c.String(), prefixedKey, err
 }
 
 // MultihashKeyFromKeccak256 converts keccak256 hash bytes into a blockstore-prefixed multihash db key string
@@ -117,8 +119,14 @@ func MultihashKeyFromKeccak256(hash common.Hash) (string, error) {
 	return blockstore.BlockPrefix.String() + dbKey.String(), nil
 }
 
-// PublishDirect diretly writes a previously derived mhkey => value pair to the ipld database
+// PublishDirect diretly writes a previously derived mhkey => value pair to the ipld database in the provided tx
 func PublishDirect(tx *sqlx.Tx, key string, value []byte) error {
 	_, err := tx.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, key, value)
 	return err
 }
+
+// PublishDirectWithDB diretly writes a previously derived mhkey => value pair to the ipld database
+func PublishDirectWithDB(db *postgres.DB, key string, value []byte) error {
+	_, err := db.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, key, value)
+	return err
+}

statediff/service.go

@@ -165,7 +165,10 @@ func New(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params
 		if err != nil {
 			return err
 		}
-		indexer = ind.NewStateDiffIndexer(blockChain.Config(), db)
+		indexer, err = ind.NewStateDiffIndexer(blockChain.Config(), db, params.DBParams.Init)
+		if err != nil {
+			return err
+		}
 	}
 	workers := params.NumWorkers
 	if workers == 0 {

statediff/types/types.go

@@ -22,16 +22,32 @@ package types
 import "github.com/ethereum/go-ethereum/common"
 
 // NodeType for explicitly setting type of node
+// we use a string because it is RLP serializable, whereas an int is not
 type NodeType string
 
 const (
 	Unknown   NodeType = "Unknown"
-	Leaf      NodeType = "Leaf"
-	Extension NodeType = "Extension"
 	Branch    NodeType = "Branch"
+	Extension NodeType = "Extension"
+	Leaf      NodeType = "Leaf"
 	Removed   NodeType = "Removed" // used to represent pathes which have been emptied
 )
 
+func (n NodeType) Int() int {
+	switch n {
+	case Branch:
+		return 0
+	case Extension:
+		return 1
+	case Leaf:
+		return 2
+	case Removed:
+		return 3
+	default:
+		return -1
+	}
+}
+
 // StateNode holds the data for a single state diff node
 type StateNode struct {
 	NodeType     NodeType      `json:"nodeType"        gencodec:"required"`