diff --git a/statediff/README.md b/statediff/README.md index 0d8163e02..74c82f2d2 100644 --- a/statediff/README.md +++ b/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 diff --git a/statediff/builder_test.go b/statediff/builder_test.go index a99a39664..916652256 100644 --- a/statediff/builder_test.go +++ b/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'}, diff --git a/statediff/indexer/helpers.go b/statediff/indexer/helpers.go index bb62fd079..0b3a9287d 100644 --- a/statediff/indexer/helpers.go +++ b/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 diff --git a/statediff/indexer/indexer.go b/statediff/indexer/indexer.go index 761bd3c29..7f6fe0dd1 100644 --- a/statediff/indexer/indexer.go +++ b/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) diff --git a/statediff/indexer/indexer_legacy_test.go b/statediff/indexer/indexer_legacy_test.go index 588107cc0..005f3e445 100644 --- a/statediff/indexer/indexer_legacy_test.go +++ b/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, diff --git a/statediff/indexer/indexer_test.go b/statediff/indexer/indexer_test.go index 70ab4cd90..725c69781 100644 --- a/statediff/indexer/indexer_test.go +++ b/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, diff --git a/statediff/indexer/shared/functions.go b/statediff/indexer/shared/functions.go index 92d5e6f2f..cb2ca6cea 100644 --- a/statediff/indexer/shared/functions.go +++ b/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 +} diff --git a/statediff/service.go b/statediff/service.go index 7935c4887..2d5f8b0ab 100644 --- a/statediff/service.go +++ b/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 { diff --git a/statediff/types/types.go b/statediff/types/types.go index 08e2124fa..b3acdb743 100644 --- a/statediff/types/types.go +++ b/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"`