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

% Checkpoint - Added the feature to write to File if writing to DB errors out. NOT TESTED

Browse Source
This commit is contained in:
Abdul Rabbani 2022-03-25 16:50:43 -04:00
parent a2bbeaa1d1
commit 4774b01573
9 changed files with 97 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
statediff/indexer/constructor.go
View File

@ -35,7 +35,7 @@ import (
func NewStateDiffIndexer(ctx context.Context, chainConfig *params.ChainConfig, nodeInfo node.Info, config interfaces.Config) (sql.Database, interfaces.StateDiffIndexer, error) {
switch config.Type() {
case shared.FILE:
log.Info("Starting statediff service in SQL file writing mode")
log.Info("Creating a statediff indexer in SQL file writing mode")
fc, ok := config.(file.Config)
if !ok {
return nil, nil, fmt.Errorf("file config is not the correct type: got %T, expected %T", config, file.Config{})
@ -44,7 +44,7 @@ func NewStateDiffIndexer(ctx context.Context, chainConfig *params.ChainConfig, n
ind, err := file.NewStateDiffIndexer(ctx, chainConfig, fc)
return nil, ind, err
case shared.POSTGRES:
log.Info("Starting statediff service in Postgres writing mode")
log.Info("Creating a statediff service in Postgres writing mode")
pgc, ok := config.(postgres.Config)
if !ok {
return nil, nil, fmt.Errorf("postgres config is not the correct type: got %T, expected %T", config, postgres.Config{})
@ -69,7 +69,7 @@ func NewStateDiffIndexer(ctx context.Context, chainConfig *params.ChainConfig, n
ind, err := sql.NewStateDiffIndexer(ctx, chainConfig, db)
return db, ind, err
case shared.DUMP:
log.Info("Starting statediff service in data dump mode")
log.Info("Creating statediff indexer in data dump mode")
dumpc, ok := config.(dump.Config)
if !ok {
return nil, nil, fmt.Errorf("dump config is not the correct type: got %T, expected %T", config, dump.Config{})

2
statediff/indexer/database/dump/indexer.go
View File

@ -534,6 +534,6 @@ func (sdi *StateDiffIndexer) SetWatchedAddresses(args []sdtypes.WatchAddressArg,
func (sdi *StateDiffIndexer) ClearWatchedAddresses() error {
return nil
}
func (sdi *StateDiffIndexer) PushKnownGaps(startingBlockNumber *big.Int, endingBlockNumber *big.Int, checkedOut bool, processingKey int64) error {
func (sdi *StateDiffIndexer) PushKnownGaps(startingBlockNumber *big.Int, endingBlockNumber *big.Int, checkedOut bool, processingKey int64, index interfaces.StateDiffIndexer) error {
return nil
}

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

@ -257,3 +257,12 @@ func (sqw *SQLWriter) upsertStorageCID(storageCID models.StorageNodeModel) {
sqw.stmts <- []byte(fmt.Sprintf(storageInsert, storageCID.BlockNumber, storageCID.HeaderID, storageCID.StatePath, storageKey, storageCID.CID,
storageCID.Path, storageCID.NodeType, true, storageCID.MhKey))
}
func (sqw *SQLWriter) upsertKnownGaps(knownGaps models.KnownGapsModel) {
sqw.stmts <- []byte(fmt.Sprintf(knownGaps.StartingBlockNumber, knownGaps.EndingBlockNumber, knownGaps.CheckedOut, knownGaps.ProcessingKey,
knownGaps.EndingBlockNumber, knownGaps.ProcessingKey, knownGaps.EndingBlockNumber))
//knownGapsInsert = "INSERT INTO eth.known_gaps (starting_block_number, ending_block_number, checked_out, processing_key) " +
// "VALUES ('%s', '%s', %t, %d) " +
// "ON CONFLICT (starting_block_number) DO UPDATE SET (ending_block_number, processing_key) = ('%s', %d) " +
// "WHERE eth.known_gaps.ending_block_number <= '%s';\n"
}

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

@ -732,7 +732,7 @@ func isGap(latestBlockInDb *big.Int, latestBlockOnChain *big.Int, expectedDiffer
// TODO:
// REmove the return value
// Write to file if err in writing to DB
func (sdi *StateDiffIndexer) FindAndUpdateGaps(latestBlockOnChain *big.Int, expectedDifference *big.Int, processingKey int64) error {
func (sdi *StateDiffIndexer) FindAndUpdateGaps(latestBlockOnChain *big.Int, expectedDifference *big.Int, processingKey int64, fileIndexer interfaces.StateDiffIndexer) error {
dbQueryString := "SELECT MAX(block_number) FROM eth.header_cids"
latestBlockInDb, err := sdi.QueryDbToBigInt(dbQueryString)
if err != nil {
@ -747,7 +747,7 @@ func (sdi *StateDiffIndexer) FindAndUpdateGaps(latestBlockOnChain *big.Int, expe
endBlock.Sub(latestBlockOnChain, expectedDifference)
log.Warn(fmt.Sprint("Found Gaps starting at, ", startBlock, " and ending at, ", endBlock))
err := sdi.PushKnownGaps(startBlock, endBlock, false, processingKey)
err := sdi.PushKnownGaps(startBlock, endBlock, false, processingKey, fileIndexer)
if err != nil {
// Write to file SQL file instead!!!
// If write to SQL file fails, write to disk. Handle this within the write to SQL file function!

20
statediff/indexer/database/sql/indexer_shared_test.go
View File

@ -3,18 +3,21 @@ package sql_test
import (
"bytes"
"context"
"errors"
"fmt"
"math/big"
"os"
"testing"
"github.com/ipfs/go-cid"
"github.com/jmoiron/sqlx"
"github.com/multiformats/go-multihash"
"github.com/stretchr/testify/require"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/statediff/indexer/database/file"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
@ -24,6 +27,7 @@ import (
var (
db sql.Database
sqlxdb *sqlx.DB
err error
ind interfaces.StateDiffIndexer
chainConf = params.MainnetChainConfig
@ -179,6 +183,16 @@ func setupDb(t *testing.T) (*sql.StateDiffIndexer, error) {
return stateDiff, err
}
func setupFile(t *testing.T) interfaces.StateDiffIndexer {
if _, err := os.Stat(file.TestConfig.FilePath); !errors.Is(err, os.ErrNotExist) {
err := os.Remove(file.TestConfig.FilePath)
require.NoError(t, err)
}
ind, err = file.NewStateDiffIndexer(context.Background(), mocks.TestConfig, file.TestConfig)
require.NoError(t, err)
return ind
}
func tearDown(t *testing.T) {
sql.TearDownDB(t, db)
err := ind.Close()
@ -197,6 +211,8 @@ func testKnownGapsUpsert(t *testing.T) {
t.Fatal(err)
}
fileInd := setupFile(t)
// Get the latest block from the DB
latestBlockInDb, err := stateDiff.QueryDbToBigInt("SELECT MAX(block_number) FROM eth.header_cids")
if err != nil {
@ -210,7 +226,7 @@ func testKnownGapsUpsert(t *testing.T) {
t.Log("The latest block on the chain is: ", latestBlockOnChain)
t.Log("The latest block on the DB is: ", latestBlockInDb)
gapUpsertErr := stateDiff.FindAndUpdateGaps(latestBlockOnChain, expectedDifference, 0)
gapUpsertErr := stateDiff.FindAndUpdateGaps(latestBlockOnChain, expectedDifference, 0, fileInd)
require.NoError(t, gapUpsertErr)
// Calculate what the start and end block should be in known_gaps
@ -223,7 +239,7 @@ func testKnownGapsUpsert(t *testing.T) {
queryString := fmt.Sprintf("SELECT starting_block_number from eth.known_gaps WHERE starting_block_number = %d AND ending_block_number = %d", startBlock, endBlock)
_, queryErr := stateDiff.QueryDb(queryString) // Figure out the string.
t.Logf("Updated Known Gaps table starting from, %d, and ending at, %d", startBlock, endBlock)
require.NoError(t, queryErr)
t.Logf("Updated Known Gaps table starting from, %d, and ending at, %d", startBlock, endBlock)
}

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

@ -110,6 +110,8 @@ func (db *DB) InsertKnownGapsStm() string {
// InsertKnownGapsStm satisfies the sql.Statements interface
func (db *DB) InsertKnownGapsStm() string {
return `INSERT INTO eth.known_gaps (starting_block_number, ending_block_number, checked_out, processing_key) VALUES ($1, $2, $3, $4) ON CONFLICT (starting_block_number) DO NOTHING`
return `INSERT INTO eth.known_gaps (starting_block_number, ending_block_number, checked_out, processing_key) VALUES ($1, $2, $3, $4)
ON CONFLICT (starting_block_number) DO UPDATE SET (ending_block_number, processing_key) = ($2, $4)
WHERE eth.known_gaps.ending_block_number <= $2`
//return `INSERT INTO eth.known_gaps (starting_block_number, ending_block_number, checked_out, processing_key) VALUES (1, 2, true, 1)`
}

26
statediff/service.go
View File

@ -148,7 +148,8 @@ type Service struct {
maxRetry uint
}
type KnownGaps struct {
// This structure keeps track of the knownGaps at any given moment in time
type KnownGapsState struct {
// Should we check for gaps by looking at the DB and comparing the latest block with head
checkForGaps bool
// Arbitrary processingKey that can be used down the line to differentiate different geth nodes.
@ -167,6 +168,9 @@ type KnownGaps struct {
// The last processed block keeps track of the last processed block.
// Its used to make sure we didn't skip over any block!
lastProcessedBlock *big.Int
// This fileIndexer is used to write the knownGaps to file
// If we can't properly write to DB
fileIndexer interfaces.StateDiffIndexer
}
// This function will capture any missed blocks that were not captured in sds.KnownGaps.knownErrorBlocks.
@ -202,7 +206,7 @@ func (sds *Service) capturedMissedBlocks(currentBlock *big.Int, knownErrorBlocks
startBlock := big.NewInt(0).Add(endErrorBlock, sds.KnownGaps.expectedDifference)
// 121 to 124
log.Warn(fmt.Sprintf("Adding the following block range to known_gaps table: %d - %d", startBlock, expectedEndErrorBlock))
sds.indexer.PushKnownGaps(startBlock, expectedEndErrorBlock, false, sds.KnownGaps.processingKey)
sds.indexer.PushKnownGaps(startBlock, expectedEndErrorBlock, false, sds.KnownGaps.processingKey, sds.KnownGaps.fileIndexer)
}
if expectedStartErrorBlock.Cmp(startErrorBlock) == -1 {
@ -214,12 +218,12 @@ func (sds *Service) capturedMissedBlocks(currentBlock *big.Int, knownErrorBlocks
endBlock := big.NewInt(0).Sub(startErrorBlock, sds.KnownGaps.expectedDifference)
// 111 to 114
log.Warn(fmt.Sprintf("Adding the following block range to known_gaps table: %d - %d", expectedStartErrorBlock, endBlock))
sds.indexer.PushKnownGaps(expectedStartErrorBlock, endBlock, false, sds.KnownGaps.processingKey)
sds.indexer.PushKnownGaps(expectedStartErrorBlock, endBlock, false, sds.KnownGaps.processingKey, sds.KnownGaps.fileIndexer)
}
log.Warn(fmt.Sprint("The following Gaps were found: ", knownErrorBlocks))
log.Warn(fmt.Sprint("Updating known Gaps table from ", startErrorBlock, " to ", endErrorBlock, " with processing key, ", sds.KnownGaps.processingKey))
sds.indexer.PushKnownGaps(startErrorBlock, endErrorBlock, false, sds.KnownGaps.processingKey)
sds.indexer.PushKnownGaps(startErrorBlock, endErrorBlock, false, sds.KnownGaps.processingKey, sds.KnownGaps.fileIndexer)
} else {
log.Warn("We missed blocks without any errors.")
@ -229,7 +233,7 @@ func (sds *Service) capturedMissedBlocks(currentBlock *big.Int, knownErrorBlocks
endBlock := big.NewInt(0).Sub(currentBlock, sds.KnownGaps.expectedDifference)
log.Warn(fmt.Sprint("Missed blocks starting from: ", startBlock))
log.Warn(fmt.Sprint("Missed blocks ending at: ", endBlock))
sds.indexer.PushKnownGaps(startBlock, endBlock, false, sds.KnownGaps.processingKey)
sds.indexer.PushKnownGaps(startBlock, endBlock, false, sds.KnownGaps.processingKey, sds.KnownGaps.fileIndexer)
}
}
@ -268,6 +272,18 @@ func New(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params
if err != nil {
return err
}
if params.IndexerConfig.Type() != shared.FILE {
fileIndexer, err = ind.NewStateDiffIndexer(params.Context, blockChain.Config(), info, params.IndexerConfig, "")
log.Info("Starting the statediff service in ", "mode", params.IndexerConfig.Type())
if err != nil {
return err
}
} else {
log.Info("Starting the statediff service in ", "mode", "File")
fileIndexer = indexer
}
//fileIndexer, fileErr = file.NewStateDiffIndexer(params.Context, blockChain.Config(), info)
indexer.ReportDBMetrics(10*time.Second, quitCh)
}

53
statediff/service_public_test.go
View File

@ -2,14 +2,18 @@ package statediff
import (
"context"
"errors"
"fmt"
"math/big"
"os"
"testing"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/statediff/indexer/database/file"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
"github.com/ethereum/go-ethereum/statediff/indexer/mocks"
"github.com/stretchr/testify/require"
)
@ -20,6 +24,9 @@ var (
chainConf = params.MainnetChainConfig
)
// Add clean db
// Test for failures when they are expected, when we go from smaller block to larger block
// We should no longer see the smaller block in DB
func TestKnownGaps(t *testing.T) {
type gapValues struct {
lastProcessedBlock int64
@ -27,40 +34,48 @@ func TestKnownGaps(t *testing.T) {
knownErrorBlocksStart int64
knownErrorBlocksEnd int64
expectedDif int64
processingKey int64
}
tests := []gapValues{
// Unprocessed gaps before and after knownErrorBlock
{lastProcessedBlock: 110, knownErrorBlocksStart: 115, knownErrorBlocksEnd: 120, currentBlock: 125, expectedDif: 1},
{lastProcessedBlock: 110, knownErrorBlocksStart: 115, knownErrorBlocksEnd: 120, currentBlock: 125, expectedDif: 1, processingKey: 1},
// No knownErrorBlocks
{lastProcessedBlock: 130, knownErrorBlocksStart: 0, knownErrorBlocksEnd: 0, currentBlock: 140, expectedDif: 1},
{lastProcessedBlock: 130, knownErrorBlocksStart: 0, knownErrorBlocksEnd: 0, currentBlock: 140, expectedDif: 1, processingKey: 1},
// No gaps before or after knownErrorBlocks
{lastProcessedBlock: 150, knownErrorBlocksStart: 151, knownErrorBlocksEnd: 159, currentBlock: 160, expectedDif: 1},
{lastProcessedBlock: 150, knownErrorBlocksStart: 151, knownErrorBlocksEnd: 159, currentBlock: 160, expectedDif: 1, processingKey: 1},
// gaps before knownErrorBlocks but not after
{lastProcessedBlock: 170, knownErrorBlocksStart: 180, knownErrorBlocksEnd: 189, currentBlock: 190, expectedDif: 1},
{lastProcessedBlock: 170, knownErrorBlocksStart: 180, knownErrorBlocksEnd: 189, currentBlock: 190, expectedDif: 1, processingKey: 1},
// gaps after knownErrorBlocks but not before
{lastProcessedBlock: 200, knownErrorBlocksStart: 201, knownErrorBlocksEnd: 205, currentBlock: 210, expectedDif: 1},
{lastProcessedBlock: 200, knownErrorBlocksStart: 201, knownErrorBlocksEnd: 205, currentBlock: 210, expectedDif: 1, processingKey: 1},
/// Same tests as above with a new expected DIF
// Unprocessed gaps before and after knownErrorBlock
{lastProcessedBlock: 1100, knownErrorBlocksStart: 1150, knownErrorBlocksEnd: 1200, currentBlock: 1250, expectedDif: 2},
{lastProcessedBlock: 1100, knownErrorBlocksStart: 1150, knownErrorBlocksEnd: 1200, currentBlock: 1250, expectedDif: 2, processingKey: 2},
// No knownErrorBlocks
{lastProcessedBlock: 1300, knownErrorBlocksStart: 0, knownErrorBlocksEnd: 0, currentBlock: 1400, expectedDif: 2},
{lastProcessedBlock: 1300, knownErrorBlocksStart: 0, knownErrorBlocksEnd: 0, currentBlock: 1400, expectedDif: 2, processingKey: 2},
// No gaps before or after knownErrorBlocks
{lastProcessedBlock: 1500, knownErrorBlocksStart: 1502, knownErrorBlocksEnd: 1598, currentBlock: 1600, expectedDif: 2},
{lastProcessedBlock: 1500, knownErrorBlocksStart: 1502, knownErrorBlocksEnd: 1598, currentBlock: 1600, expectedDif: 2, processingKey: 2},
// gaps before knownErrorBlocks but not after
{lastProcessedBlock: 1700, knownErrorBlocksStart: 1800, knownErrorBlocksEnd: 1898, currentBlock: 1900, expectedDif: 2},
{lastProcessedBlock: 1700, knownErrorBlocksStart: 1800, knownErrorBlocksEnd: 1898, currentBlock: 1900, expectedDif: 2, processingKey: 2},
// gaps after knownErrorBlocks but not before
{lastProcessedBlock: 2000, knownErrorBlocksStart: 2002, knownErrorBlocksEnd: 2050, currentBlock: 2100, expectedDif: 2},
{lastProcessedBlock: 2000, knownErrorBlocksStart: 2002, knownErrorBlocksEnd: 2050, currentBlock: 2100, expectedDif: 2, processingKey: 2},
// Test update when block number is larger!!
{lastProcessedBlock: 2000, knownErrorBlocksStart: 2002, knownErrorBlocksEnd: 2052, currentBlock: 2100, expectedDif: 2, processingKey: 2},
{lastProcessedBlock: 2000, knownErrorBlocksStart: 2002, knownErrorBlocksEnd: 2052, currentBlock: 2100, expectedDif: 2, processingKey: 10},
}
for _, tc := range tests {
// Reuse processing key from expecteDiff
testCaptureMissedBlocks(t, tc.lastProcessedBlock, tc.currentBlock, tc.knownErrorBlocksStart, tc.knownErrorBlocksEnd, tc.expectedDif, tc.expectedDif)
testCaptureMissedBlocks(t, tc.lastProcessedBlock, tc.currentBlock, tc.knownErrorBlocksStart, tc.knownErrorBlocksEnd, tc.expectedDif, tc.expectedDif, false)
}
for _, tc := range tests {
// Reuse processing key from expecteDiff
testCaptureMissedBlocks(t, tc.lastProcessedBlock, tc.currentBlock, tc.knownErrorBlocksStart, tc.knownErrorBlocksEnd, tc.expectedDif, tc.expectedDif, true)
}
}
// It also makes sure we properly calculate any missed gaps not in the known gaps lists
// either before or after the list.
func testCaptureMissedBlocks(t *testing.T, lastBlockProcessed int64, currentBlockNum int64, knownErrorBlocksStart int64, knownErrorBlocksEnd int64, expectedDif int64, processingKey int64) {
func testCaptureMissedBlocks(t *testing.T, lastBlockProcessed int64, currentBlockNum int64, knownErrorBlocksStart int64, knownErrorBlocksEnd int64, expectedDif int64, processingKey int64, skipDb bool) {
lastProcessedBlock := big.NewInt(lastBlockProcessed)
currentBlock := big.NewInt(currentBlockNum)
@ -74,9 +89,11 @@ func testCaptureMissedBlocks(t *testing.T, lastBlockProcessed int64, currentBloc
}
// Comment out values which should not be used
knownGaps := KnownGaps{
fileInd := setupFile(t)
knownGaps := KnownGapsState{
processingKey: processingKey,
expectedDifference: big.NewInt(expectedDif),
fileIndexer: fileInd,
}
stateDiff, err := setupDb(t)
if err != nil {
@ -134,6 +151,16 @@ func setupDb(t *testing.T) (*sql.StateDiffIndexer, error) {
return stateDiff, err
}
func setupFile(t *testing.T) interfaces.StateDiffIndexer {
if _, err := os.Stat(file.TestConfig.FilePath); !errors.Is(err, os.ErrNotExist) {
err := os.Remove(file.TestConfig.FilePath)
require.NoError(t, err)
}
indexer, err = file.NewStateDiffIndexer(context.Background(), mocks.TestConfig, file.TestConfig)
require.NoError(t, err)
return indexer
}
// Teardown the DB
func tearDown(t *testing.T, stateDiff *sql.StateDiffIndexer) {
t.Log("Starting tearDown")

0
statediff/statediffing_test_file.sql Normal file
View File