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

fix unit tests

Browse Source
This commit is contained in:
i-norden 2021-11-17 19:37:41 -06:00
parent aa984711bd
commit 4ee8214562
6 changed files with 84 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
statediff/indexer/database/sql/pgx_indexer_legacy_test.go
View File

@ -20,7 +20,6 @@ import (
"context" "context"
"testing" "testing"
"github.com/jmoiron/sqlx"
"github.com/multiformats/go-multihash" "github.com/multiformats/go-multihash"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@ -64,20 +63,21 @@ func TestPGXIndexerLegacy(t *testing.T) {
t.Run("Publish and index header IPLDs in a legacy tx", func(t *testing.T) { t.Run("Publish and index header IPLDs in a legacy tx", func(t *testing.T) {
setupLegacyPGX(t) setupLegacyPGX(t)
defer tearDown(t) defer tearDown(t)
pgStr := `SELECT cid, td, reward, id, base_fee pgStr := `SELECT cid, cast(td AS TEXT), cast(reward AS TEXT), block_hash, base_fee
FROM eth.header_cids FROM eth.header_cids
WHERE block_number = $1` WHERE block_number = $1`
// check header was properly indexed // check header was properly indexed
type res struct { type res struct {
CID string CID string
TD string TD string
Reward string Reward string
ID int BlockHash string `db:"block_hash"`
BaseFee *int64 `db:"base_fee"` BaseFee *int64 `db:"base_fee"`
} }
header := new(res) header := new(res)
err = db.QueryRow(context.Background(), pgStr, legacyData.BlockNumber.Uint64()).(*sqlx.Row).StructScan(header) err = db.QueryRow(context.Background(), pgStr, legacyData.BlockNumber.Uint64()).Scan(
&header.CID, &header.TD, &header.Reward, &header.BlockHash, &header.BaseFee)
require.NoError(t, err) require.NoError(t, err)
test_helpers.ExpectEqual(t, header.CID, legacyHeaderCID.String()) test_helpers.ExpectEqual(t, header.CID, legacyHeaderCID.String())

60
statediff/indexer/database/sql/pgx_indexer_test.go
View File

@ -140,7 +140,7 @@ func setupPGX(t *testing.T) {
} }
}() }()
for _, node := range mocks.StateDiffs { for _, node := range mocks.StateDiffs {
err = ind.PushStateNode(tx, node) err = ind.PushStateNode(tx, node, mockBlock.Hash().String())
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -153,16 +153,16 @@ func TestPGXIndexer(t *testing.T) {
t.Run("Publish and index header IPLDs in a single tx", func(t *testing.T) { t.Run("Publish and index header IPLDs in a single tx", func(t *testing.T) {
setupPGX(t) setupPGX(t)
defer tearDown(t) defer tearDown(t)
pgStr := `SELECT cid, td, reward, id, base_fee pgStr := `SELECT cid, td, reward, block_hash, base_fee
FROM eth.header_cids FROM eth.header_cids
WHERE block_number = $1` WHERE block_number = $1`
// check header was properly indexed // check header was properly indexed
type res struct { type res struct {
CID string CID string
TD string TD string
Reward string Reward string
ID int BlockHash string `db:"block_hash"`
BaseFee *int64 `db:"base_fee"` BaseFee *int64 `db:"base_fee"`
} }
header := new(res) header := new(res)
err = db.QueryRow(context.Background(), pgStr, mocks.BlockNumber.Uint64()).(*sqlx.Row).StructScan(header) err = db.QueryRow(context.Background(), pgStr, mocks.BlockNumber.Uint64()).(*sqlx.Row).StructScan(header)
@ -192,7 +192,7 @@ func TestPGXIndexer(t *testing.T) {
defer tearDown(t) defer tearDown(t)
// check that txs were properly indexed // check that txs were properly indexed
trxs := make([]string, 0) trxs := make([]string, 0)
pgStr := `SELECT transaction_cids.cid FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.id) pgStr := `SELECT transaction_cids.cid FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.block_hash)
WHERE header_cids.block_number = $1` WHERE header_cids.block_number = $1`
err = db.Select(context.Background(), &trxs, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &trxs, pgStr, mocks.BlockNumber.Uint64())
if err != nil { if err != nil {
@ -260,7 +260,7 @@ func TestPGXIndexer(t *testing.T) {
t.Fatalf("expected AccessListTxType (1), got %d", *txType) t.Fatalf("expected AccessListTxType (1), got %d", *txType)
} }
accessListElementModels := make([]models.AccessListElementModel, 0) accessListElementModels := make([]models.AccessListElementModel, 0)
pgStr = `SELECT access_list_element.* FROM eth.access_list_element INNER JOIN eth.transaction_cids ON (tx_id = transaction_cids.id) WHERE cid = $1 ORDER BY access_list_element.index ASC` pgStr = `SELECT access_list_element.* FROM eth.access_list_element INNER JOIN eth.transaction_cids ON (tx_id = transaction_cids.tx_hash) WHERE cid = $1 ORDER BY access_list_element.index ASC`
err = db.Select(context.Background(), &accessListElementModels, pgStr, c) err = db.Select(context.Background(), &accessListElementModels, pgStr, c)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@ -299,8 +299,8 @@ func TestPGXIndexer(t *testing.T) {
rcts := make([]string, 0) rcts := make([]string, 0)
pgStr := `SELECT receipt_cids.leaf_cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids pgStr := `SELECT receipt_cids.leaf_cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids
WHERE receipt_cids.tx_id = transaction_cids.id WHERE receipt_cids.tx_id = transaction_cids.tx_hash
AND transaction_cids.header_id = header_cids.id AND transaction_cids.header_id = header_cids.block_hash
AND header_cids.block_number = $1 AND header_cids.block_number = $1
ORDER BY transaction_cids.index` ORDER BY transaction_cids.index`
err = db.Select(context.Background(), &rcts, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &rcts, pgStr, mocks.BlockNumber.Uint64())
@ -317,8 +317,8 @@ func TestPGXIndexer(t *testing.T) {
} }
for i := range rcts { for i := range rcts {
results := make([]logIPLD, 0) results := make([]logIPLD, 0)
pgStr = `SELECT log_cids.index, log_cids.address, log_cids.Topic0, log_cids.Topic1, data FROM eth.log_cids pgStr = `SELECT log_cids.index, log_cids.address, log_cids.topic0, log_cids.topic1, data FROM eth.log_cids
INNER JOIN eth.receipt_cids ON (log_cids.receipt_id = receipt_cids.id) INNER JOIN eth.receipt_cids ON (log_cids.rct_id = receipt_cids.tx_id)
INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key) INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key)
WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC` WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC`
err = db.Select(context.Background(), &results, pgStr, rcts[i]) err = db.Select(context.Background(), &results, pgStr, rcts[i])
@ -350,9 +350,9 @@ func TestPGXIndexer(t *testing.T) {
// check receipts were properly indexed // check receipts were properly indexed
rcts := make([]string, 0) rcts := make([]string, 0)
pgStr := `SELECT receipt_cids.leaf_cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids pgStr := `SELECT receipt_cids.leaf_cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids
WHERE receipt_cids.tx_id = transaction_cids.id WHERE receipt_cids.tx_id = transaction_cids.tx_hash
AND transaction_cids.header_id = header_cids.id AND transaction_cids.header_id = header_cids.block_hash
AND header_cids.block_number = $1 order by transaction_cids.id` AND header_cids.block_number = $1 order by transaction_cids.index`
err = db.Select(context.Background(), &rcts, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &rcts, pgStr, mocks.BlockNumber.Uint64())
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@ -447,8 +447,8 @@ func TestPGXIndexer(t *testing.T) {
defer tearDown(t) defer tearDown(t)
// check that state nodes were properly indexed and published // check that state nodes were properly indexed and published
stateNodes := make([]models.StateNodeModel, 0) stateNodes := make([]models.StateNodeModel, 0)
pgStr := `SELECT state_cids.id, state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.id) FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
WHERE header_cids.block_number = $1 AND node_type != 3` WHERE header_cids.block_number = $1 AND node_type != 3`
err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
if err != nil { if err != nil {
@ -467,9 +467,9 @@ func TestPGXIndexer(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
pgStr = `SELECT * from eth.state_accounts WHERE state_id = $1` pgStr = `SELECT * from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
var account models.StateAccountModel var account models.StateAccountModel
err = db.Get(context.Background(), &account, pgStr, stateNode.ID) err = db.Get(context.Background(), &account, pgStr, stateNode.HeaderID, stateNode.Path)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -479,8 +479,8 @@ func TestPGXIndexer(t *testing.T) {
test_helpers.ExpectEqual(t, stateNode.Path, []byte{'\x06'}) test_helpers.ExpectEqual(t, stateNode.Path, []byte{'\x06'})
test_helpers.ExpectEqual(t, data, mocks.ContractLeafNode) test_helpers.ExpectEqual(t, data, mocks.ContractLeafNode)
test_helpers.ExpectEqual(t, account, models.StateAccountModel{ test_helpers.ExpectEqual(t, account, models.StateAccountModel{
ID: account.ID, HeaderID: account.HeaderID,
StateID: stateNode.ID, StatePath: stateNode.Path,
Balance: "0", Balance: "0",
CodeHash: mocks.ContractCodeHash.Bytes(), CodeHash: mocks.ContractCodeHash.Bytes(),
StorageRoot: mocks.ContractRoot, StorageRoot: mocks.ContractRoot,
@ -493,8 +493,8 @@ func TestPGXIndexer(t *testing.T) {
test_helpers.ExpectEqual(t, stateNode.Path, []byte{'\x0c'}) test_helpers.ExpectEqual(t, stateNode.Path, []byte{'\x0c'})
test_helpers.ExpectEqual(t, data, mocks.AccountLeafNode) test_helpers.ExpectEqual(t, data, mocks.AccountLeafNode)
test_helpers.ExpectEqual(t, account, models.StateAccountModel{ test_helpers.ExpectEqual(t, account, models.StateAccountModel{
ID: account.ID, HeaderID: account.HeaderID,
StateID: stateNode.ID, StatePath: stateNode.Path,
Balance: "1000", Balance: "1000",
CodeHash: mocks.AccountCodeHash.Bytes(), CodeHash: mocks.AccountCodeHash.Bytes(),
StorageRoot: mocks.AccountRoot, StorageRoot: mocks.AccountRoot,
@ -505,8 +505,8 @@ func TestPGXIndexer(t *testing.T) {
// check that Removed state nodes were properly indexed and published // check that Removed state nodes were properly indexed and published
stateNodes = make([]models.StateNodeModel, 0) stateNodes = make([]models.StateNodeModel, 0)
pgStr = `SELECT state_cids.id, state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.id) FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
WHERE header_cids.block_number = $1 AND node_type = 3` WHERE header_cids.block_number = $1 AND node_type = 3`
err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
if err != nil { if err != nil {
@ -538,8 +538,8 @@ func TestPGXIndexer(t *testing.T) {
storageNodes := make([]models.StorageNodeWithStateKeyModel, 0) storageNodes := make([]models.StorageNodeWithStateKeyModel, 0)
pgStr := `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path pgStr := `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
FROM eth.storage_cids, eth.state_cids, eth.header_cids FROM eth.storage_cids, eth.state_cids, eth.header_cids
WHERE storage_cids.state_id = state_cids.id WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
AND state_cids.header_id = header_cids.id AND state_cids.header_id = header_cids.block_hash
AND header_cids.block_number = $1 AND header_cids.block_number = $1
AND storage_cids.node_type != 3` AND storage_cids.node_type != 3`
err = db.Select(context.Background(), &storageNodes, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &storageNodes, pgStr, mocks.BlockNumber.Uint64())
@ -571,8 +571,8 @@ func TestPGXIndexer(t *testing.T) {
storageNodes = make([]models.StorageNodeWithStateKeyModel, 0) storageNodes = make([]models.StorageNodeWithStateKeyModel, 0)
pgStr = `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path pgStr = `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
FROM eth.storage_cids, eth.state_cids, eth.header_cids FROM eth.storage_cids, eth.state_cids, eth.header_cids
WHERE storage_cids.state_id = state_cids.id WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
AND state_cids.header_id = header_cids.id AND state_cids.header_id = header_cids.block_hash
AND header_cids.block_number = $1 AND header_cids.block_number = $1
AND storage_cids.node_type = 3` AND storage_cids.node_type = 3`
err = db.Select(context.Background(), &storageNodes, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &storageNodes, pgStr, mocks.BlockNumber.Uint64())

14
statediff/indexer/database/sql/sqlx_indexer_legacy_test.go
View File

@ -62,7 +62,7 @@ func setupLegacySQLX(t *testing.T) {
} }
}() }()
for _, node := range legacyData.StateDiffs { for _, node := range legacyData.StateDiffs {
err = ind.PushStateNode(tx, node) err = ind.PushStateNode(tx, node, mockLegacyBlock.Hash().String())
require.NoError(t, err) require.NoError(t, err)
} }
@ -73,16 +73,16 @@ func TestSQLXIndexerLegacy(t *testing.T) {
t.Run("Publish and index header IPLDs in a legacy tx", func(t *testing.T) { t.Run("Publish and index header IPLDs in a legacy tx", func(t *testing.T) {
setupLegacySQLX(t) setupLegacySQLX(t)
defer tearDown(t) defer tearDown(t)
pgStr := `SELECT cid, td, reward, id, base_fee pgStr := `SELECT cid, td, reward, block_hash, base_fee
FROM eth.header_cids FROM eth.header_cids
WHERE block_number = $1` WHERE block_number = $1`
// check header was properly indexed // check header was properly indexed
type res struct { type res struct {
CID string CID string
TD string TD string
Reward string Reward string
ID int BlockHash string `db:"block_hash"`
BaseFee *int64 `db:"base_fee"` BaseFee *int64 `db:"base_fee"`
} }
header := new(res) header := new(res)
err = db.QueryRow(context.Background(), pgStr, legacyData.BlockNumber.Uint64()).(*sqlx.Row).StructScan(header) err = db.QueryRow(context.Background(), pgStr, legacyData.BlockNumber.Uint64()).(*sqlx.Row).StructScan(header)

58
statediff/indexer/database/sql/sqlx_indexer_test.go
View File

@ -179,16 +179,16 @@ func TestSQLXIndexer(t *testing.T) {
t.Run("Publish and index header IPLDs in a single tx", func(t *testing.T) { t.Run("Publish and index header IPLDs in a single tx", func(t *testing.T) {
setupSQLX(t) setupSQLX(t)
defer tearDown(t) defer tearDown(t)
pgStr := `SELECT cid, td, reward, id, base_fee pgStr := `SELECT cid, td, reward, block_hash, base_fee
FROM eth.header_cids FROM eth.header_cids
WHERE block_number = $1` WHERE block_number = $1`
// check header was properly indexed // check header was properly indexed
type res struct { type res struct {
CID string CID string
TD string TD string
Reward string Reward string
ID int BlockHash string `db:"block_hash"`
BaseFee *int64 `db:"base_fee"` BaseFee *int64 `db:"base_fee"`
} }
header := new(res) header := new(res)
err = db.QueryRow(context.Background(), pgStr, mocks.BlockNumber.Uint64()).(*sqlx.Row).StructScan(header) err = db.QueryRow(context.Background(), pgStr, mocks.BlockNumber.Uint64()).(*sqlx.Row).StructScan(header)
@ -218,7 +218,7 @@ func TestSQLXIndexer(t *testing.T) {
defer tearDown(t) defer tearDown(t)
// check that txs were properly indexed // check that txs were properly indexed
trxs := make([]string, 0) trxs := make([]string, 0)
pgStr := `SELECT transaction_cids.cid FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.id) pgStr := `SELECT transaction_cids.cid FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.block_hash)
WHERE header_cids.block_number = $1` WHERE header_cids.block_number = $1`
err = db.Select(context.Background(), &trxs, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &trxs, pgStr, mocks.BlockNumber.Uint64())
if err != nil { if err != nil {
@ -286,7 +286,7 @@ func TestSQLXIndexer(t *testing.T) {
t.Fatalf("expected AccessListTxType (1), got %d", txType) t.Fatalf("expected AccessListTxType (1), got %d", txType)
} }
accessListElementModels := make([]models.AccessListElementModel, 0) accessListElementModels := make([]models.AccessListElementModel, 0)
pgStr = `SELECT access_list_element.* FROM eth.access_list_element INNER JOIN eth.transaction_cids ON (tx_id = transaction_cids.id) WHERE cid = $1 ORDER BY access_list_element.index ASC` pgStr = `SELECT access_list_element.* FROM eth.access_list_element INNER JOIN eth.transaction_cids ON (tx_id = transaction_cids.tx_hash) WHERE cid = $1 ORDER BY access_list_element.index ASC`
err = db.Select(context.Background(), &accessListElementModels, pgStr, c) err = db.Select(context.Background(), &accessListElementModels, pgStr, c)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@ -325,8 +325,8 @@ func TestSQLXIndexer(t *testing.T) {
rcts := make([]string, 0) rcts := make([]string, 0)
pgStr := `SELECT receipt_cids.leaf_cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids pgStr := `SELECT receipt_cids.leaf_cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids
WHERE receipt_cids.tx_id = transaction_cids.id WHERE receipt_cids.tx_id = transaction_cids.tx_hash
AND transaction_cids.header_id = header_cids.id AND transaction_cids.header_id = header_cids.block_hash
AND header_cids.block_number = $1 AND header_cids.block_number = $1
ORDER BY transaction_cids.index` ORDER BY transaction_cids.index`
err = db.Select(context.Background(), &rcts, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &rcts, pgStr, mocks.BlockNumber.Uint64())
@ -343,8 +343,8 @@ func TestSQLXIndexer(t *testing.T) {
} }
for i := range rcts { for i := range rcts {
results := make([]logIPLD, 0) results := make([]logIPLD, 0)
pgStr = `SELECT log_cids.index, log_cids.address, log_cids.Topic0, log_cids.Topic1, data FROM eth.log_cids pgStr = `SELECT log_cids.index, log_cids.address, log_cids.topic0, log_cids.topic1, data FROM eth.log_cids
INNER JOIN eth.receipt_cids ON (log_cids.receipt_id = receipt_cids.id) INNER JOIN eth.receipt_cids ON (log_cids.rct_id = receipt_cids.tx_id)
INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key) INNER JOIN public.blocks ON (log_cids.leaf_mh_key = blocks.key)
WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC` WHERE receipt_cids.leaf_cid = $1 ORDER BY eth.log_cids.index ASC`
err = db.Select(context.Background(), &results, pgStr, rcts[i]) err = db.Select(context.Background(), &results, pgStr, rcts[i])
@ -376,9 +376,9 @@ func TestSQLXIndexer(t *testing.T) {
// check receipts were properly indexed // check receipts were properly indexed
rcts := make([]string, 0) rcts := make([]string, 0)
pgStr := `SELECT receipt_cids.leaf_cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids pgStr := `SELECT receipt_cids.leaf_cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids
WHERE receipt_cids.tx_id = transaction_cids.id WHERE receipt_cids.tx_id = transaction_cids.tx_hash
AND transaction_cids.header_id = header_cids.id AND transaction_cids.header_id = header_cids.block_hash
AND header_cids.block_number = $1 order by transaction_cids.id` AND header_cids.block_number = $1 order by transaction_cids.index`
err = db.Select(context.Background(), &rcts, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &rcts, pgStr, mocks.BlockNumber.Uint64())
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@ -472,8 +472,8 @@ func TestSQLXIndexer(t *testing.T) {
defer tearDown(t) defer tearDown(t)
// check that state nodes were properly indexed and published // check that state nodes were properly indexed and published
stateNodes := make([]models.StateNodeModel, 0) stateNodes := make([]models.StateNodeModel, 0)
pgStr := `SELECT state_cids.id, state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id pgStr := `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.id) FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
WHERE header_cids.block_number = $1 AND node_type != 3` WHERE header_cids.block_number = $1 AND node_type != 3`
err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
if err != nil { if err != nil {
@ -492,9 +492,9 @@ func TestSQLXIndexer(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
pgStr = `SELECT * from eth.state_accounts WHERE state_id = $1` pgStr = `SELECT * from eth.state_accounts WHERE header_id = $1 AND state_path = $2`
var account models.StateAccountModel var account models.StateAccountModel
err = db.Get(context.Background(), &account, pgStr, stateNode.ID) err = db.Get(context.Background(), &account, pgStr, stateNode.HeaderID, stateNode.Path)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -504,8 +504,8 @@ func TestSQLXIndexer(t *testing.T) {
test_helpers.ExpectEqual(t, stateNode.Path, []byte{'\x06'}) test_helpers.ExpectEqual(t, stateNode.Path, []byte{'\x06'})
test_helpers.ExpectEqual(t, data, mocks.ContractLeafNode) test_helpers.ExpectEqual(t, data, mocks.ContractLeafNode)
test_helpers.ExpectEqual(t, account, models.StateAccountModel{ test_helpers.ExpectEqual(t, account, models.StateAccountModel{
ID: account.ID, HeaderID: account.HeaderID,
StateID: stateNode.ID, StatePath: stateNode.Path,
Balance: "0", Balance: "0",
CodeHash: mocks.ContractCodeHash.Bytes(), CodeHash: mocks.ContractCodeHash.Bytes(),
StorageRoot: mocks.ContractRoot, StorageRoot: mocks.ContractRoot,
@ -518,8 +518,8 @@ func TestSQLXIndexer(t *testing.T) {
test_helpers.ExpectEqual(t, stateNode.Path, []byte{'\x0c'}) test_helpers.ExpectEqual(t, stateNode.Path, []byte{'\x0c'})
test_helpers.ExpectEqual(t, data, mocks.AccountLeafNode) test_helpers.ExpectEqual(t, data, mocks.AccountLeafNode)
test_helpers.ExpectEqual(t, account, models.StateAccountModel{ test_helpers.ExpectEqual(t, account, models.StateAccountModel{
ID: account.ID, HeaderID: account.HeaderID,
StateID: stateNode.ID, StatePath: stateNode.Path,
Balance: "1000", Balance: "1000",
CodeHash: mocks.AccountCodeHash.Bytes(), CodeHash: mocks.AccountCodeHash.Bytes(),
StorageRoot: mocks.AccountRoot, StorageRoot: mocks.AccountRoot,
@ -530,8 +530,8 @@ func TestSQLXIndexer(t *testing.T) {
// check that Removed state nodes were properly indexed and published // check that Removed state nodes were properly indexed and published
stateNodes = make([]models.StateNodeModel, 0) stateNodes = make([]models.StateNodeModel, 0)
pgStr = `SELECT state_cids.id, state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.id) FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.block_hash)
WHERE header_cids.block_number = $1 AND node_type = 3` WHERE header_cids.block_number = $1 AND node_type = 3`
err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &stateNodes, pgStr, mocks.BlockNumber.Uint64())
if err != nil { if err != nil {
@ -563,8 +563,8 @@ func TestSQLXIndexer(t *testing.T) {
storageNodes := make([]models.StorageNodeWithStateKeyModel, 0) storageNodes := make([]models.StorageNodeWithStateKeyModel, 0)
pgStr := `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path pgStr := `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
FROM eth.storage_cids, eth.state_cids, eth.header_cids FROM eth.storage_cids, eth.state_cids, eth.header_cids
WHERE storage_cids.state_id = state_cids.id WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
AND state_cids.header_id = header_cids.id AND state_cids.header_id = header_cids.block_hash
AND header_cids.block_number = $1 AND header_cids.block_number = $1
AND storage_cids.node_type != 3` AND storage_cids.node_type != 3`
err = db.Select(context.Background(), &storageNodes, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &storageNodes, pgStr, mocks.BlockNumber.Uint64())
@ -596,8 +596,8 @@ func TestSQLXIndexer(t *testing.T) {
storageNodes = make([]models.StorageNodeWithStateKeyModel, 0) storageNodes = make([]models.StorageNodeWithStateKeyModel, 0)
pgStr = `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path pgStr = `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
FROM eth.storage_cids, eth.state_cids, eth.header_cids FROM eth.storage_cids, eth.state_cids, eth.header_cids
WHERE storage_cids.state_id = state_cids.id WHERE (storage_cids.state_path, storage_cids.header_id) = (state_cids.state_path, state_cids.header_id)
AND state_cids.header_id = header_cids.id AND state_cids.header_id = header_cids.block_hash
AND header_cids.block_number = $1 AND header_cids.block_number = $1
AND storage_cids.node_type = 3` AND storage_cids.node_type = 3`
err = db.Select(context.Background(), &storageNodes, pgStr, mocks.BlockNumber.Uint64()) err = db.Select(context.Background(), &storageNodes, pgStr, mocks.BlockNumber.Uint64())

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

@ -42,7 +42,7 @@ func NewWriter(db Database) *Writer {
/* /*
INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, base_fee) INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, base_fee)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
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, base_fee) = ($3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, eth.header_cids.times_validated + 1, $16) ON CONFLICT (block_hash) DO UPDATE SET (block_number, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, base_fee) = ($1, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, eth.header_cids.times_validated + 1, $16)
*/ */
func (in *Writer) upsertHeaderCID(tx Tx, header models.HeaderModel) error { func (in *Writer) upsertHeaderCID(tx Tx, header models.HeaderModel) error {
_, err := tx.Exec(in.db.Context(), in.db.InsertHeaderStm(), _, err := tx.Exec(in.db.Context(), in.db.InsertHeaderStm(),

9
statediff/service.go
View File

@ -239,8 +239,14 @@ func (sds *Service) WriteLoop(chainEventCh chan core.ChainEvent) {
statediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh))) statediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh)))
chainEventFwd <- chainEvent chainEventFwd <- chainEvent
case err := <-errCh: case err := <-errCh:
println("here")
log.Error("Error from chain event subscription", "error", err) log.Error("Error from chain event subscription", "error", err)
close(sds.QuitChan) close(sds.QuitChan)
log.Info("Quitting the statediffing writing loop")
if err := sds.indexer.Close(); err != nil {
log.Error("Error closing indexer", "err", err)
}
return
case <-sds.QuitChan: case <-sds.QuitChan:
log.Info("Quitting the statediffing writing loop") log.Info("Quitting the statediffing writing loop")
if err := sds.indexer.Close(); err != nil { if err := sds.indexer.Close(); err != nil {
@ -339,6 +345,9 @@ func (sds *Service) Loop(chainEventCh chan core.ChainEvent) {
case err := <-errCh: case err := <-errCh:
log.Error("Error from chain event subscription", "error", err) log.Error("Error from chain event subscription", "error", err)
close(sds.QuitChan) close(sds.QuitChan)
log.Info("Quitting the statediffing listening loop")
sds.close()
return
case <-sds.QuitChan: case <-sds.QuitChan:
log.Info("Quitting the statediffing listening loop") log.Info("Quitting the statediffing listening loop")
sds.close() sds.close()