cerc-io/go-ethereum
13
0
Fork 0
Code Issues 9 Pull Requests Packages Projects Releases 66 Wiki Activity

[WIP] V1.10.18 statediff 5.0.0 alpha #235

Closed
i-norden wants to merge 6 commits from v1.10.18-statediff-5.0.0-alpha into v1.10.18-statediff-v5
pull from: v1.10.18-statediff-5.0.0-alpha
merge into: cerc-io:v1.10.18-statediff-v5
Conversation 0 Commits 6 Files Changed 9 +195 -282
9 changed files with 195 additions and 282 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
statediff/indexer/database/file/indexer.go
View File

@ -27,7 +27,6 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/ipfs/go-cid"
node "github.com/ipfs/go-ipld-format" node "github.com/ipfs/go-ipld-format"
"github.com/multiformats/go-multihash" "github.com/multiformats/go-multihash"
pg_query "github.com/pganalyze/pg_query_go/v2" pg_query "github.com/pganalyze/pg_query_go/v2"
@ -124,16 +123,13 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
} }
// Generate the block iplds // Generate the block iplds
headerNode, uncleNodes, txNodes, txTrieNodes, rctNodes, rctTrieNodes, logTrieNodes, logLeafNodeCIDs, rctLeafNodeCIDs, err := ipld2.FromBlockAndReceipts(block, receipts) headerNode, uncleNodes, txNodes, rctNodes, logNodes, err := ipld2.FromBlockAndReceipts(block, receipts)
if err != nil { if err != nil {
return nil, fmt.Errorf("error creating IPLD nodes from block and receipts: %v", err) return nil, fmt.Errorf("error creating IPLD nodes from block and receipts: %v", err)
} }
if len(txNodes) != len(rctNodes) || len(rctNodes) != len(rctLeafNodeCIDs) { if len(txNodes) != len(rctNodes) {
return nil, fmt.Errorf("expected number of transactions (%d), receipts (%d), and receipt trie leaf nodes (%d) to be equal", len(txNodes), len(rctNodes), len(rctLeafNodeCIDs)) return nil, fmt.Errorf("expected number of transactions (%d) and receipts (%d) to be equal", len(txNodes), len(rctNodes))
}
if len(txTrieNodes) != len(rctTrieNodes) {
return nil, fmt.Errorf("expected number of tx trie (%d) and rct trie (%d) nodes to be equal", len(txTrieNodes), len(rctTrieNodes))
} }
// Calculate reward // Calculate reward
@ -188,12 +184,8 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
receipts: receipts, receipts: receipts,
txs: transactions, txs: transactions,
rctNodes: rctNodes, rctNodes: rctNodes,
rctTrieNodes: rctTrieNodes,
txNodes: txNodes, txNodes: txNodes,
txTrieNodes: txTrieNodes, logNodes: logNodes,
logTrieNodes: logTrieNodes,
logLeafNodeCIDs: logLeafNodeCIDs,
rctLeafNodeCIDs: rctLeafNodeCIDs,
}) })
if err != nil { if err != nil {
return nil, err return nil, err
@ -268,12 +260,8 @@ type processArgs struct {
receipts types.Receipts receipts types.Receipts
txs types.Transactions txs types.Transactions
rctNodes []*ipld2.EthReceipt rctNodes []*ipld2.EthReceipt
rctTrieNodes []*ipld2.EthRctTrie
txNodes []*ipld2.EthTx txNodes []*ipld2.EthTx
txTrieNodes []*ipld2.EthTxTrie logNodes [][]*ipld2.EthLog
logTrieNodes [][]node.Node
logLeafNodeCIDs [][]cid.Cid
rctLeafNodeCIDs []cid.Cid
} }
// processReceiptsAndTxs writes receipt and tx IPLD insert SQL stmts to a file // processReceiptsAndTxs writes receipt and tx IPLD insert SQL stmts to a file
@ -281,9 +269,6 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(args processArgs) error {
// Process receipts and txs // Process receipts and txs
signer := types.MakeSigner(sdi.chainConfig, args.blockNumber) signer := types.MakeSigner(sdi.chainConfig, args.blockNumber)
for i, receipt := range args.receipts { for i, receipt := range args.receipts {
for _, logTrieNode := range args.logTrieNodes[i] {
sdi.fileWriter.upsertIPLDNode(sdi.blockNumber, logTrieNode)
}
txNode := args.txNodes[i] txNode := args.txNodes[i]
sdi.fileWriter.upsertIPLDNode(sdi.blockNumber, txNode) sdi.fileWriter.upsertIPLDNode(sdi.blockNumber, txNode)
@ -301,6 +286,7 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(args processArgs) error {
if err != nil { if err != nil {
return fmt.Errorf("error deriving tx sender: %v", err) return fmt.Errorf("error deriving tx sender: %v", err)
} }
txCid := txNode.Cid()
txModel := models.TxModel{ txModel := models.TxModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: args.headerID, HeaderID: args.headerID,
@ -309,8 +295,8 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(args processArgs) error {
TxHash: txID, TxHash: txID,
Index: int64(i), Index: int64(i),
Data: trx.Data(), Data: trx.Data(),
CID: txNode.Cid().String(), CID: txCid.String(),
MhKey: shared.MultihashKeyFromCID(txNode.Cid()), MhKey: shared.MultihashKeyFromCID(txCid),
Type: trx.Type(), Type: trx.Type(),
Value: val, Value: val,
} }
@ -338,20 +324,14 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(args processArgs) error {
if contract != "" { if contract != "" {
contractHash = crypto.Keccak256Hash(common.HexToAddress(contract).Bytes()).String() contractHash = crypto.Keccak256Hash(common.HexToAddress(contract).Bytes()).String()
} }
rctCid := args.rctNodes[i].Cid()
// index receipt
if !args.rctLeafNodeCIDs[i].Defined() {
return fmt.Errorf("invalid receipt leaf node cid")
}
rctModel := &models.ReceiptModel{ rctModel := &models.ReceiptModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
TxID: txID, TxID: txID,
Contract: contract, Contract: contract,
ContractHash: contractHash, ContractHash: contractHash,
LeafCID: args.rctLeafNodeCIDs[i].String(), CID: rctCid.String(),
LeafMhKey: shared.MultihashKeyFromCID(args.rctLeafNodeCIDs[i]), MhKey: shared.MultihashKeyFromCID(rctCid),
LogRoot: args.rctNodes[i].LogRoot.String(),
} }
if len(receipt.PostState) == 0 { if len(receipt.PostState) == 0 {
rctModel.PostStatus = receipt.Status rctModel.PostStatus = receipt.Status
@ -367,19 +347,15 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(args processArgs) error {
for ti, topic := range l.Topics { for ti, topic := range l.Topics {
topicSet[ti] = topic.Hex() topicSet[ti] = topic.Hex()
} }
logCid := args.logNodes[i][idx].Cid()
if !args.logLeafNodeCIDs[i][idx].Defined() {
return fmt.Errorf("invalid log cid")
}
logDataSet[idx] = &models.LogsModel{ logDataSet[idx] = &models.LogsModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
ReceiptID: txID, ReceiptID: txID,
Address: l.Address.String(), Address: l.Address.String(),
Index: int64(l.Index), Index: int64(l.Index),
Data: l.Data, Data: l.Data,
LeafCID: args.logLeafNodeCIDs[i][idx].String(), CID: logCid.String(),
LeafMhKey: shared.MultihashKeyFromCID(args.logLeafNodeCIDs[i][idx]), MhKey: shared.MultihashKeyFromCID(logCid),
Topic0: topicSet[0], Topic0: topicSet[0],
Topic1: topicSet[1], Topic1: topicSet[1],
Topic2: topicSet[2], Topic2: topicSet[2],
@ -389,54 +365,32 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(args processArgs) error {
sdi.fileWriter.upsertLogCID(logDataSet) sdi.fileWriter.upsertLogCID(logDataSet)
} }
// publish trie nodes, these aren't indexed directly
for i, n := range args.txTrieNodes {
sdi.fileWriter.upsertIPLDNode(sdi.blockNumber, n)
sdi.fileWriter.upsertIPLDNode(sdi.blockNumber, args.rctTrieNodes[i])
}
return nil return nil
} }
// PushStateNode writes a state diff node object (including any child storage nodes) IPLD insert SQL stmt to a file // PushStateNode writes a state diff node object (including any child storage nodes) IPLD insert SQL stmt to a file
func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdtypes.StateNode, headerID string) error { func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdtypes.StateNode, headerID string) error {
// publish the state node // publish the state node
var stateModel models.StateNodeModel var stateModel models.StateLeafModel
if stateNode.NodeType == sdtypes.Removed { if stateNode.NodeType == sdtypes.Removed {
// TODO: still need to handle removed leaves
if atomic.LoadUint32(sdi.removedCacheFlag) == 0 { if atomic.LoadUint32(sdi.removedCacheFlag) == 0 {
atomic.StoreUint32(sdi.removedCacheFlag, 1) atomic.StoreUint32(sdi.removedCacheFlag, 1)
sdi.fileWriter.upsertIPLDDirect(sdi.blockNumber, shared.RemovedNodeMhKey, []byte{}) sdi.fileWriter.upsertIPLDDirect(sdi.blockNumber, shared.RemovedNodeMhKey, []byte{})
} }
stateModel = models.StateNodeModel{ stateModel = models.StateLeafModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: headerID, HeaderID: headerID,
Path: stateNode.Path, Path: stateNode.Path,
StateKey: common.BytesToHash(stateNode.LeafKey).String(), StateKey: common.BytesToHash(stateNode.LeafKey).String(),
CID: shared.RemovedNodeStateCID, CID: shared.RemovedNodeStateCID,
MhKey: shared.RemovedNodeMhKey, MhKey: shared.RemovedNodeMhKey,
NodeType: stateNode.NodeType.Int(),
} }
} else { } else {
stateCIDStr, stateMhKey, err := sdi.fileWriter.upsertIPLDRaw(sdi.blockNumber, ipld2.MEthStateTrie, multihash.KECCAK_256, stateNode.NodeValue) stateCIDStr, stateMhKey, err := sdi.fileWriter.upsertIPLDRaw(sdi.blockNumber, ipld2.MEthStateTrie, multihash.KECCAK_256, stateNode.NodeValue)
if err != nil { if err != nil {
return fmt.Errorf("error generating and cacheing state node IPLD: %v", err) return fmt.Errorf("error generating and cacheing state node IPLD: %v", err)
} }
stateModel = models.StateNodeModel{
BlockNumber: sdi.blockNumber,
HeaderID: headerID,
Path: stateNode.Path,
StateKey: common.BytesToHash(stateNode.LeafKey).String(),
CID: stateCIDStr,
MhKey: stateMhKey,
NodeType: stateNode.NodeType.Int(),
}
}
// index the state node
sdi.fileWriter.upsertStateCID(stateModel)
// if we have a leaf, decode and index the account data
if stateNode.NodeType == sdtypes.Leaf {
var i []interface{} var i []interface{}
if err := rlp.DecodeBytes(stateNode.NodeValue, &i); err != nil { if err := rlp.DecodeBytes(stateNode.NodeValue, &i); err != nil {
return fmt.Errorf("error decoding state leaf node rlp: %s", err.Error()) return fmt.Errorf("error decoding state leaf node rlp: %s", err.Error())
@ -448,26 +402,32 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
if err := rlp.DecodeBytes(i[1].([]byte), &account); err != nil { if err := rlp.DecodeBytes(i[1].([]byte), &account); err != nil {
return fmt.Errorf("error decoding state account rlp: %s", err.Error()) return fmt.Errorf("error decoding state account rlp: %s", err.Error())
} }
accountModel := models.StateAccountModel{ stateModel = models.StateLeafModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: headerID, HeaderID: headerID,
StatePath: stateNode.Path, Path: stateNode.Path,
StateKey: common.BytesToHash(stateNode.LeafKey).String(),
CID: stateCIDStr,
MhKey: stateMhKey,
Balance: account.Balance.String(), Balance: account.Balance.String(),
Nonce: account.Nonce, Nonce: account.Nonce,
CodeHash: account.CodeHash, CodeHash: account.CodeHash,
StorageRoot: account.Root.String(), StorageRoot: account.Root.String(),
} }
sdi.fileWriter.upsertStateAccount(accountModel)
} }
// index the state node
sdi.fileWriter.upsertStateCID(stateModel)
// if there are any storage nodes associated with this node, publish and index them // if there are any storage nodes associated with this node, publish and index them
for _, storageNode := range stateNode.StorageNodes { for _, storageNode := range stateNode.StorageNodes {
if storageNode.NodeType == sdtypes.Removed { if storageNode.NodeType == sdtypes.Removed {
// TODO: still need to handle leaf deletions
if atomic.LoadUint32(sdi.removedCacheFlag) == 0 { if atomic.LoadUint32(sdi.removedCacheFlag) == 0 {
atomic.StoreUint32(sdi.removedCacheFlag, 1) atomic.StoreUint32(sdi.removedCacheFlag, 1)
sdi.fileWriter.upsertIPLDDirect(sdi.blockNumber, shared.RemovedNodeMhKey, []byte{}) sdi.fileWriter.upsertIPLDDirect(sdi.blockNumber, shared.RemovedNodeMhKey, []byte{})
} }
storageModel := models.StorageNodeModel{ storageModel := models.StorageLeafModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: headerID, HeaderID: headerID,
StatePath: stateNode.Path, StatePath: stateNode.Path,
@ -475,7 +435,6 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
StorageKey: common.BytesToHash(storageNode.LeafKey).String(), StorageKey: common.BytesToHash(storageNode.LeafKey).String(),
CID: shared.RemovedNodeStorageCID, CID: shared.RemovedNodeStorageCID,
MhKey: shared.RemovedNodeMhKey, MhKey: shared.RemovedNodeMhKey,
NodeType: storageNode.NodeType.Int(),
} }
sdi.fileWriter.upsertStorageCID(storageModel) sdi.fileWriter.upsertStorageCID(storageModel)
continue continue
@ -484,7 +443,7 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
if err != nil { if err != nil {
return fmt.Errorf("error generating and cacheing storage node IPLD: %v", err) return fmt.Errorf("error generating and cacheing storage node IPLD: %v", err)
} }
storageModel := models.StorageNodeModel{ storageModel := models.StorageLeafModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: headerID, HeaderID: headerID,
StatePath: stateNode.Path, StatePath: stateNode.Path,
@ -492,7 +451,6 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
StorageKey: common.BytesToHash(storageNode.LeafKey).String(), StorageKey: common.BytesToHash(storageNode.LeafKey).String(),
CID: storageCIDStr, CID: storageCIDStr,
MhKey: storageMhKey, MhKey: storageMhKey,
NodeType: storageNode.NodeType.Int(),
} }
sdi.fileWriter.upsertStorageCID(storageModel) sdi.fileWriter.upsertStorageCID(storageModel)
} }

42
statediff/indexer/database/file/writer.go
View File

@ -76,11 +76,11 @@ func (sqw *SQLWriter) Loop() {
l = len(stmt) l = len(stmt)
if sqw.collationIndex+l > writeBufferSize { if sqw.collationIndex+l > writeBufferSize {
if err := sqw.flush(); err != nil { if err := sqw.flush(); err != nil {
panic(fmt.Sprintf("error writing sql stmts buffer to file: %v", err)) panic((any)(fmt.Sprintf("error writing sql stmts buffer to file: %v", err)))
} }
if l > writeBufferSize { if l > writeBufferSize {
if _, err := sqw.wc.Write(stmt); err != nil { if _, err := sqw.wc.Write(stmt); err != nil {
panic(fmt.Sprintf("error writing large sql stmt to file: %v", err)) panic((any)(fmt.Sprintf("error writing large sql stmt to file: %v", err)))
} }
continue continue
} }
@ -89,12 +89,12 @@ func (sqw *SQLWriter) Loop() {
sqw.collationIndex += l sqw.collationIndex += l
case <-sqw.quitChan: case <-sqw.quitChan:
if err := sqw.flush(); err != nil { if err := sqw.flush(); err != nil {
panic(fmt.Sprintf("error writing sql stmts buffer to file: %v", err)) panic((any)(fmt.Sprintf("error writing sql stmts buffer to file: %v", err)))
} }
return return
case <-sqw.flushChan: case <-sqw.flushChan:
if err := sqw.flush(); err != nil { if err := sqw.flush(); err != nil {
panic(fmt.Sprintf("error writing sql stmts buffer to file: %v", err)) panic((any)(fmt.Sprintf("error writing sql stmts buffer to file: %v", err)))
} }
sqw.flushFinished <- struct{}{} sqw.flushFinished <- struct{}{}
} }
@ -142,20 +142,17 @@ const (
alInsert = "INSERT INTO eth.access_list_elements (block_number, tx_id, index, address, storage_keys) VALUES " + alInsert = "INSERT INTO eth.access_list_elements (block_number, tx_id, index, address, storage_keys) VALUES " +
"('%s', '%s', %d, '%s', '%s');\n" "('%s', '%s', %d, '%s', '%s');\n"
rctInsert = "INSERT INTO eth.receipt_cids (block_number, tx_id, leaf_cid, contract, contract_hash, leaf_mh_key, post_state, " + rctInsert = "INSERT INTO eth.receipt_cids (block_number, tx_id, cid, contract, contract_hash, mh_key, post_state, " +
"post_status, log_root) VALUES ('%s', '%s', '%s', '%s', '%s', '%s', '%s', %d, '%s');\n" "post_status) VALUES ('%s', '%s', '%s', '%s', '%s', '%s', '%s', %d);\n"
logInsert = "INSERT INTO eth.log_cids (block_number, leaf_cid, leaf_mh_key, rct_id, address, index, topic0, topic1, topic2, " + logInsert = "INSERT INTO eth.log_cids (block_number, cid, mh_key, rct_id, address, index, topic0, topic1, topic2, " +
"topic3, log_data) VALUES ('%s', '%s', '%s', '%s', '%s', %d, '%s', '%s', '%s', '%s', '\\x%x');\n" "topic3, log_data) VALUES ('%s', '%s', '%s', '%s', '%s', %d, '%s', '%s', '%s', '%s', '\\x%x');\n"
stateInsert = "INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, node_type, diff, mh_key) " + stateInsert = "INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, diff, mh_key, " +
"VALUES ('%s', '%s', '%s', '%s', '\\x%x', %d, %t, '%s');\n" "balance, nonce, code_hash, storage_root) VALUES ('%s', '%s', '%s', '%s', '\\x%x', %t, '%s', '%s', %d, '\\x%x', '%s');\n"
accountInsert = "INSERT INTO eth.state_accounts (block_number, header_id, state_path, balance, nonce, code_hash, storage_root) " +
"VALUES ('%s', '%s', '\\x%x', '%s', %d, '\\x%x', '%s');\n"
storageInsert = "INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, " + storageInsert = "INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, " +
"node_type, diff, mh_key) VALUES ('%s', '%s', '\\x%x', '%s', '%s', '\\x%x', %d, %t, '%s');\n" "diff, mh_key) VALUES ('%s', '%s', '\\x%x', '%s', '%s', '\\x%x', %t, '%s');\n"
) )
func (sqw *SQLWriter) upsertNode(node nodeinfo.Info) { func (sqw *SQLWriter) upsertNode(node nodeinfo.Info) {
@ -222,38 +219,33 @@ func (sqw *SQLWriter) upsertAccessListElement(accessListElement models.AccessLis
} }
func (sqw *SQLWriter) upsertReceiptCID(rct *models.ReceiptModel) { func (sqw *SQLWriter) upsertReceiptCID(rct *models.ReceiptModel) {
sqw.stmts <- []byte(fmt.Sprintf(rctInsert, rct.BlockNumber, rct.TxID, rct.LeafCID, rct.Contract, rct.ContractHash, rct.LeafMhKey, sqw.stmts <- []byte(fmt.Sprintf(rctInsert, rct.BlockNumber, rct.TxID, rct.CID, rct.Contract, rct.ContractHash, rct.MhKey,
rct.PostState, rct.PostStatus, rct.LogRoot)) rct.PostState, rct.PostStatus))
indexerMetrics.receipts.Inc(1) indexerMetrics.receipts.Inc(1)
} }
func (sqw *SQLWriter) upsertLogCID(logs []*models.LogsModel) { func (sqw *SQLWriter) upsertLogCID(logs []*models.LogsModel) {
for _, l := range logs { for _, l := range logs {
sqw.stmts <- []byte(fmt.Sprintf(logInsert, l.BlockNumber, l.LeafCID, l.LeafMhKey, l.ReceiptID, l.Address, l.Index, l.Topic0, sqw.stmts <- []byte(fmt.Sprintf(logInsert, l.BlockNumber, l.CID, l.MhKey, l.ReceiptID, l.Address, l.Index, l.Topic0,
l.Topic1, l.Topic2, l.Topic3, l.Data)) l.Topic1, l.Topic2, l.Topic3, l.Data))
indexerMetrics.logs.Inc(1) indexerMetrics.logs.Inc(1)
} }
} }
func (sqw *SQLWriter) upsertStateCID(stateNode models.StateNodeModel) { func (sqw *SQLWriter) upsertStateCID(stateNode models.StateLeafModel) {
var stateKey string var stateKey string
if stateNode.StateKey != nullHash.String() { if stateNode.StateKey != nullHash.String() {
stateKey = stateNode.StateKey stateKey = stateNode.StateKey
} }
sqw.stmts <- []byte(fmt.Sprintf(stateInsert, stateNode.BlockNumber, stateNode.HeaderID, stateKey, stateNode.CID, stateNode.Path, sqw.stmts <- []byte(fmt.Sprintf(stateInsert, stateNode.BlockNumber, stateNode.HeaderID, stateKey, stateNode.CID, stateNode.Path,
stateNode.NodeType, true, stateNode.MhKey)) true, stateNode.MhKey, stateNode.Balance, stateNode.Nonce, stateNode.CodeHash, stateNode.StorageRoot))
} }
func (sqw *SQLWriter) upsertStateAccount(stateAccount models.StateAccountModel) { func (sqw *SQLWriter) upsertStorageCID(storageCID models.StorageLeafModel) {
sqw.stmts <- []byte(fmt.Sprintf(accountInsert, stateAccount.BlockNumber, stateAccount.HeaderID, stateAccount.StatePath, stateAccount.Balance,
stateAccount.Nonce, stateAccount.CodeHash, stateAccount.StorageRoot))
}
func (sqw *SQLWriter) upsertStorageCID(storageCID models.StorageNodeModel) {
var storageKey string var storageKey string
if storageCID.StorageKey != nullHash.String() { if storageCID.StorageKey != nullHash.String() {
storageKey = storageCID.StorageKey storageKey = storageCID.StorageKey
} }
sqw.stmts <- []byte(fmt.Sprintf(storageInsert, storageCID.BlockNumber, storageCID.HeaderID, storageCID.StatePath, storageKey, storageCID.CID, sqw.stmts <- []byte(fmt.Sprintf(storageInsert, storageCID.BlockNumber, storageCID.HeaderID, storageCID.StatePath, storageKey, storageCID.CID,
storageCID.Path, storageCID.NodeType, true, storageCID.MhKey)) storageCID.Path, true, storageCID.MhKey))
} }

114
statediff/indexer/database/sql/indexer.go
View File

@ -25,7 +25,6 @@ import (
"math/big" "math/big"
"time" "time"
"github.com/ipfs/go-cid"
node "github.com/ipfs/go-ipld-format" node "github.com/ipfs/go-ipld-format"
"github.com/multiformats/go-multihash" "github.com/multiformats/go-multihash"
@ -102,16 +101,13 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
} }
// Generate the block iplds // Generate the block iplds
headerNode, uncleNodes, txNodes, txTrieNodes, rctNodes, rctTrieNodes, logTrieNodes, logLeafNodeCIDs, rctLeafNodeCIDs, err := ipld2.FromBlockAndReceipts(block, receipts) headerNode, uncleNodes, txNodes, rctNodes, logNodes, err := ipld2.FromBlockAndReceipts(block, receipts)
if err != nil { if err != nil {
return nil, fmt.Errorf("error creating IPLD nodes from block and receipts: %v", err) return nil, fmt.Errorf("error creating IPLD nodes from block and receipts: %v", err)
} }
if len(txNodes) != len(rctNodes) || len(rctNodes) != len(rctLeafNodeCIDs) { if len(txNodes) != len(rctNodes) {
return nil, fmt.Errorf("expected number of transactions (%d), receipts (%d), and receipt trie leaf nodes (%d) to be equal", len(txNodes), len(rctNodes), len(rctLeafNodeCIDs)) return nil, fmt.Errorf("expected number of transactions (%d) and receipts (%d) to be equal", len(txNodes), len(rctNodes))
}
if len(txTrieNodes) != len(rctTrieNodes) {
return nil, fmt.Errorf("expected number of tx trie (%d) and rct trie (%d) nodes to be equal", len(txTrieNodes), len(rctTrieNodes))
} }
// Calculate reward // Calculate reward
@ -130,7 +126,7 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
return nil, err return nil, err
} }
defer func() { defer func() {
if p := recover(); p != nil { if p := recover(); p != (any)(nil) {
rollback(sdi.ctx, tx) rollback(sdi.ctx, tx)
panic(p) panic(p)
} else if err != nil { } else if err != nil {
@ -156,7 +152,7 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
close(self.quit) close(self.quit)
close(self.iplds) close(self.iplds)
}() }()
if p := recover(); p != nil { if p := recover(); p != (any)(nil) {
log.Info("panic detected before tx submission, rolling back the tx", "panic", p) log.Info("panic detected before tx submission, rolling back the tx", "panic", p)
rollback(sdi.ctx, tx) rollback(sdi.ctx, tx)
panic(p) panic(p)
@ -218,12 +214,8 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
receipts: receipts, receipts: receipts,
txs: transactions, txs: transactions,
rctNodes: rctNodes, rctNodes: rctNodes,
rctTrieNodes: rctTrieNodes,
txNodes: txNodes, txNodes: txNodes,
txTrieNodes: txTrieNodes, logNodes: logNodes,
logTrieNodes: logTrieNodes,
logLeafNodeCIDs: logLeafNodeCIDs,
rctLeafNodeCIDs: rctLeafNodeCIDs,
}) })
if err != nil { if err != nil {
return nil, err return nil, err
@ -301,12 +293,8 @@ type processArgs struct {
receipts types.Receipts receipts types.Receipts
txs types.Transactions txs types.Transactions
rctNodes []*ipld2.EthReceipt rctNodes []*ipld2.EthReceipt
rctTrieNodes []*ipld2.EthRctTrie
txNodes []*ipld2.EthTx txNodes []*ipld2.EthTx
txTrieNodes []*ipld2.EthTxTrie logNodes [][]*ipld2.EthLog
logTrieNodes [][]node.Node
logLeafNodeCIDs [][]cid.Cid
rctLeafNodeCIDs []cid.Cid
} }
// processReceiptsAndTxs publishes and indexes receipt and transaction IPLDs in Postgres // processReceiptsAndTxs publishes and indexes receipt and transaction IPLDs in Postgres
@ -314,9 +302,6 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs
// Process receipts and txs // Process receipts and txs
signer := types.MakeSigner(sdi.chainConfig, args.blockNumber) signer := types.MakeSigner(sdi.chainConfig, args.blockNumber)
for i, receipt := range args.receipts { for i, receipt := range args.receipts {
for _, logTrieNode := range args.logTrieNodes[i] {
tx.cacheIPLD(logTrieNode)
}
txNode := args.txNodes[i] txNode := args.txNodes[i]
tx.cacheIPLD(txNode) tx.cacheIPLD(txNode)
@ -334,6 +319,7 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs
if err != nil { if err != nil {
return fmt.Errorf("error deriving tx sender: %v", err) return fmt.Errorf("error deriving tx sender: %v", err)
} }
txCid := txNode.Cid()
txModel := models.TxModel{ txModel := models.TxModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: args.headerID, HeaderID: args.headerID,
@ -342,8 +328,8 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs
TxHash: txID, TxHash: txID,
Index: int64(i), Index: int64(i),
Data: trx.Data(), Data: trx.Data(),
CID: txNode.Cid().String(), CID: txCid.String(),
MhKey: shared.MultihashKeyFromCID(txNode.Cid()), MhKey: shared.MultihashKeyFromCID(txCid),
Type: trx.Type(), Type: trx.Type(),
Value: val, Value: val,
} }
@ -375,20 +361,14 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs
if contract != "" { if contract != "" {
contractHash = crypto.Keccak256Hash(common.HexToAddress(contract).Bytes()).String() contractHash = crypto.Keccak256Hash(common.HexToAddress(contract).Bytes()).String()
} }
rctCid := args.rctNodes[i].Cid()
// index receipt
if !args.rctLeafNodeCIDs[i].Defined() {
return fmt.Errorf("invalid receipt leaf node cid")
}
rctModel := &models.ReceiptModel{ rctModel := &models.ReceiptModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
TxID: txID, TxID: txID,
Contract: contract, Contract: contract,
ContractHash: contractHash, ContractHash: contractHash,
LeafCID: args.rctLeafNodeCIDs[i].String(), CID: rctCid.String(),
LeafMhKey: shared.MultihashKeyFromCID(args.rctLeafNodeCIDs[i]), MhKey: shared.MultihashKeyFromCID(rctCid),
LogRoot: args.rctNodes[i].LogRoot.String(),
} }
if len(receipt.PostState) == 0 { if len(receipt.PostState) == 0 {
rctModel.PostStatus = receipt.Status rctModel.PostStatus = receipt.Status
@ -407,19 +387,15 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs
for ti, topic := range l.Topics { for ti, topic := range l.Topics {
topicSet[ti] = topic.Hex() topicSet[ti] = topic.Hex()
} }
logCid := args.logNodes[i][idx].Cid()
if !args.logLeafNodeCIDs[i][idx].Defined() {
return fmt.Errorf("invalid log cid")
}
logDataSet[idx] = &models.LogsModel{ logDataSet[idx] = &models.LogsModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
ReceiptID: txID, ReceiptID: txID,
Address: l.Address.String(), Address: l.Address.String(),
Index: int64(l.Index), Index: int64(l.Index),
Data: l.Data, Data: l.Data,
LeafCID: args.logLeafNodeCIDs[i][idx].String(), CID: logCid.String(),
LeafMhKey: shared.MultihashKeyFromCID(args.logLeafNodeCIDs[i][idx]), MhKey: shared.MultihashKeyFromCID(logCid),
Topic0: topicSet[0], Topic0: topicSet[0],
Topic1: topicSet[1], Topic1: topicSet[1],
Topic2: topicSet[2], Topic2: topicSet[2],
@ -432,12 +408,6 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs
} }
} }
// publish trie nodes, these aren't indexed directly
for i, n := range args.txTrieNodes {
tx.cacheIPLD(n)
tx.cacheIPLD(args.rctTrieNodes[i])
}
return nil return nil
} }
@ -448,41 +418,23 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
return fmt.Errorf("sql batch is expected to be of type %T, got %T", &BatchTx{}, batch) return fmt.Errorf("sql batch is expected to be of type %T, got %T", &BatchTx{}, batch)
} }
// publish the state node // publish the state node
var stateModel models.StateNodeModel var stateModel models.StateLeafModel
if stateNode.NodeType == sdtypes.Removed { if stateNode.NodeType == sdtypes.Removed {
// TODO: we need to continue tracking removed nodes for leaves?
tx.cacheRemoved(shared.RemovedNodeMhKey, []byte{}) tx.cacheRemoved(shared.RemovedNodeMhKey, []byte{})
stateModel = models.StateNodeModel{ stateModel = models.StateLeafModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: headerID, HeaderID: headerID,
Path: stateNode.Path, Path: stateNode.Path,
StateKey: common.BytesToHash(stateNode.LeafKey).String(), StateKey: common.BytesToHash(stateNode.LeafKey).String(),
CID: shared.RemovedNodeStateCID, CID: shared.RemovedNodeStateCID,
MhKey: shared.RemovedNodeMhKey, MhKey: shared.RemovedNodeMhKey,
NodeType: stateNode.NodeType.Int(),
} }
} else { } else {
stateCIDStr, stateMhKey, err := tx.cacheRaw(ipld2.MEthStateTrie, multihash.KECCAK_256, stateNode.NodeValue) stateCIDStr, stateMhKey, err := tx.cacheRaw(ipld2.MEthStateTrie, multihash.KECCAK_256, stateNode.NodeValue)
if err != nil { if err != nil {
return fmt.Errorf("error generating and cacheing state node IPLD: %v", err) return fmt.Errorf("error generating and cacheing state node IPLD: %v", err)
} }
stateModel = models.StateNodeModel{
BlockNumber: sdi.blockNumber,
HeaderID: headerID,
Path: stateNode.Path,
StateKey: common.BytesToHash(stateNode.LeafKey).String(),
CID: stateCIDStr,
MhKey: stateMhKey,
NodeType: stateNode.NodeType.Int(),
}
}
// index the state node
if err := sdi.dbWriter.upsertStateCID(tx.dbtx, stateModel); err != nil {
return err
}
// if we have a leaf, decode and index the account data
if stateNode.NodeType == sdtypes.Leaf {
var i []interface{} var i []interface{}
if err := rlp.DecodeBytes(stateNode.NodeValue, &i); err != nil { if err := rlp.DecodeBytes(stateNode.NodeValue, &i); err != nil {
return fmt.Errorf("error decoding state leaf node rlp: %s", err.Error()) return fmt.Errorf("error decoding state leaf node rlp: %s", err.Error())
@ -494,25 +446,31 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
if err := rlp.DecodeBytes(i[1].([]byte), &account); err != nil { if err := rlp.DecodeBytes(i[1].([]byte), &account); err != nil {
return fmt.Errorf("error decoding state account rlp: %s", err.Error()) return fmt.Errorf("error decoding state account rlp: %s", err.Error())
} }
accountModel := models.StateAccountModel{ stateModel = models.StateLeafModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: headerID, HeaderID: headerID,
StatePath: stateNode.Path, Path: stateNode.Path,
StateKey: common.BytesToHash(stateNode.LeafKey).String(),
CID: stateCIDStr,
MhKey: stateMhKey,
Balance: account.Balance.String(), Balance: account.Balance.String(),
Nonce: account.Nonce, Nonce: account.Nonce,
CodeHash: account.CodeHash, CodeHash: account.CodeHash,
StorageRoot: account.Root.String(), StorageRoot: account.Root.String(),
} }
if err := sdi.dbWriter.upsertStateAccount(tx.dbtx, accountModel); err != nil { }
return err
} // index the state node
if err := sdi.dbWriter.upsertStateCID(tx.dbtx, stateModel); err != nil {
return err
} }
// if there are any storage nodes associated with this node, publish and index them // if there are any storage nodes associated with this node, publish and index them
for _, storageNode := range stateNode.StorageNodes { for _, storageNode := range stateNode.StorageNodes {
// TODO: we still need to handle leaf deletions
if storageNode.NodeType == sdtypes.Removed { if storageNode.NodeType == sdtypes.Removed {
tx.cacheRemoved(shared.RemovedNodeMhKey, []byte{}) tx.cacheRemoved(shared.RemovedNodeMhKey, []byte{})
storageModel := models.StorageNodeModel{ storageModel := models.StorageLeafModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: headerID, HeaderID: headerID,
StatePath: stateNode.Path, StatePath: stateNode.Path,
@ -520,7 +478,6 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
StorageKey: common.BytesToHash(storageNode.LeafKey).String(), StorageKey: common.BytesToHash(storageNode.LeafKey).String(),
CID: shared.RemovedNodeStorageCID, CID: shared.RemovedNodeStorageCID,
MhKey: shared.RemovedNodeMhKey, MhKey: shared.RemovedNodeMhKey,
NodeType: storageNode.NodeType.Int(),
} }
if err := sdi.dbWriter.upsertStorageCID(tx.dbtx, storageModel); err != nil { if err := sdi.dbWriter.upsertStorageCID(tx.dbtx, storageModel); err != nil {
return err return err
@ -531,7 +488,7 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
if err != nil { if err != nil {
return fmt.Errorf("error generating and cacheing storage node IPLD: %v", err) return fmt.Errorf("error generating and cacheing storage node IPLD: %v", err)
} }
storageModel := models.StorageNodeModel{ storageModel := models.StorageLeafModel{
BlockNumber: sdi.blockNumber, BlockNumber: sdi.blockNumber,
HeaderID: headerID, HeaderID: headerID,
StatePath: stateNode.Path, StatePath: stateNode.Path,
@ -539,7 +496,6 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
StorageKey: common.BytesToHash(storageNode.LeafKey).String(), StorageKey: common.BytesToHash(storageNode.LeafKey).String(),
CID: storageCIDStr, CID: storageCIDStr,
MhKey: storageMhKey, MhKey: storageMhKey,
NodeType: storageNode.NodeType.Int(),
} }
if err := sdi.dbWriter.upsertStorageCID(tx.dbtx, storageModel); err != nil { if err := sdi.dbWriter.upsertStorageCID(tx.dbtx, storageModel); err != nil {
return err return err
@ -595,7 +551,7 @@ func (sdi *StateDiffIndexer) InsertWatchedAddresses(args []sdtypes.WatchAddressA
return err return err
} }
defer func() { defer func() {
if p := recover(); p != nil { if p := recover(); p != (any)(nil) {
rollback(sdi.ctx, tx) rollback(sdi.ctx, tx)
panic(p) panic(p)
} else if err != nil { } else if err != nil {
@ -623,7 +579,7 @@ func (sdi *StateDiffIndexer) RemoveWatchedAddresses(args []sdtypes.WatchAddressA
return err return err
} }
defer func() { defer func() {
if p := recover(); p != nil { if p := recover(); p != (any)(nil) {
rollback(sdi.ctx, tx) rollback(sdi.ctx, tx)
panic(p) panic(p)
} else if err != nil { } else if err != nil {
@ -650,7 +606,7 @@ func (sdi *StateDiffIndexer) SetWatchedAddresses(args []sdtypes.WatchAddressArg,
return err return err
} }
defer func() { defer func() {
if p := recover(); p != nil { if p := recover(); p != (any)(nil) {
rollback(sdi.ctx, tx) rollback(sdi.ctx, tx)
panic(p) panic(p)
} else if err != nil { } else if err != nil {

18
statediff/indexer/database/sql/postgres/database.go
View File

@ -63,32 +63,26 @@ func (db *DB) InsertAccessListElementStm() string {
// InsertRctStm satisfies the sql.Statements interface // InsertRctStm satisfies the sql.Statements interface
func (db *DB) InsertRctStm() string { func (db *DB) InsertRctStm() string {
return `INSERT INTO eth.receipt_cids (block_number, tx_id, leaf_cid, contract, contract_hash, leaf_mh_key, post_state, post_status, log_root) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) return `INSERT INTO eth.receipt_cids (block_number, tx_id, cid, contract, contract_hash, mh_key, post_state, post_status) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (tx_id, block_number) DO NOTHING` ON CONFLICT (tx_id, block_number) DO NOTHING`
} }
// InsertLogStm satisfies the sql.Statements interface // InsertLogStm satisfies the sql.Statements interface
func (db *DB) InsertLogStm() string { func (db *DB) InsertLogStm() string {
return `INSERT INTO eth.log_cids (block_number, leaf_cid, leaf_mh_key, rct_id, address, index, topic0, topic1, topic2, topic3, log_data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) return `INSERT INTO eth.log_cids (block_number, cid, mh_key, rct_id, address, index, topic0, topic1, topic2, topic3, log_data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT (rct_id, index, block_number) DO NOTHING` ON CONFLICT (rct_id, index, block_number) DO NOTHING`
} }
// InsertStateStm satisfies the sql.Statements interface // InsertStateStm satisfies the sql.Statements interface
func (db *DB) InsertStateStm() string { func (db *DB) InsertStateStm() string {
return `INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) return `INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, diff, mh_key, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT (header_id, state_path, block_number) DO UPDATE SET (block_number, state_leaf_key, cid, node_type, diff, mh_key) = ($1, $3, $4, $6, $7, $8)` ON CONFLICT (header_id, state_path, block_number) DO UPDATE SET (block_number, state_leaf_key, cid, diff, mh_key) = ($1, $3, $4, $6, $7, $8, $9, $10, $11)`
}
// InsertAccountStm satisfies the sql.Statements interface
func (db *DB) InsertAccountStm() string {
return `INSERT INTO eth.state_accounts (block_number, header_id, state_path, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (header_id, state_path, block_number) DO NOTHING`
} }
// InsertStorageStm satisfies the sql.Statements interface // InsertStorageStm satisfies the sql.Statements interface
func (db *DB) InsertStorageStm() string { func (db *DB) InsertStorageStm() string {
return `INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) return `INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (header_id, state_path, storage_path, block_number) DO UPDATE SET (block_number, storage_leaf_key, cid, node_type, diff, mh_key) = ($1, $4, $5, $7, $8, $9)` ON CONFLICT (header_id, state_path, storage_path, block_number) DO UPDATE SET (block_number, storage_leaf_key, cid, diff, mh_key) = ($1, $4, $5, $7, $8)`
} }
// InsertIPLDStm satisfies the sql.Statements interface // InsertIPLDStm satisfies the sql.Statements interface

42
statediff/indexer/database/sql/writer.go
View File

@ -105,13 +105,13 @@ func (w *Writer) upsertAccessListElement(tx Tx, accessListElement models.AccessL
} }
/* /*
INSERT INTO eth.receipt_cids (block_number, tx_id, leaf_cid, contract, contract_hash, leaf_mh_key, post_state, post_status, log_root) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) INSERT INTO eth.receipt_cids (block_number, tx_id, cid, contract, contract_hash, mh_key, post_state, post_status) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (tx_id, block_number) DO NOTHING ON CONFLICT (tx_id, block_number) DO NOTHING
*/ */
func (w *Writer) upsertReceiptCID(tx Tx, rct *models.ReceiptModel) error { func (w *Writer) upsertReceiptCID(tx Tx, rct *models.ReceiptModel) error {
_, err := tx.Exec(w.db.Context(), w.db.InsertRctStm(), _, err := tx.Exec(w.db.Context(), w.db.InsertRctStm(),
rct.BlockNumber, rct.TxID, rct.LeafCID, rct.Contract, rct.ContractHash, rct.LeafMhKey, rct.PostState, rct.BlockNumber, rct.TxID, rct.CID, rct.Contract, rct.ContractHash, rct.MhKey, rct.PostState,
rct.PostStatus, rct.LogRoot) rct.PostStatus)
if err != nil { if err != nil {
return fmt.Errorf("error upserting receipt_cids entry: %w", err) return fmt.Errorf("error upserting receipt_cids entry: %w", err)
} }
@ -120,13 +120,13 @@ func (w *Writer) upsertReceiptCID(tx Tx, rct *models.ReceiptModel) error {
} }
/* /*
INSERT INTO eth.log_cids (block_number, leaf_cid, leaf_mh_key, rct_id, address, index, topic0, topic1, topic2, topic3, log_data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) INSERT INTO eth.log_cids (block_number, cid, mh_key, rct_id, address, index, topic0, topic1, topic2, topic3, log_data) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT (rct_id, index, block_number) DO NOTHING ON CONFLICT (rct_id, index, block_number) DO NOTHING
*/ */
func (w *Writer) upsertLogCID(tx Tx, logs []*models.LogsModel) error { func (w *Writer) upsertLogCID(tx Tx, logs []*models.LogsModel) error {
for _, log := range logs { for _, log := range logs {
_, err := tx.Exec(w.db.Context(), w.db.InsertLogStm(), _, err := tx.Exec(w.db.Context(), w.db.InsertLogStm(),
log.BlockNumber, log.LeafCID, log.LeafMhKey, log.ReceiptID, log.Address, log.Index, log.Topic0, log.Topic1, log.BlockNumber, log.CID, log.MhKey, log.ReceiptID, log.Address, log.Index, log.Topic0, log.Topic1,
log.Topic2, log.Topic3, log.Data) log.Topic2, log.Topic3, log.Data)
if err != nil { if err != nil {
return fmt.Errorf("error upserting logs entry: %w", err) return fmt.Errorf("error upserting logs entry: %w", err)
@ -137,17 +137,17 @@ func (w *Writer) upsertLogCID(tx Tx, logs []*models.LogsModel) error {
} }
/* /*
INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) INSERT INTO eth.state_cids (block_number, header_id, state_leaf_key, cid, state_path, diff, mh_key, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT (header_id, state_path, block_number) DO UPDATE SET (block_number, state_leaf_key, cid, node_type, diff, mh_key) = ($1 $3, $4, $6, $7, $8) ON CONFLICT (header_id, state_path, block_number) DO UPDATE SET (block_number, state_leaf_key, cid, diff, mh_key, balance, nonce, code_hash, storage_root) = ($1 $3, $4, $6, $7, $8, $9, $10, $11)
*/ */
func (w *Writer) upsertStateCID(tx Tx, stateNode models.StateNodeModel) error { func (w *Writer) upsertStateCID(tx Tx, stateNode models.StateLeafModel) error {
var stateKey string var stateKey string
if stateNode.StateKey != nullHash.String() { if stateNode.StateKey != nullHash.String() {
stateKey = stateNode.StateKey stateKey = stateNode.StateKey
} }
_, err := tx.Exec(w.db.Context(), w.db.InsertStateStm(), _, err := tx.Exec(w.db.Context(), w.db.InsertStateStm(),
stateNode.BlockNumber, stateNode.HeaderID, stateKey, stateNode.CID, stateNode.Path, stateNode.NodeType, true, stateNode.BlockNumber, stateNode.HeaderID, stateKey, stateNode.CID, stateNode.Path, true,
stateNode.MhKey) stateNode.MhKey, stateNode.Balance, stateNode.Nonce, stateNode.CodeHash, stateNode.StorageRoot)
if err != nil { if err != nil {
return fmt.Errorf("error upserting state_cids entry: %v", err) return fmt.Errorf("error upserting state_cids entry: %v", err)
} }
@ -155,31 +155,17 @@ func (w *Writer) upsertStateCID(tx Tx, stateNode models.StateNodeModel) error {
} }
/* /*
INSERT INTO eth.state_accounts (block_number, header_id, state_path, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5, $6, $7) INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (header_id, state_path, block_number) DO NOTHING ON CONFLICT (header_id, state_path, storage_path, block_number) DO UPDATE SET (block_number, storage_leaf_key, cid, diff, mh_key) = ($1, $4, $5, $7, $8)
*/ */
func (w *Writer) upsertStateAccount(tx Tx, stateAccount models.StateAccountModel) error { func (w *Writer) upsertStorageCID(tx Tx, storageCID models.StorageLeafModel) error {
_, err := tx.Exec(w.db.Context(), w.db.InsertAccountStm(),
stateAccount.BlockNumber, stateAccount.HeaderID, stateAccount.StatePath, stateAccount.Balance,
stateAccount.Nonce, stateAccount.CodeHash, stateAccount.StorageRoot)
if err != nil {
return fmt.Errorf("error upserting state_accounts entry: %v", err)
}
return nil
}
/*
INSERT INTO eth.storage_cids (block_number, header_id, state_path, storage_leaf_key, cid, storage_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (header_id, state_path, storage_path, block_number) DO UPDATE SET (block_number, storage_leaf_key, cid, node_type, diff, mh_key) = ($1, $4, $5, $7, $8, $9)
*/
func (w *Writer) upsertStorageCID(tx Tx, storageCID models.StorageNodeModel) error {
var storageKey string var storageKey string
if storageCID.StorageKey != nullHash.String() { if storageCID.StorageKey != nullHash.String() {
storageKey = storageCID.StorageKey storageKey = storageCID.StorageKey
} }
_, err := tx.Exec(w.db.Context(), w.db.InsertStorageStm(), _, err := tx.Exec(w.db.Context(), w.db.InsertStorageStm(),
storageCID.BlockNumber, storageCID.HeaderID, storageCID.StatePath, storageKey, storageCID.CID, storageCID.Path, storageCID.BlockNumber, storageCID.HeaderID, storageCID.StatePath, storageKey, storageCID.CID, storageCID.Path,
storageCID.NodeType, true, storageCID.MhKey) true, storageCID.MhKey)
if err != nil { if err != nil {
return fmt.Errorf("error upserting storage_cids entry: %v", err) return fmt.Errorf("error upserting storage_cids entry: %v", err)
} }

4
statediff/indexer/ipld/eth_header_test.go
View File

@ -430,10 +430,10 @@ func TestEthBlockCopy(t *testing.T) {
defer func() { defer func() {
r := recover() r := recover()
if r == nil { if r == (any)(nil) {
t.Fatal("Expected panic") t.Fatal("Expected panic")
} }
if r != "implement me" { if r != (any)("implement me") {
t.Fatalf("Wrong panic message\r\nexpected %s\r\ngot %s", "'implement me'", r) t.Fatalf("Wrong panic message\r\nexpected %s\r\ngot %s", "'implement me'", r)
} }
}() }()

68
statediff/indexer/ipld/eth_parser.go
View File

@ -124,12 +124,12 @@ func FromBlockJSON(r io.Reader) (*EthHeader, []*EthTx, []*EthTxTrie, error) {
} }
// FromBlockAndReceipts takes a block and processes it // FromBlockAndReceipts takes a block and processes it
// to return it a set of IPLD nodes for further processing. // to return a set of IPLD nodes for further processing.
func FromBlockAndReceipts(block *types.Block, receipts []*types.Receipt) (*EthHeader, []*EthHeader, []*EthTx, []*EthTxTrie, []*EthReceipt, []*EthRctTrie, [][]node.Node, [][]cid.Cid, []cid.Cid, error) { func FromBlockAndReceipts(block *types.Block, receipts []*types.Receipt) (*EthHeader, []*EthHeader, []*EthTx, []*EthReceipt, [][]*EthLog, error) {
// Process the header // Process the header
headerNode, err := NewEthHeader(block.Header()) headerNode, err := NewEthHeader(block.Header())
if err != nil { if err != nil {
return nil, nil, nil, nil, nil, nil, nil, nil, nil, err return nil, nil, nil, nil, nil, err
} }
// Process the uncles // Process the uncles
@ -137,23 +137,71 @@ func FromBlockAndReceipts(block *types.Block, receipts []*types.Receipt) (*EthHe
for i, uncle := range block.Uncles() { for i, uncle := range block.Uncles() {
uncleNode, err := NewEthHeader(uncle) uncleNode, err := NewEthHeader(uncle)
if err != nil { if err != nil {
return nil, nil, nil, nil, nil, nil, nil, nil, nil, err return nil, nil, nil, nil, nil, err
} }
uncleNodes[i] = uncleNode uncleNodes[i] = uncleNode
} }
// Process the txs // Process the txs
txNodes, txTrieNodes, err := processTransactions(block.Transactions(), txNodes, err := processTransactionsSuccinct(block.Transactions())
block.Header().TxHash[:])
if err != nil { if err != nil {
return nil, nil, nil, nil, nil, nil, nil, nil, nil, err return nil, nil, nil, nil, nil, err
} }
// Process the receipts and logs // Process the receipts and logs
rctNodes, tctTrieNodes, logTrieAndLogNodes, logLeafNodeCIDs, rctLeafNodeCIDs, err := processReceiptsAndLogs(receipts, rctNodes, logNodes, err := processReceiptsAndLogsSuccinct(receipts)
block.Header().ReceiptHash[:])
return headerNode, uncleNodes, txNodes, txTrieNodes, rctNodes, tctTrieNodes, logTrieAndLogNodes, logLeafNodeCIDs, rctLeafNodeCIDs, err return headerNode, uncleNodes, txNodes, rctNodes, logNodes, err
}
// processTransactions converts txs to IPLDs
func processTransactionsSuccinct(txs []*types.Transaction) ([]*EthTx, error) {
ethTxNodes := make([]*EthTx, len(txs))
for i, tx := range txs {
ethTx, err := NewEthTx(tx)
if err != nil {
return nil, err
}
ethTxNodes[i] = ethTx
}
return ethTxNodes, nil
}
// processReceiptsAndLogsSuccinct will take in receipts and returns the IPLDs for the receipts and logs contained within
func processReceiptsAndLogsSuccinct(rcts []*types.Receipt) ([]*EthReceipt, [][]*EthLog, error) {
// Pre allocating memory.
ethRctNodes := make([]*EthReceipt, len(rcts))
ethLogNodes := make([][]*EthLog, 0, len(rcts))
for i, rct := range rcts {
// Process logs for each receipt.
ethLogs, err := processLogsSuccinct(rct.Logs)
if err != nil {
return nil, nil, err
}
ethRct, err := NewReceipt(rct)
if err != nil {
return nil, nil, err
}
ethRctNodes[i] = ethRct
ethLogNodes[i] = ethLogs
}
return ethRctNodes, ethLogNodes, nil
}
func processLogsSuccinct(logs []*types.Log) ([]*EthLog, error) {
ethLogs := make([]*EthLog, len(logs))
for i, log := range logs {
logNode, err := NewLog(log)
if err != nil {
return nil, err
}
ethLogs[i] = logNode
}
return ethLogs, nil
} }
// processTransactions will take the found transactions in a parsed block body // processTransactions will take the found transactions in a parsed block body

18
statediff/indexer/models/batch.go
View File

@ -69,7 +69,6 @@ type ReceiptBatch struct {
PostStates []string PostStates []string
Contracts []string Contracts []string
ContractHashes []string ContractHashes []string
LogRoots []string
} }
// LogBatch holds the arguments for a batch insert of log data // LogBatch holds the arguments for a batch insert of log data
@ -87,37 +86,28 @@ type LogBatch struct {
Topic3s []string Topic3s []string
} }
// StateBatch holds the arguments for a batch insert of state data // StateLeafBatch holds the arguments for a batch insert of state data
type StateBatch struct { type StateLeafBatch struct {
BlockNumbers []string BlockNumbers []string
HeaderID string HeaderID string
Paths [][]byte Paths [][]byte
StateKeys []string StateKeys []string
NodeTypes []int
CIDs []string CIDs []string
MhKeys []string MhKeys []string
Diff bool Diff bool
}
// AccountBatch holds the arguments for a batch insert of account data
type AccountBatch struct {
BlockNumbers []string
HeaderID string
StatePaths [][]byte
Balances []string Balances []string
Nonces []uint64 Nonces []uint64
CodeHashes [][]byte CodeHashes [][]byte
StorageRoots []string StorageRoots []string
} }
// StorageBatch holds the arguments for a batch insert of storage data // StorageLeafBatch holds the arguments for a batch insert of storage data
type StorageBatch struct { type StorageLeafBatch struct {
BlockNumbers []string BlockNumbers []string
HeaderID string HeaderID string
StatePaths [][]string StatePaths [][]string
Paths [][]byte Paths [][]byte
StorageKeys []string StorageKeys []string
NodeTypes []int
CIDs []string CIDs []string
MhKeys []string MhKeys []string
Diff bool Diff bool

73
statediff/indexer/models/models.go
View File

@ -84,71 +84,60 @@ type AccessListElementModel struct {
type ReceiptModel struct { type ReceiptModel struct {
BlockNumber string `db:"block_number"` BlockNumber string `db:"block_number"`
TxID string `db:"tx_id"` TxID string `db:"tx_id"`
LeafCID string `db:"leaf_cid"` CID string `db:"cid"`
LeafMhKey string `db:"leaf_mh_key"` MhKey string `db:"mh_key"`
PostStatus uint64 `db:"post_status"` PostStatus uint64 `db:"post_status"`
PostState string `db:"post_state"` PostState string `db:"post_state"`
Contract string `db:"contract"` Contract string `db:"contract"`
ContractHash string `db:"contract_hash"` ContractHash string `db:"contract_hash"`
LogRoot string `db:"log_root"`
} }
// StateNodeModel is the db model for eth.state_cids // StateLeafModel is the db model for eth.state_cids
type StateNodeModel struct { type StateLeafModel struct {
BlockNumber string `db:"block_number"` BlockNumber string `db:"block_number"`
HeaderID string `db:"header_id"` HeaderID string `db:"header_id"`
Path []byte `db:"state_path"` Path []byte `db:"state_path"`
StateKey string `db:"state_leaf_key"` StateKey string `db:"state_leaf_key"`
NodeType int `db:"node_type"`
CID string `db:"cid"` CID string `db:"cid"`
MhKey string `db:"mh_key"` MhKey string `db:"mh_key"`
Diff bool `db:"diff"` Diff bool `db:"diff"`
}
// StorageNodeModel is the db model for eth.storage_cids
type StorageNodeModel struct {
BlockNumber string `db:"block_number"`
HeaderID string `db:"header_id"`
StatePath []byte `db:"state_path"`
Path []byte `db:"storage_path"`
StorageKey string `db:"storage_leaf_key"`
NodeType int `db:"node_type"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Diff bool `db:"diff"`
}
// StorageNodeWithStateKeyModel is a db model for eth.storage_cids + eth.state_cids.state_key
type StorageNodeWithStateKeyModel struct {
BlockNumber string `db:"block_number"`
HeaderID string `db:"header_id"`
StatePath []byte `db:"state_path"`
Path []byte `db:"storage_path"`
StateKey string `db:"state_leaf_key"`
StorageKey string `db:"storage_leaf_key"`
NodeType int `db:"node_type"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Diff bool `db:"diff"`
}
// StateAccountModel is a db model for an eth state account (decoded value of state leaf node)
type StateAccountModel struct {
BlockNumber string `db:"block_number"`
HeaderID string `db:"header_id"`
StatePath []byte `db:"state_path"`
Balance string `db:"balance"` Balance string `db:"balance"`
Nonce uint64 `db:"nonce"` Nonce uint64 `db:"nonce"`
CodeHash []byte `db:"code_hash"` CodeHash []byte `db:"code_hash"`
StorageRoot string `db:"storage_root"` StorageRoot string `db:"storage_root"`
} }
// StorageLeafModel is the db model for eth.storage_cids
type StorageLeafModel struct {
BlockNumber string `db:"block_number"`
HeaderID string `db:"header_id"`
StatePath []byte `db:"state_path"`
Path []byte `db:"storage_path"`
StorageKey string `db:"storage_leaf_key"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Diff bool `db:"diff"`
}
// StorageLeafWithStateKeyModel is a db model for eth.storage_cids + eth.state_cids.state_key
type StorageLeafWithStateKeyModel struct {
BlockNumber string `db:"block_number"`
HeaderID string `db:"header_id"`
StatePath []byte `db:"state_path"`
Path []byte `db:"storage_path"`
StateKey string `db:"state_leaf_key"`
StorageKey string `db:"storage_leaf_key"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Diff bool `db:"diff"`
}
// LogsModel is the db model for eth.logs // LogsModel is the db model for eth.logs
type LogsModel struct { type LogsModel struct {
BlockNumber string `db:"block_number"` BlockNumber string `db:"block_number"`
ReceiptID string `db:"rct_id"` ReceiptID string `db:"rct_id"`
LeafCID string `db:"leaf_cid"` CID string `db:"cid"`
LeafMhKey string `db:"leaf_mh_key"` MhKey string `db:"mh_key"`
Address string `db:"address"` Address string `db:"address"`
Index int64 `db:"index"` Index int64 `db:"index"`
Data []byte `db:"log_data"` Data []byte `db:"log_data"`