cerc-io/plugeth-statediff
11
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Port: Check if the statediff is already in progress or completed.

Browse Source
This commit is contained in:
roysc
2023-07-14 12:57:34 +08:00
parent 49e0beb389
commit c96215f9be
10 changed files with 89 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
indexer/database/dump/indexer.go
+6
View File
@@ -413,6 +413,12 @@ func (sdi *StateDiffIndexer) PushIPLD(batch interfaces.Batch, ipld sdtypes.IPLD)
return nil
}
// HasBlock checks whether the indicated block already exists in the output.
// In the "dump" case, this is presumed to be false.
func (sdi *StateDiffIndexer) HasBlock(hash common.Hash, number uint64) (bool, error) {
return false, nil
}
// Close satisfies io.Closer
func (sdi *StateDiffIndexer) Close() error {
return sdi.dump.Close()
indexer/database/file/indexer.go
+6
View File
@@ -459,6 +459,12 @@ func (sdi *StateDiffIndexer) PushIPLD(batch interfaces.Batch, ipld sdtypes.IPLD)
return nil
}
// HasBlock checks whether the indicated block already exists in the output.
// In the "file" case this is presumed to be false.
func (sdi *StateDiffIndexer) HasBlock(hash common.Hash, number uint64) (bool, error) {
return false, nil
}
// Close satisfies io.Closer
func (sdi *StateDiffIndexer) Close() error {
return sdi.fileWriter.Close()
indexer/database/sql/indexer.go
+18 -13
View File
@@ -25,7 +25,7 @@ import (
"math/big"
"time"
core "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/metrics"
@@ -251,7 +251,7 @@ func (sdi *StateDiffIndexer) processHeader(tx *BatchTx, header *types.Header, he
}
// processUncles publishes and indexes uncle IPLDs in Postgres
func (sdi *StateDiffIndexer) processUncles(tx *BatchTx, headerID string, blockNumber *big.Int, unclesHash core.Hash, uncles []*types.Header) error {
func (sdi *StateDiffIndexer) processUncles(tx *BatchTx, headerID string, blockNumber *big.Int, unclesHash common.Hash, uncles []*types.Header) error {
// publish and index uncles
uncleEncoding, err := rlp.EncodeToBytes(uncles)
if err != nil {
@@ -352,7 +352,7 @@ func (sdi *StateDiffIndexer) processReceiptsAndTxs(tx *BatchTx, args processArgs
if len(receipt.PostState) == 0 {
rctModel.PostStatus = receipt.Status
} else {
rctModel.PostState = core.BytesToHash(receipt.PostState).String()
rctModel.PostState = common.BytesToHash(receipt.PostState).String()
}
if err := sdi.dbWriter.upsertReceiptCID(tx.dbtx, rctModel); err != nil {
@@ -403,7 +403,7 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
stateModel = models.StateNodeModel{
BlockNumber: tx.BlockNumber,
HeaderID: headerID,
StateKey: core.BytesToHash(stateNode.AccountWrapper.LeafKey).String(),
StateKey: common.BytesToHash(stateNode.AccountWrapper.LeafKey).String(),
CID: shared.RemovedNodeStateCID,
Removed: true,
}
@@ -411,12 +411,12 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
stateModel = models.StateNodeModel{
BlockNumber: tx.BlockNumber,
HeaderID: headerID,
StateKey: core.BytesToHash(stateNode.AccountWrapper.LeafKey).String(),
StateKey: common.BytesToHash(stateNode.AccountWrapper.LeafKey).String(),
CID: stateNode.AccountWrapper.CID,
Removed: false,
Balance: stateNode.AccountWrapper.Account.Balance.String(),
Nonce: stateNode.AccountWrapper.Account.Nonce,
CodeHash: core.BytesToHash(stateNode.AccountWrapper.Account.CodeHash).String(),
CodeHash: common.BytesToHash(stateNode.AccountWrapper.Account.CodeHash).String(),
StorageRoot: stateNode.AccountWrapper.Account.Root.String(),
}
}
@@ -433,8 +433,8 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
storageModel := models.StorageNodeModel{
BlockNumber: tx.BlockNumber,
HeaderID: headerID,
StateKey: core.BytesToHash(stateNode.AccountWrapper.LeafKey).String(),
StorageKey: core.BytesToHash(storageNode.LeafKey).String(),
StateKey: common.BytesToHash(stateNode.AccountWrapper.LeafKey).String(),
StorageKey: common.BytesToHash(storageNode.LeafKey).String(),
CID: shared.RemovedNodeStorageCID,
Removed: true,
Value: []byte{},
@@ -447,8 +447,8 @@ func (sdi *StateDiffIndexer) PushStateNode(batch interfaces.Batch, stateNode sdt
storageModel := models.StorageNodeModel{
BlockNumber: tx.BlockNumber,
HeaderID: headerID,
StateKey: core.BytesToHash(stateNode.AccountWrapper.LeafKey).String(),
StorageKey: core.BytesToHash(storageNode.LeafKey).String(),
StateKey: common.BytesToHash(stateNode.AccountWrapper.LeafKey).String(),
StorageKey: common.BytesToHash(storageNode.LeafKey).String(),
CID: storageNode.CID,
Removed: false,
Value: storageNode.Value,
@@ -471,6 +471,11 @@ func (sdi *StateDiffIndexer) PushIPLD(batch interfaces.Batch, ipld sdtypes.IPLD)
return nil
}
// HasBlock checks whether the indicated block already exists in the database.
func (sdi *StateDiffIndexer) HasBlock(hash common.Hash, number uint64) (bool, error) {
return sdi.dbWriter.hasHeader(hash, number)
}
// Close satisfies io.Closer
func (sdi *StateDiffIndexer) Close() error {
return sdi.dbWriter.Close()
@@ -479,7 +484,7 @@ func (sdi *StateDiffIndexer) Close() error {
// Update the known gaps table with the gap information.
// LoadWatchedAddresses reads watched addresses from the database
func (sdi *StateDiffIndexer) LoadWatchedAddresses() ([]core.Address, error) {
func (sdi *StateDiffIndexer) LoadWatchedAddresses() ([]common.Address, error) {
addressStrings := make([]string, 0)
pgStr := "SELECT address FROM eth_meta.watched_addresses"
err := sdi.dbWriter.db.Select(sdi.ctx, &addressStrings, pgStr)
@@ -487,9 +492,9 @@ func (sdi *StateDiffIndexer) LoadWatchedAddresses() ([]core.Address, error) {
return nil, fmt.Errorf("error loading watched addresses: %v", err)
}
watchedAddresses := []core.Address{}
watchedAddresses := []common.Address{}
for _, addressString := range addressStrings {
watchedAddresses = append(watchedAddresses, core.HexToAddress(addressString))
watchedAddresses = append(watchedAddresses, common.HexToAddress(addressString))
}
return watchedAddresses, nil
indexer/database/sql/interfaces.go
+1
View File
@@ -45,6 +45,7 @@ type Driver interface {
// Statements interface to accommodate different SQL query syntax
type Statements interface {
ExistsHeaderStm() string
InsertHeaderStm() string
InsertUncleStm() string
InsertTxStm() string
indexer/database/sql/postgres/database.go
+6
View File
@@ -17,6 +17,8 @@
package postgres
import (
"fmt"
"github.com/cerc-io/plugeth-statediff/indexer/database/sql"
"github.com/cerc-io/plugeth-statediff/indexer/shared/schema"
)
@@ -39,6 +41,10 @@ type DB struct {
sql.Driver
}
func (db *DB) ExistsHeaderStm() string {
return fmt.Sprintf("SELECT EXISTS(SELECT 1 from %s WHERE block_number = $1 AND block_hash = $2 LIMIT 1)", schema.TableHeader.Name)
}
// InsertHeaderStm satisfies the sql.Statements interface
// Stm == Statement
func (db *DB) InsertHeaderStm() string {
indexer/database/sql/writer.go
+6
View File
@@ -20,6 +20,7 @@ import (
"fmt"
"strconv"
"github.com/ethereum/go-ethereum/common"
"github.com/jackc/pgtype"
shopspring "github.com/jackc/pgtype/ext/shopspring-numeric"
"github.com/lib/pq"
@@ -46,6 +47,11 @@ func (w *Writer) Close() error {
return w.db.Close()
}
func (w *Writer) hasHeader(blockHash common.Hash, blockNumber uint64) (exists bool, err error) {
err = w.db.QueryRow(w.db.Context(), w.db.ExistsHeaderStm(), blockNumber, blockHash.String()).Scan(&exists)
return exists, err
}
/*
INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_ids, reward, state_root, tx_root, receipt_root, uncles_hash, bloom, timestamp, coinbase)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)