status-im
/
status-go
mirror of https://github.com/status-im/status-go.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

Basic test for findBlocksCommand

This commit is contained in:
Roman Volosovskyi 2023-09-19 13:17:36 +02:00
parent ad971278d9
commit 1dca3adb89
No known key found for this signature in database
GPG Key ID: 0238A4B5ECEE70DE
7 changed files with 288 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
rpc/chain/client.go
View File

@ -25,8 +25,23 @@ type FeeHistory struct {
BaseFeePerGas []string `json:"baseFeePerGas"` BaseFeePerGas []string `json:"baseFeePerGas"`
} }
type BatchCallClient interface {
BatchCallContext(ctx context.Context, b []rpc.BatchElem) error
}
type ClientInterface interface { type ClientInterface interface {
BatchCallContext(ctx context.Context, b []rpc.BatchElem) error BatchCallContext(ctx context.Context, b []rpc.BatchElem) error
HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error)
BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error)
BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error)
NonceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (uint64, error)
FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error)
BalanceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (*big.Int, error)
HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error)
FullTransactionByBlockNumberAndIndex(ctx context.Context, blockNumber *big.Int, index uint) (*FullTransaction, error)
GetBaseFeeFromBlock(blockNumber *big.Int) (string, error)
NetworkID() uint64
ToBigInt() *big.Int
} }
type ClientWithFallback struct { type ClientWithFallback struct {

4
rpc/client.go
View File

@ -34,7 +34,7 @@ var (
type Handler func(context.Context, uint64, ...interface{}) (interface{}, error) type Handler func(context.Context, uint64, ...interface{}) (interface{}, error)
type ClientInterface interface { type ClientInterface interface {
AbstractEthClient(chainID common.ChainID) (chain.ClientInterface, error) AbstractEthClient(chainID common.ChainID) (chain.BatchCallClient, error)
} }
// Client represents RPC client with custom routing // Client represents RPC client with custom routing
@ -160,7 +160,7 @@ func (c *Client) EthClient(chainID uint64) (*chain.ClientWithFallback, error) {
} }
// AbstractEthClient returns a partial abstraction used by new components for testing purposes // AbstractEthClient returns a partial abstraction used by new components for testing purposes
func (c *Client) AbstractEthClient(chainID common.ChainID) (chain.ClientInterface, error) { func (c *Client) AbstractEthClient(chainID common.ChainID) (chain.BatchCallClient, error) {
client, err := c.getClientUsingCache(uint64(chainID)) client, err := c.getClientUsingCache(uint64(chainID))
if err != nil { if err != nil {
return nil, err return nil, err

58
services/wallet/transfer/commands.go
View File

@ -55,7 +55,7 @@ var (
type ethHistoricalCommand struct { type ethHistoricalCommand struct {
address common.Address address common.Address
chainClient *chain.ClientWithFallback chainClient chain.ClientInterface
balanceCacher balance.Cacher balanceCacher balance.Cacher
feed *event.Feed feed *event.Feed
foundHeaders []*DBHeader foundHeaders []*DBHeader
@ -77,22 +77,22 @@ func (c *ethHistoricalCommand) Command() async.Command {
} }
func (c *ethHistoricalCommand) Run(ctx context.Context) (err error) { func (c *ethHistoricalCommand) Run(ctx context.Context) (err error) {
log.Info("eth historical downloader start", "chainID", c.chainClient.ChainID, "address", c.address, log.Info("eth historical downloader start", "chainID", c.chainClient.NetworkID(), "address", c.address,
"from", c.from.Number, "to", c.to, "noLimit", c.noLimit) "from", c.from.Number, "to", c.to, "noLimit", c.noLimit)
start := time.Now() start := time.Now()
if c.from.Number != nil && c.from.Balance != nil { if c.from.Number != nil && c.from.Balance != nil {
c.balanceCacher.Cache().AddBalance(c.address, c.chainClient.ChainID, c.from.Number, c.from.Balance) c.balanceCacher.Cache().AddBalance(c.address, c.chainClient.NetworkID(), c.from.Number, c.from.Balance)
} }
if c.from.Number != nil && c.from.Nonce != nil { if c.from.Number != nil && c.from.Nonce != nil {
c.balanceCacher.Cache().AddNonce(c.address, c.chainClient.ChainID, c.from.Number, c.from.Nonce) c.balanceCacher.Cache().AddNonce(c.address, c.chainClient.NetworkID(), c.from.Number, c.from.Nonce)
} }
from, headers, startBlock, err := findBlocksWithEthTransfers(ctx, c.chainClient, from, headers, startBlock, err := findBlocksWithEthTransfers(ctx, c.chainClient,
c.balanceCacher, c.address, c.from.Number, c.to, c.noLimit, c.threadLimit) c.balanceCacher, c.address, c.from.Number, c.to, c.noLimit, c.threadLimit)
if err != nil { if err != nil {
c.error = err c.error = err
log.Error("failed to find blocks with transfers", "error", err, "chainID", c.chainClient.ChainID, log.Error("failed to find blocks with transfers", "error", err, "chainID", c.chainClient.NetworkID(),
"address", c.address, "from", c.from.Number, "to", c.to) "address", c.address, "from", c.from.Number, "to", c.to)
return nil return nil
} }
@ -101,7 +101,7 @@ func (c *ethHistoricalCommand) Run(ctx context.Context) (err error) {
c.resultingFrom = from c.resultingFrom = from
c.startBlock = startBlock c.startBlock = startBlock
log.Info("eth historical downloader finished successfully", "chain", c.chainClient.ChainID, log.Info("eth historical downloader finished successfully", "chain", c.chainClient.NetworkID(),
"address", c.address, "from", from, "to", c.to, "total blocks", len(headers), "time", time.Since(start)) "address", c.address, "from", from, "to", c.to, "total blocks", len(headers), "time", time.Since(start))
return nil return nil
@ -110,7 +110,7 @@ func (c *ethHistoricalCommand) Run(ctx context.Context) (err error) {
type erc20HistoricalCommand struct { type erc20HistoricalCommand struct {
erc20 BatchDownloader erc20 BatchDownloader
address common.Address address common.Address
chainClient *chain.ClientWithFallback chainClient chain.ClientInterface
feed *event.Feed feed *event.Feed
iterator *IterativeDownloader iterator *IterativeDownloader
@ -147,14 +147,14 @@ func getErc20BatchSize(chainID uint64) *big.Int {
} }
func (c *erc20HistoricalCommand) Run(ctx context.Context) (err error) { func (c *erc20HistoricalCommand) Run(ctx context.Context) (err error) {
log.Info("wallet historical downloader for erc20 transfers start", "chainID", c.chainClient.ChainID, "address", c.address, log.Info("wallet historical downloader for erc20 transfers start", "chainID", c.chainClient.NetworkID(), "address", c.address,
"from", c.from, "to", c.to) "from", c.from, "to", c.to)
start := time.Now() start := time.Now()
if c.iterator == nil { if c.iterator == nil {
c.iterator, err = SetupIterativeDownloader( c.iterator, err = SetupIterativeDownloader(
c.chainClient, c.address, c.chainClient, c.address,
c.erc20, getErc20BatchSize(c.chainClient.ChainID), c.to, c.from) c.erc20, getErc20BatchSize(c.chainClient.NetworkID()), c.to, c.from)
if err != nil { if err != nil {
log.Error("failed to setup historical downloader for erc20") log.Error("failed to setup historical downloader for erc20")
return err return err
@ -168,7 +168,7 @@ func (c *erc20HistoricalCommand) Run(ctx context.Context) (err error) {
} }
c.foundHeaders = append(c.foundHeaders, headers...) c.foundHeaders = append(c.foundHeaders, headers...)
} }
log.Info("wallet historical downloader for erc20 transfers finished", "chainID", c.chainClient.ChainID, "address", c.address, log.Info("wallet historical downloader for erc20 transfers finished", "chainID", c.chainClient.NetworkID(), "address", c.address,
"from", c.from, "to", c.to, "time", time.Since(start), "headers", len(c.foundHeaders)) "from", c.from, "to", c.to, "time", time.Since(start), "headers", len(c.foundHeaders))
return nil return nil
} }
@ -220,7 +220,7 @@ func (c *controlCommand) Run(parent context.Context) error {
log.Info("current head is", "block number", head.Number) log.Info("current head is", "block number", head.Number)
// Get last known block for each account // Get last known block for each account
lastKnownEthBlocks, accountsWithoutHistory, err := c.blockDAO.GetLastKnownBlockByAddresses(c.chainClient.ChainID, c.accounts) lastKnownEthBlocks, accountsWithoutHistory, err := c.blockDAO.GetLastKnownBlockByAddresses(c.chainClient.NetworkID(), c.accounts)
if err != nil { if err != nil {
log.Error("failed to load last head from database", "error", err) log.Error("failed to load last head from database", "error", err)
if c.NewError(err) { if c.NewError(err) {
@ -301,7 +301,7 @@ func (c *controlCommand) Run(parent context.Context) error {
event := walletevent.Event{ event := walletevent.Event{
Type: EventNewTransfers, Type: EventNewTransfers,
Accounts: []common.Address{address}, Accounts: []common.Address{address},
ChainID: c.chainClient.ChainID, ChainID: c.chainClient.NetworkID(),
} }
for _, header := range cmnd.foundHeaders[address] { for _, header := range cmnd.foundHeaders[address] {
if event.BlockNumber == nil || header.Number.Cmp(event.BlockNumber) == 1 { if event.BlockNumber == nil || header.Number.Cmp(event.BlockNumber) == 1 {
@ -334,9 +334,9 @@ func nonArchivalNodeError(err error) bool {
func (c *controlCommand) NewError(err error) bool { func (c *controlCommand) NewError(err error) bool {
c.errorsCount++ c.errorsCount++
log.Error("controlCommand error", "chainID", c.chainClient.ChainID, "error", err, "counter", c.errorsCount) log.Error("controlCommand error", "chainID", c.chainClient.NetworkID(), "error", err, "counter", c.errorsCount)
if nonArchivalNodeError(err) { if nonArchivalNodeError(err) {
log.Info("Non archival node detected", "chainID", c.chainClient.ChainID) log.Info("Non archival node detected", "chainID", c.chainClient.NetworkID())
c.nonArchivalRPCNode = true c.nonArchivalRPCNode = true
c.feed.Send(walletevent.Event{ c.feed.Send(walletevent.Event{
Type: EventNonArchivalNodeDetected, Type: EventNonArchivalNodeDetected,
@ -387,17 +387,17 @@ func (c *transfersCommand) Run(ctx context.Context) (err error) {
// Take blocks from cache if available and disrespect the limit // Take blocks from cache if available and disrespect the limit
// If no blocks are available in cache, take blocks from DB respecting the limit // If no blocks are available in cache, take blocks from DB respecting the limit
// If no limit is set, take all blocks from DB // If no limit is set, take all blocks from DB
log.Info("start transfersCommand", "chain", c.chainClient.ChainID, "address", c.address, "blockNums", c.blockNums) log.Info("start transfersCommand", "chain", c.chainClient.NetworkID(), "address", c.address, "blockNums", c.blockNums)
startTs := time.Now() startTs := time.Now()
for { for {
blocks := c.blockNums blocks := c.blockNums
if blocks == nil { if blocks == nil {
blocks, _ = c.blockDAO.GetBlocksToLoadByAddress(c.chainClient.ChainID, c.address, numberOfBlocksCheckedPerIteration) blocks, _ = c.blockDAO.GetBlocksToLoadByAddress(c.chainClient.NetworkID(), c.address, numberOfBlocksCheckedPerIteration)
} }
for _, blockNum := range blocks { for _, blockNum := range blocks {
log.Debug("transfersCommand block start", "chain", c.chainClient.ChainID, "address", c.address, "block", blockNum) log.Debug("transfersCommand block start", "chain", c.chainClient.NetworkID(), "address", c.address, "block", blockNum)
allTransfers, err := c.eth.GetTransfersByNumber(ctx, blockNum) allTransfers, err := c.eth.GetTransfersByNumber(ctx, blockNum)
if err != nil { if err != nil {
@ -421,9 +421,9 @@ func (c *transfersCommand) Run(ctx context.Context) (err error) {
} }
} else { } else {
// If no transfers found, that is suspecting, because downloader returned this block as containing transfers // If no transfers found, that is suspecting, because downloader returned this block as containing transfers
log.Error("no transfers found in block", "chain", c.chainClient.ChainID, "address", c.address, "block", blockNum) log.Error("no transfers found in block", "chain", c.chainClient.NetworkID(), "address", c.address, "block", blockNum)
err = markBlocksAsLoaded(c.chainClient.ChainID, c.db.client, c.address, []*big.Int{blockNum}) err = markBlocksAsLoaded(c.chainClient.NetworkID(), c.db.client, c.address, []*big.Int{blockNum})
if err != nil { if err != nil {
log.Error("Mark blocks loaded error", "error", err) log.Error("Mark blocks loaded error", "error", err)
return err return err
@ -434,19 +434,19 @@ func (c *transfersCommand) Run(ctx context.Context) (err error) {
c.notifyOfNewTransfers(allTransfers) c.notifyOfNewTransfers(allTransfers)
log.Debug("transfersCommand block end", "chain", c.chainClient.ChainID, "address", c.address, log.Debug("transfersCommand block end", "chain", c.chainClient.NetworkID(), "address", c.address,
"block", blockNum, "tranfers.len", len(allTransfers), "fetchedTransfers.len", len(c.fetchedTransfers)) "block", blockNum, "tranfers.len", len(allTransfers), "fetchedTransfers.len", len(c.fetchedTransfers))
} }
if c.blockNums != nil || len(blocks) == 0 || if c.blockNums != nil || len(blocks) == 0 ||
(c.blocksLimit > noBlockLimit && len(blocks) >= c.blocksLimit) { (c.blocksLimit > noBlockLimit && len(blocks) >= c.blocksLimit) {
log.Debug("loadTransfers breaking loop on block limits reached or 0 blocks", "chain", c.chainClient.ChainID, log.Debug("loadTransfers breaking loop on block limits reached or 0 blocks", "chain", c.chainClient.NetworkID(),
"address", c.address, "limit", c.blocksLimit, "blocks", len(blocks)) "address", c.address, "limit", c.blocksLimit, "blocks", len(blocks))
break break
} }
} }
log.Info("end transfersCommand", "chain", c.chainClient.ChainID, "address", c.address, log.Info("end transfersCommand", "chain", c.chainClient.NetworkID(), "address", c.address,
"blocks.len", len(c.blockNums), "transfers.len", len(c.fetchedTransfers), "in", time.Since(startTs)) "blocks.len", len(c.blockNums), "transfers.len", len(c.fetchedTransfers), "in", time.Since(startTs))
return nil return nil
@ -516,7 +516,7 @@ func (c *transfersCommand) saveAndConfirmPending(allTransfers []Transfer, blockN
} }
} }
resErr = saveTransfersMarkBlocksLoaded(tx, c.chainClient.ChainID, c.address, allTransfers, []*big.Int{blockNum}) resErr = saveTransfersMarkBlocksLoaded(tx, c.chainClient.NetworkID(), c.address, allTransfers, []*big.Int{blockNum})
if resErr != nil { if resErr != nil {
log.Error("SaveTransfers error", "error", resErr) log.Error("SaveTransfers error", "error", resErr)
} }
@ -615,7 +615,7 @@ func (c *transfersCommand) notifyOfNewTransfers(transfers []Transfer) {
c.feed.Send(walletevent.Event{ c.feed.Send(walletevent.Event{
Type: EventNewTransfers, Type: EventNewTransfers,
Accounts: []common.Address{c.address}, Accounts: []common.Address{c.address},
ChainID: c.chainClient.ChainID, ChainID: c.chainClient.NetworkID(),
}) })
} }
} }
@ -717,16 +717,16 @@ func (c *findAndCheckBlockRangeCommand) Run(parent context.Context) error {
lastBlockNumber := c.toByAddress[address] lastBlockNumber := c.toByAddress[address]
log.Debug("saving headers", "len", len(uniqHeaders), "lastBlockNumber", lastBlockNumber, log.Debug("saving headers", "len", len(uniqHeaders), "lastBlockNumber", lastBlockNumber,
"balance", c.balanceCacher.Cache().GetBalance(address, c.chainClient.ChainID, lastBlockNumber), "balance", c.balanceCacher.Cache().GetBalance(address, c.chainClient.NetworkID(), lastBlockNumber),
"nonce", c.balanceCacher.Cache().GetNonce(address, c.chainClient.ChainID, lastBlockNumber)) "nonce", c.balanceCacher.Cache().GetNonce(address, c.chainClient.NetworkID(), lastBlockNumber))
to := &Block{ to := &Block{
Number: lastBlockNumber, Number: lastBlockNumber,
Balance: c.balanceCacher.Cache().GetBalance(address, c.chainClient.ChainID, lastBlockNumber), Balance: c.balanceCacher.Cache().GetBalance(address, c.chainClient.NetworkID(), lastBlockNumber),
Nonce: c.balanceCacher.Cache().GetNonce(address, c.chainClient.ChainID, lastBlockNumber), Nonce: c.balanceCacher.Cache().GetNonce(address, c.chainClient.NetworkID(), lastBlockNumber),
} }
log.Debug("uniqHeaders found for account", "address", address, "uniqHeaders.len", len(uniqHeaders)) log.Debug("uniqHeaders found for account", "address", address, "uniqHeaders.len", len(uniqHeaders))
err = c.db.ProcessBlocks(c.chainClient.ChainID, address, newFromByAddress[address], to, uniqHeaders) err = c.db.ProcessBlocks(c.chainClient.NetworkID(), address, newFromByAddress[address], to, uniqHeaders)
if err != nil { if err != nil {
return err return err
} }

66
services/wallet/transfer/commands_sequential.go
View File

@ -29,7 +29,7 @@ func (c *findNewBlocksCommand) Command() async.Command {
} }
func (c *findNewBlocksCommand) Run(parent context.Context) (err error) { func (c *findNewBlocksCommand) Run(parent context.Context) (err error) {
log.Debug("start findNewBlocksCommand", "account", c.account, "chain", c.chainClient.ChainID, "noLimit", c.noLimit) log.Debug("start findNewBlocksCommand", "account", c.account, "chain", c.chainClient.NetworkID(), "noLimit", c.noLimit)
headNum, err := getHeadBlockNumber(parent, c.chainClient) headNum, err := getHeadBlockNumber(parent, c.chainClient)
if err != nil { if err != nil {
@ -37,7 +37,7 @@ func (c *findNewBlocksCommand) Run(parent context.Context) (err error) {
return err // Might need to retry a couple of times return err // Might need to retry a couple of times
} }
blockRange, err := loadBlockRangeInfo(c.chainClient.ChainID, c.account, c.blockRangeDAO) blockRange, err := loadBlockRangeInfo(c.chainClient.NetworkID(), c.account, c.blockRangeDAO)
if err != nil { if err != nil {
log.Error("findBlocksCommand loadBlockRangeInfo", "error", err) log.Error("findBlocksCommand loadBlockRangeInfo", "error", err)
// c.error = err // c.error = err
@ -47,7 +47,7 @@ func (c *findNewBlocksCommand) Run(parent context.Context) (err error) {
if blockRange != nil { if blockRange != nil {
c.fromBlockNumber = blockRange.LastKnown c.fromBlockNumber = blockRange.LastKnown
log.Debug("Launching new blocks command", "chainID", c.chainClient.ChainID, "account", c.account, log.Debug("Launching new blocks command", "chainID", c.chainClient.NetworkID(), "account", c.account,
"from", c.fromBlockNumber, "headNum", headNum) "from", c.fromBlockNumber, "headNum", headNum)
// In case interval between checks is set smaller than block mining time, // In case interval between checks is set smaller than block mining time,
@ -69,7 +69,7 @@ type findBlocksCommand struct {
account common.Address account common.Address
db *Database db *Database
blockRangeDAO *BlockRangeSequentialDAO blockRangeDAO *BlockRangeSequentialDAO
chainClient *chain.ClientWithFallback chainClient chain.ClientInterface
balanceCacher balance.Cacher balanceCacher balance.Cacher
feed *event.Feed feed *event.Feed
noLimit bool noLimit bool
@ -92,7 +92,7 @@ func (c *findBlocksCommand) Command() async.Command {
} }
func (c *findBlocksCommand) Run(parent context.Context) (err error) { func (c *findBlocksCommand) Run(parent context.Context) (err error) {
log.Debug("start findBlocksCommand", "account", c.account, "chain", c.chainClient.ChainID, "noLimit", c.noLimit) log.Debug("start findBlocksCommand", "account", c.account, "chain", c.chainClient.NetworkID(), "noLimit", c.noLimit)
rangeSize := big.NewInt(DefaultNodeBlockChunkSize) rangeSize := big.NewInt(DefaultNodeBlockChunkSize)
@ -107,16 +107,16 @@ func (c *findBlocksCommand) Run(parent context.Context) (err error) {
headers, _ := c.checkRange(parent, from, to) headers, _ := c.checkRange(parent, from, to)
if c.error != nil { if c.error != nil {
log.Error("findBlocksCommand checkRange", "error", c.error, "account", c.account, log.Error("findBlocksCommand checkRange", "error", c.error, "account", c.account,
"chain", c.chainClient.ChainID, "from", from, "to", to) "chain", c.chainClient.NetworkID(), "from", from, "to", to)
break break
} }
if len(headers) > 0 { if len(headers) > 0 {
log.Debug("findBlocksCommand saving headers", "len", len(headers), "lastBlockNumber", to, log.Debug("findBlocksCommand saving headers", "len", len(headers), "lastBlockNumber", to,
"balance", c.balanceCacher.Cache().GetBalance(c.account, c.chainClient.ChainID, to), "balance", c.balanceCacher.Cache().GetBalance(c.account, c.chainClient.NetworkID(), to),
"nonce", c.balanceCacher.Cache().GetNonce(c.account, c.chainClient.ChainID, to)) "nonce", c.balanceCacher.Cache().GetNonce(c.account, c.chainClient.NetworkID(), to))
err = c.db.SaveBlocks(c.chainClient.ChainID, c.account, headers) err = c.db.SaveBlocks(c.chainClient.NetworkID(), c.account, headers)
if err != nil { if err != nil {
c.error = err c.error = err
// return err // return err
@ -140,7 +140,7 @@ func (c *findBlocksCommand) Run(parent context.Context) (err error) {
} }
} }
log.Debug("end findBlocksCommand", "account", c.account, "chain", c.chainClient.ChainID, "noLimit", c.noLimit) log.Debug("end findBlocksCommand", "account", c.account, "chain", c.chainClient.NetworkID(), "noLimit", c.noLimit)
return nil return nil
} }
@ -151,9 +151,9 @@ func (c *findBlocksCommand) blocksFound(headers []*DBHeader) {
func (c *findBlocksCommand) upsertBlockRange(blockRange *BlockRange) error { func (c *findBlocksCommand) upsertBlockRange(blockRange *BlockRange) error {
log.Debug("upsert block range", "Start", blockRange.Start, "FirstKnown", blockRange.FirstKnown, "LastKnown", blockRange.LastKnown, log.Debug("upsert block range", "Start", blockRange.Start, "FirstKnown", blockRange.FirstKnown, "LastKnown", blockRange.LastKnown,
"chain", c.chainClient.ChainID, "account", c.account) "chain", c.chainClient.NetworkID(), "account", c.account)
err := c.blockRangeDAO.upsertRange(c.chainClient.ChainID, c.account, blockRange) err := c.blockRangeDAO.upsertRange(c.chainClient.NetworkID(), c.account, blockRange)
if err != nil { if err != nil {
c.error = err c.error = err
log.Error("findBlocksCommand upsertRange", "error", err) log.Error("findBlocksCommand upsertRange", "error", err)
@ -171,19 +171,19 @@ func (c *findBlocksCommand) checkRange(parent context.Context, from *big.Int, to
newFromBlock, ethHeaders, startBlock, err := c.fastIndex(parent, c.balanceCacher, fromBlock, to) newFromBlock, ethHeaders, startBlock, err := c.fastIndex(parent, c.balanceCacher, fromBlock, to)
if err != nil { if err != nil {
log.Error("findBlocksCommand checkRange fastIndex", "err", err, "account", c.account, log.Error("findBlocksCommand checkRange fastIndex", "err", err, "account", c.account,
"chain", c.chainClient.ChainID) "chain", c.chainClient.NetworkID())
c.error = err c.error = err
// return err // In case c.noLimit is true, hystrix "max concurrency" may be reached and we will not be able to index ETH transfers // return err // In case c.noLimit is true, hystrix "max concurrency" may be reached and we will not be able to index ETH transfers
return nil, nil return nil, nil
} }
log.Debug("findBlocksCommand checkRange", "chainID", c.chainClient.ChainID, "account", c.account, log.Debug("findBlocksCommand checkRange", "chainID", c.chainClient.NetworkID(), "account", c.account,
"startBlock", startBlock, "newFromBlock", newFromBlock.Number, "toBlockNumber", to, "noLimit", c.noLimit) "startBlock", startBlock, "newFromBlock", newFromBlock.Number, "toBlockNumber", to, "noLimit", c.noLimit)
// There could be incoming ERC20 transfers which don't change the balance // There could be incoming ERC20 transfers which don't change the balance
// and nonce of ETH account, so we keep looking for them // and nonce of ETH account, so we keep looking for them
erc20Headers, err := c.fastIndexErc20(parent, newFromBlock.Number, to) erc20Headers, err := c.fastIndexErc20(parent, newFromBlock.Number, to)
if err != nil { if err != nil {
log.Error("findBlocksCommand checkRange fastIndexErc20", "err", err, "account", c.account, "chain", c.chainClient.ChainID) log.Error("findBlocksCommand checkRange fastIndexErc20", "err", err, "account", c.account, "chain", c.chainClient.NetworkID())
c.error = err c.error = err
// return err // return err
return nil, nil return nil, nil
@ -198,7 +198,7 @@ func (c *findBlocksCommand) checkRange(parent context.Context, from *big.Int, to
c.resFromBlock = newFromBlock c.resFromBlock = newFromBlock
c.startBlockNumber = startBlock c.startBlockNumber = startBlock
log.Debug("end findBlocksCommand checkRange", "chainID", c.chainClient.ChainID, "account", c.account, log.Debug("end findBlocksCommand checkRange", "chainID", c.chainClient.NetworkID(), "account", c.account,
"c.startBlock", c.startBlockNumber, "newFromBlock", newFromBlock.Number, "c.startBlock", c.startBlockNumber, "newFromBlock", newFromBlock.Number,
"toBlockNumber", to, "c.resFromBlock", c.resFromBlock.Number) "toBlockNumber", to, "c.resFromBlock", c.resFromBlock.Number)
@ -227,7 +227,6 @@ func areAllHistoryBlocksLoaded(blockInfo *BlockRange) bool {
if blockInfo.FirstKnown != nil && blockInfo.Start != nil && if blockInfo.FirstKnown != nil && blockInfo.Start != nil &&
blockInfo.Start.Cmp(blockInfo.FirstKnown) >= 0 { blockInfo.Start.Cmp(blockInfo.FirstKnown) >= 0 {
return true return true
} }
@ -252,7 +251,7 @@ func (c *findBlocksCommand) fastIndex(ctx context.Context, bCacher balance.Cache
fromBlock *Block, toBlockNumber *big.Int) (resultingFrom *Block, headers []*DBHeader, fromBlock *Block, toBlockNumber *big.Int) (resultingFrom *Block, headers []*DBHeader,
startBlock *big.Int, err error) { startBlock *big.Int, err error) {
log.Debug("fast index started", "chainID", c.chainClient.ChainID, "account", c.account, log.Debug("fast index started", "chainID", c.chainClient.NetworkID(), "account", c.account,
"from", fromBlock.Number, "to", toBlockNumber) "from", fromBlock.Number, "to", toBlockNumber)
start := time.Now() start := time.Now()
@ -283,7 +282,7 @@ func (c *findBlocksCommand) fastIndex(ctx context.Context, bCacher balance.Cache
resultingFrom = &Block{Number: command.resultingFrom} resultingFrom = &Block{Number: command.resultingFrom}
headers = command.foundHeaders headers = command.foundHeaders
startBlock = command.startBlock startBlock = command.startBlock
log.Debug("fast indexer finished", "chainID", c.chainClient.ChainID, "account", c.account, "in", time.Since(start), log.Debug("fast indexer finished", "chainID", c.chainClient.NetworkID(), "account", c.account, "in", time.Since(start),
"startBlock", command.startBlock, "resultingFrom", resultingFrom.Number, "headers", len(headers)) "startBlock", command.startBlock, "resultingFrom", resultingFrom.Number, "headers", len(headers))
return return
} }
@ -313,7 +312,7 @@ func (c *findBlocksCommand) fastIndexErc20(ctx context.Context, fromBlockNumber
return nil, ctx.Err() return nil, ctx.Err()
case <-group.WaitAsync(): case <-group.WaitAsync():
headers := erc20.foundHeaders headers := erc20.foundHeaders
log.Debug("fast indexer Erc20 finished", "chainID", c.chainClient.ChainID, "account", c.account, log.Debug("fast indexer Erc20 finished", "chainID", c.chainClient.NetworkID(), "account", c.account,
"in", time.Since(start), "headers", len(headers)) "in", time.Since(start), "headers", len(headers))
return headers, nil return headers, nil
} }
@ -388,7 +387,7 @@ type loadBlocksAndTransfersCommand struct {
} }
func (c *loadBlocksAndTransfersCommand) Run(parent context.Context) error { func (c *loadBlocksAndTransfersCommand) Run(parent context.Context) error {
log.Debug("start load all transfers command", "chain", c.chainClient.ChainID, "account", c.account) log.Debug("start load all transfers command", "chain", c.chainClient.NetworkID(), "account", c.account)
ctx := parent ctx := parent
group := async.NewGroup(ctx) group := async.NewGroup(ctx)
@ -414,7 +413,7 @@ func (c *loadBlocksAndTransfersCommand) Run(parent context.Context) error {
c.balanceCacher.Clear() c.balanceCacher.Clear()
return ctx.Err() return ctx.Err()
case <-group.WaitAsync(): case <-group.WaitAsync():
log.Debug("end loadBlocksAndTransfers command", "chain", c.chainClient.ChainID, "account", c.account) log.Debug("end loadBlocksAndTransfers command", "chain", c.chainClient.NetworkID(), "account", c.account)
return nil return nil
} }
} }
@ -433,7 +432,7 @@ func (c *loadBlocksAndTransfersCommand) startTransfersLoop(ctx context.Context)
func (c *loadBlocksAndTransfersCommand) fetchHistoryBlocks(ctx context.Context, group *async.Group, blocksLoadedCh chan []*DBHeader) error { func (c *loadBlocksAndTransfersCommand) fetchHistoryBlocks(ctx context.Context, group *async.Group, blocksLoadedCh chan []*DBHeader) error {
log.Debug("fetchHistoryBlocks start", "chainID", c.chainClient.ChainID, "account", c.account) log.Debug("fetchHistoryBlocks start", "chainID", c.chainClient.NetworkID(), "account", c.account)
headNum, err := getHeadBlockNumber(ctx, c.chainClient) headNum, err := getHeadBlockNumber(ctx, c.chainClient)
if err != nil { if err != nil {
@ -441,17 +440,18 @@ func (c *loadBlocksAndTransfersCommand) fetchHistoryBlocks(ctx context.Context,
return err // Might need to retry a couple of times return err // Might need to retry a couple of times
} }
blockRange, err := loadBlockRangeInfo(c.chainClient.ChainID, c.account, c.blockRangeDAO) blockRange, err := loadBlockRangeInfo(c.chainClient.NetworkID(), c.account, c.blockRangeDAO)
if err != nil { if err != nil {
log.Error("findBlocksCommand loadBlockRangeInfo", "error", err) log.Error("findBlocksCommand loadBlockRangeInfo", "error", err)
// c.error = err // c.error = err
return err // Will keep spinning forever nomatter what return err // Will keep spinning forever nomatter what
} }
/// first
allHistoryLoaded := areAllHistoryBlocksLoaded(blockRange) allHistoryLoaded := areAllHistoryBlocksLoaded(blockRange)
to := getToHistoryBlockNumber(headNum, blockRange, allHistoryLoaded) to := getToHistoryBlockNumber(headNum, blockRange, allHistoryLoaded)
log.Debug("fetchHistoryBlocks", "chainID", c.chainClient.ChainID, "account", c.account, "to", to, "allHistoryLoaded", allHistoryLoaded) log.Debug("fetchHistoryBlocks", "chainID", c.chainClient.NetworkID(), "account", c.account, "to", to, "allHistoryLoaded", allHistoryLoaded)
if !allHistoryLoaded { if !allHistoryLoaded {
fbc := &findBlocksCommand{ fbc := &findBlocksCommand{
@ -482,14 +482,14 @@ func (c *loadBlocksAndTransfersCommand) fetchHistoryBlocks(ctx context.Context,
} }
} }
log.Debug("fetchHistoryBlocks end", "chainID", c.chainClient.ChainID, "account", c.account) log.Debug("fetchHistoryBlocks end", "chainID", c.chainClient.NetworkID(), "account", c.account)
return nil return nil
} }
func (c *loadBlocksAndTransfersCommand) startFetchingNewBlocks(group *async.Group, address common.Address, blocksLoadedCh chan<- []*DBHeader) { func (c *loadBlocksAndTransfersCommand) startFetchingNewBlocks(group *async.Group, address common.Address, blocksLoadedCh chan<- []*DBHeader) {
log.Debug("startFetchingNewBlocks", "chainID", c.chainClient.ChainID, "account", address) log.Debug("startFetchingNewBlocks", "chainID", c.chainClient.NetworkID(), "account", address)
newBlocksCmd := &findNewBlocksCommand{ newBlocksCmd := &findNewBlocksCommand{
findBlocksCommand: &findBlocksCommand{ findBlocksCommand: &findBlocksCommand{
@ -509,9 +509,9 @@ func (c *loadBlocksAndTransfersCommand) startFetchingNewBlocks(group *async.Grou
func (c *loadBlocksAndTransfersCommand) fetchTransfersForLoadedBlocks(group *async.Group) error { func (c *loadBlocksAndTransfersCommand) fetchTransfersForLoadedBlocks(group *async.Group) error {
log.Debug("fetchTransfers start", "chainID", c.chainClient.ChainID, "account", c.account) log.Debug("fetchTransfers start", "chainID", c.chainClient.NetworkID(), "account", c.account)
blocks, err := c.blockDAO.GetBlocksToLoadByAddress(c.chainClient.ChainID, c.account, numberOfBlocksCheckedPerIteration) blocks, err := c.blockDAO.GetBlocksToLoadByAddress(c.chainClient.NetworkID(), c.account, numberOfBlocksCheckedPerIteration)
if err != nil { if err != nil {
log.Error("loadBlocksAndTransfersCommand GetBlocksToLoadByAddress", "error", err) log.Error("loadBlocksAndTransfersCommand GetBlocksToLoadByAddress", "error", err)
return err return err
@ -542,20 +542,20 @@ func (c *loadBlocksAndTransfersCommand) notifyHistoryReady() {
c.feed.Send(walletevent.Event{ c.feed.Send(walletevent.Event{
Type: EventRecentHistoryReady, Type: EventRecentHistoryReady,
Accounts: []common.Address{c.account}, Accounts: []common.Address{c.account},
ChainID: c.chainClient.ChainID, ChainID: c.chainClient.NetworkID(),
}) })
} }
} }
func (c *loadBlocksAndTransfersCommand) areAllTransfersLoaded() (bool, error) { func (c *loadBlocksAndTransfersCommand) areAllTransfersLoaded() (bool, error) {
allBlocksLoaded, err := areAllHistoryBlocksLoadedForAddress(c.blockRangeDAO, c.chainClient.ChainID, c.account) allBlocksLoaded, err := areAllHistoryBlocksLoadedForAddress(c.blockRangeDAO, c.chainClient.NetworkID(), c.account)
if err != nil { if err != nil {
log.Error("loadBlockAndTransfersCommand allHistoryBlocksLoaded", "error", err) log.Error("loadBlockAndTransfersCommand allHistoryBlocksLoaded", "error", err)
return false, err return false, err
} }
if allBlocksLoaded { if allBlocksLoaded {
firstHeader, err := c.blockDAO.GetFirstSavedBlock(c.chainClient.ChainID, c.account) firstHeader, err := c.blockDAO.GetFirstSavedBlock(c.chainClient.NetworkID(), c.account)
if err != nil { if err != nil {
log.Error("loadBlocksAndTransfersCommand GetFirstSavedBlock", "error", err) log.Error("loadBlocksAndTransfersCommand GetFirstSavedBlock", "error", err)
return false, err return false, err
@ -572,7 +572,7 @@ func (c *loadBlocksAndTransfersCommand) areAllTransfersLoaded() (bool, error) {
// TODO - make it a common method for every service that wants head block number, that will cache the latest block // TODO - make it a common method for every service that wants head block number, that will cache the latest block
// and updates it on timeout // and updates it on timeout
func getHeadBlockNumber(parent context.Context, chainClient *chain.ClientWithFallback) (*big.Int, error) { func getHeadBlockNumber(parent context.Context, chainClient chain.ClientInterface) (*big.Int, error) {
ctx, cancel := context.WithTimeout(parent, 3*time.Second) ctx, cancel := context.WithTimeout(parent, 3*time.Second)
head, err := chainClient.HeaderByNumber(ctx, nil) head, err := chainClient.HeaderByNumber(ctx, nil)
cancel() cancel()

202
services/wallet/transfer/commands_sequential_test.go Normal file
View File

@ -0,0 +1,202 @@
package transfer
import (
"context"
"math/big"
"testing"
"github.com/stretchr/testify/require"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/rpc"
"github.com/status-im/status-go/rpc/chain"
"github.com/status-im/status-go/services/wallet/async"
"github.com/status-im/status-go/services/wallet/balance"
"github.com/status-im/status-go/t/helpers"
"github.com/status-im/status-go/walletdatabase"
)
type TestClient struct {
t *testing.T
// [][block, newBalance, nonceDiff]
balances [][]int
balanceHistory map[uint64]*big.Int
nonceHistory map[uint64]uint64
}
func (tc TestClient) BatchCallContext(ctx context.Context, b []rpc.BatchElem) error {
tc.t.Log("BatchCallContext")
return nil
}
func (tc TestClient) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
tc.t.Log("HeaderByHash")
return nil, nil
}
func (tc TestClient) BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error) {
tc.t.Log("BlockByHash")
return nil, nil
}
func (tc TestClient) BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error) {
tc.t.Log("BlockByNumber")
return nil, nil
}
func (tc TestClient) NonceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (uint64, error) {
nonce := tc.nonceHistory[blockNumber.Uint64()]
tc.t.Log("NonceAt", blockNumber, "result:", nonce)
return nonce, nil
}
func (tc TestClient) FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error) {
tc.t.Log("FilterLogs")
return nil, nil
}
func (tc *TestClient) prepareBalanceHistory(toBlock int) {
var currentBlock, currentBalance, currentNonce int
tc.balanceHistory = map[uint64]*big.Int{}
tc.nonceHistory = map[uint64]uint64{}
if len(tc.balances) == 0 {
tc.balances = append(tc.balances, []int{toBlock + 1, 0, 0})
} else {
lastBlock := tc.balances[len(tc.balances)-1]
tc.balances = append(tc.balances, []int{toBlock + 1, lastBlock[1], 0})
}
for _, change := range tc.balances {
for blockN := currentBlock; blockN < change[0]; blockN++ {
tc.balanceHistory[uint64(blockN)] = big.NewInt(int64(currentBalance))
tc.nonceHistory[uint64(blockN)] = uint64(currentNonce)
}
currentBlock = change[0]
currentBalance = change[1]
currentNonce += change[2]
}
tc.t.Log("=========================================")
tc.t.Log(tc.balanceHistory)
tc.t.Log(tc.nonceHistory)
tc.t.Log("=========================================")
}
func (tc TestClient) BalanceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (*big.Int, error) {
balance := tc.balanceHistory[blockNumber.Uint64()]
tc.t.Log("BalanceAt", blockNumber, "result:", balance)
return balance, nil
}
func (tc *TestClient) HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) {
tc.t.Log("HeaderByNumber", number)
header := &types.Header{
Number: number,
Time: 0,
}
return header, nil
}
func (tc TestClient) FullTransactionByBlockNumberAndIndex(ctx context.Context, blockNumber *big.Int, index uint) (*chain.FullTransaction, error) {
tc.t.Log("FullTransactionByBlockNumberAndIndex")
blockHash := common.BigToHash(blockNumber)
tx := &chain.FullTransaction{
Tx: &types.Transaction{},
TxExtraInfo: chain.TxExtraInfo{
BlockNumber: (*hexutil.Big)(big.NewInt(0)),
BlockHash: &blockHash,
},
}
return tx, nil
}
func (tc TestClient) GetBaseFeeFromBlock(blockNumber *big.Int) (string, error) {
tc.t.Log("GetBaseFeeFromBloc")
return "", nil
}
func (tc TestClient) NetworkID() uint64 {
return 1
}
func (tc TestClient) ToBigInt() *big.Int {
tc.t.Log("ToBigInt")
return nil
}
type findBlockCase struct {
balanceChanges [][]int
fromBlock int64
toBlock int64
expectedBlocksFound int
}
var findBlocksCommandCases = []findBlockCase{
{
balanceChanges: [][]int{
{5, 1, 0},
{20, 2, 0},
{45, 1, 1},
{46, 50, 0},
{75, 0, 1},
},
toBlock: 100,
expectedBlocksFound: 5,
},
{
balanceChanges: [][]int{},
toBlock: 100,
expectedBlocksFound: 0,
},
}
func TestFindBlocksCommand(t *testing.T) {
for _, testCase := range findBlocksCommandCases {
ctx := context.Background()
group := async.NewGroup(ctx)
db, err := helpers.SetupTestMemorySQLDB(walletdatabase.DbInitializer{})
require.NoError(t, err)
tm := &TransactionManager{db, nil, nil, nil, nil, nil, nil}
wdb := NewDB(db)
tc := &TestClient{
t: t,
balances: testCase.balanceChanges,
}
tc.prepareBalanceHistory(100)
blockChannel := make(chan []*DBHeader, 100)
fbc := &findBlocksCommand{
account: common.HexToAddress("0x1234"),
db: wdb,
blockRangeDAO: &BlockRangeSequentialDAO{wdb.client},
chainClient: tc,
balanceCacher: balance.NewCache(),
feed: &event.Feed{},
noLimit: false,
fromBlockNumber: big.NewInt(testCase.fromBlock),
toBlockNumber: big.NewInt(testCase.toBlock),
transactionManager: tm,
blocksLoadedCh: blockChannel,
}
group.Add(fbc.Command())
select {
case <-ctx.Done():
t.Log("ERROR")
case <-group.WaitAsync():
close(blockChannel)
require.Equal(t, testCase.expectedBlocksFound, len(<-blockChannel))
}
}
}

12
services/wallet/transfer/downloader.go
View File

@ -236,7 +236,7 @@ func (d *ETHDownloader) getTransfersInBlock(ctx context.Context, blk *types.Bloc
} }
// NewERC20TransfersDownloader returns new instance. // NewERC20TransfersDownloader returns new instance.
func NewERC20TransfersDownloader(client *chain.ClientWithFallback, accounts []common.Address, signer types.Signer) *ERC20TransfersDownloader { func NewERC20TransfersDownloader(client chain.ClientInterface, accounts []common.Address, signer types.Signer) *ERC20TransfersDownloader {
signature := w_common.GetEventSignatureHash(w_common.Erc20_721TransferEventSignature) signature := w_common.GetEventSignatureHash(w_common.Erc20_721TransferEventSignature)
return &ERC20TransfersDownloader{ return &ERC20TransfersDownloader{
@ -253,7 +253,7 @@ func NewERC20TransfersDownloader(client *chain.ClientWithFallback, accounts []co
// database gets implemented, differentiation between erc20 and erc721 will handled // database gets implemented, differentiation between erc20 and erc721 will handled
// in the controller. // in the controller.
type ERC20TransfersDownloader struct { type ERC20TransfersDownloader struct {
client *chain.ClientWithFallback client chain.ClientInterface
accounts []common.Address accounts []common.Address
// hash of the Transfer event signature // hash of the Transfer event signature
@ -414,7 +414,7 @@ func (d *ERC20TransfersDownloader) blocksFromLogs(parent context.Context, logs [
// time to get logs for 100000 blocks = 1.144686979s. with 249 events in the result set. // time to get logs for 100000 blocks = 1.144686979s. with 249 events in the result set.
func (d *ERC20TransfersDownloader) GetHeadersInRange(parent context.Context, from, to *big.Int) ([]*DBHeader, error) { func (d *ERC20TransfersDownloader) GetHeadersInRange(parent context.Context, from, to *big.Int) ([]*DBHeader, error) {
start := time.Now() start := time.Now()
log.Debug("get erc20 transfers in range start", "chainID", d.client.ChainID, "from", from, "to", to) log.Debug("get erc20 transfers in range start", "chainID", d.client.NetworkID(), "from", from, "to", to)
headers := []*DBHeader{} headers := []*DBHeader{}
ctx := context.Background() ctx := context.Background()
for _, address := range d.accounts { for _, address := range d.accounts {
@ -444,15 +444,15 @@ func (d *ERC20TransfersDownloader) GetHeadersInRange(parent context.Context, fro
return nil, err return nil, err
} }
if len(rst) == 0 { if len(rst) == 0 {
log.Warn("no headers found in logs for account", "chainID", d.client.ChainID, "address", address, "from", from, "to", to) log.Warn("no headers found in logs for account", "chainID", d.client.NetworkID(), "address", address, "from", from, "to", to)
continue continue
} else { } else {
headers = append(headers, rst...) headers = append(headers, rst...)
log.Debug("found erc20 transfers for account", "chainID", d.client.ChainID, "address", address, log.Debug("found erc20 transfers for account", "chainID", d.client.NetworkID(), "address", address,
"from", from, "to", to, "headers", len(headers)) "from", from, "to", to, "headers", len(headers))
} }
} }
log.Debug("get erc20 transfers in range end", "chainID", d.client.ChainID, log.Debug("get erc20 transfers in range end", "chainID", d.client.NetworkID(),
"from", from, "to", to, "headers", len(headers), "took", time.Since(start)) "from", from, "to", to, "headers", len(headers), "took", time.Since(start))
return headers, nil return headers, nil
} }

2
transactions/pendingtxtracker_test.go
View File

@ -55,7 +55,7 @@ func (m *MockChainClient) setAvailableClients(chainIDs []common.ChainID) *MockCh
return m return m
} }
func (m *MockChainClient) AbstractEthClient(chainID common.ChainID) (chain.ClientInterface, error) { func (m *MockChainClient) AbstractEthClient(chainID common.ChainID) (chain.BatchCallClient, error) {
if _, ok := m.clients[chainID]; !ok { if _, ok := m.clients[chainID]; !ok {
panic(fmt.Sprintf("no mock client for chainID %d", chainID)) panic(fmt.Sprintf("no mock client for chainID %d", chainID))
} }