fix(desktop/wallet): fix bug in balance_cache - balances and nonces (#3509)

were stored in cache by pointers, which caused falsy cache hits in loop
because pointers with same address were created for different block
numbers. Now cache uses block numbers of uint64 as key, which can
overflow but it is not a problem since we use this cache for values
comparison, not as user data.
Fix crash on nil pointer in log.
Remove some unused code.
Protect nonceRanges with mutex while reading.

Updates #10246

services/wallet/transfer/balance_cache.go

@@ -17,8 +17,8 @@ type nonceRange struct {
 
 type balanceCache struct {
 	// balances maps an address to a map of a block number and the balance of this particular address
-	balances     map[common.Address]map[*big.Int]*big.Int
+	balances     map[common.Address]map[uint64]*big.Int // we don't care about block number overflow as we use cache only for comparing balances when fetching, not for UI
-	nonces       map[common.Address]map[*big.Int]*int64
+	nonces       map[common.Address]map[uint64]*int64   // we don't care about block number overflow as we use cache only for comparing balances when fetching, not for UI
 	nonceRanges  map[common.Address]map[int64]nonceRange
 	sortedRanges map[common.Address][]nonceRange
 	rw           sync.RWMutex
@@ -33,7 +33,7 @@ func (b *balanceCache) ReadCachedBalance(account common.Address, blockNumber *bi
 	b.rw.RLock()
 	defer b.rw.RUnlock()
 
-	return b.balances[account][blockNumber]
+	return b.balances[account][blockNumber.Uint64()]
 }
 
 func (b *balanceCache) addBalanceToCache(account common.Address, blockNumber *big.Int, balance *big.Int) {
@@ -42,9 +42,9 @@ func (b *balanceCache) addBalanceToCache(account common.Address, blockNumber *bi
 
 	_, exists := b.balances[account]
 	if !exists {
-		b.balances[account] = make(map[*big.Int]*big.Int)
+		b.balances[account] = make(map[uint64]*big.Int)
 	}
-	b.balances[account][blockNumber] = balance
+	b.balances[account][blockNumber.Uint64()] = balance
 }
 
 func (b *balanceCache) BalanceAt(ctx context.Context, client BalanceReader, account common.Address, blockNumber *big.Int) (*big.Int, error) {
@@ -65,7 +65,7 @@ func (b *balanceCache) ReadCachedNonce(account common.Address, blockNumber *big.
 	b.rw.RLock()
 	defer b.rw.RUnlock()
 
-	return b.nonces[account][blockNumber]
+	return b.nonces[account][blockNumber.Uint64()]
 }
 
 func (b *balanceCache) sortRanges(account common.Address) {
@@ -74,7 +74,7 @@ func (b *balanceCache) sortRanges(account common.Address) {
 		keys = append(keys, int(k))
 	}
 
-	sort.Ints(keys)
+	sort.Ints(keys) // This will not work for keys > 2^31
 
 	ranges := []nonceRange{}
 	for _, k := range keys {
@@ -86,6 +86,9 @@ func (b *balanceCache) sortRanges(account common.Address) {
 }
 
 func (b *balanceCache) findNonceInRange(account common.Address, block *big.Int) *int64 {
+	b.rw.RLock()
+	defer b.rw.RUnlock()
+
 	for k := range b.sortedRanges[account] {
 		nr := b.sortedRanges[account][k]
 		cmpMin := nr.min.Cmp(block)
@@ -137,9 +140,9 @@ func (b *balanceCache) addNonceToCache(account common.Address, blockNumber *big.
 
 	_, exists := b.nonces[account]
 	if !exists {
-		b.nonces[account] = make(map[*big.Int]*int64)
+		b.nonces[account] = make(map[uint64]*int64)
 	}
-	b.nonces[account][blockNumber] = nonce
+	b.nonces[account][blockNumber.Uint64()] = nonce
 	b.updateNonceRange(account, blockNumber, nonce)
 }
 
@@ -166,8 +169,8 @@ func (b *balanceCache) NonceAt(ctx context.Context, client BalanceReader, accoun
 
 func newBalanceCache() *balanceCache {
 	return &balanceCache{
-		balances:     make(map[common.Address]map[*big.Int]*big.Int),
+		balances:     make(map[common.Address]map[uint64]*big.Int),
-		nonces:       make(map[common.Address]map[*big.Int]*int64),
+		nonces:       make(map[common.Address]map[uint64]*int64),
 		nonceRanges:  make(map[common.Address]map[int64]nonceRange),
 		sortedRanges: make(map[common.Address][]nonceRange),
 	}

services/wallet/transfer/commands.go

@@ -78,7 +78,8 @@ func (c *ethHistoricalCommand) Run(ctx context.Context) (err error) {
 	if c.from.Number != nil && c.from.Nonce != nil {
 		c.balanceCache.addNonceToCache(c.address, c.from.Number, c.from.Nonce)
 	}
-	from, headers, startBlock, err := findBlocksWithEthTransfers(ctx, c.chainClient, c.balanceCache, c.eth, c.address, c.from.Number, c.to, c.noLimit, c.threadLimit)
+	from, headers, startBlock, err := findBlocksWithEthTransfers(ctx, c.chainClient,
+		c.balanceCache, c.eth, c.address, c.from.Number, c.to, c.noLimit, c.threadLimit)
 
 	if err != nil {
 		c.error = err
@@ -387,6 +388,15 @@ func (c *transfersCommand) Run(ctx context.Context) (err error) {
 			log.Error("SaveTransfers error", "error", err)
 			return err
 		}
+	} else {
+		// 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", c.blockNum)
+
+		err = markBlocksAsLoaded(c.chainClient.ChainID, c.db.client, c.address, []*big.Int{c.blockNum})
+		if err != nil {
+			log.Error("Mark blocks loaded error", "error", err)
+			return err
+		}
 	}
 
 	c.fetchedTransfers = allTransfers
@@ -496,7 +506,8 @@ func (c *findAndCheckBlockRangeCommand) Run(parent context.Context) (err error)
 		foundHeaders[address] = uniqHeaders
 
 		lastBlockNumber := c.toByAddress[address]
-		log.Debug("saving headers", "len", len(uniqHeaders), "lastBlockNumber", lastBlockNumber, "balance", c.balanceCache.ReadCachedBalance(address, lastBlockNumber), "nonce", c.balanceCache.ReadCachedNonce(address, lastBlockNumber))
+		log.Debug("saving headers", "len", len(uniqHeaders), "lastBlockNumber", lastBlockNumber,
+			"balance", c.balanceCache.ReadCachedBalance(address, lastBlockNumber), "nonce", c.balanceCache.ReadCachedNonce(address, lastBlockNumber))
 		to := &Block{
 			Number:  lastBlockNumber,
 			Balance: c.balanceCache.ReadCachedBalance(address, lastBlockNumber),

services/wallet/transfer/commands_sequential.go

@@ -3,7 +3,6 @@ package transfer
 import (
 	"context"
 	"math/big"
-	"sort"
 	"time"
 
 	"github.com/pkg/errors"
@@ -56,7 +55,7 @@ func (c *findBlocksCommand) Run(parent context.Context) (err error) {
 			c.error = err
 		}
 
-		return
+		return nil // We break the loop if we fetched all the blocks
 	}
 
 	var head *types.Header = nil
@@ -93,7 +92,7 @@ func (c *findBlocksCommand) Run(parent context.Context) (err error) {
 
 		// 'to' is set to 'head' if 'last' block not found in DB
 		if head != nil && to.Cmp(head.Number) == 0 {
-			log.Info("update blockrange", "head", head.Number, "to", to, "chain", c.chainClient.ChainID, "account", c.account)
+			log.Info("upsert blockrange", "head", head.Number, "to", to, "chain", c.chainClient.ChainID, "account", c.account)
 
 			err = c.blockDAO.upsertRange(c.chainClient.ChainID, c.account, c.startBlockNumber,
 				c.resFromBlock.Number, to)
@@ -147,13 +146,13 @@ func (c *findBlocksCommand) checkRange(parent context.Context, from *big.Int, to
 	fromBlock := &Block{Number: from}
 
 	newFromBlock, ethHeaders, startBlock, err := c.fastIndex(parent, c.balanceCache, fromBlock, to)
-	log.Info("findBlocksCommand checkRange", "startBlock", startBlock, "newFromBlock", newFromBlock.Number, "toBlockNumber", to, "noLimit", c.noLimit)
 	if err != nil {
 		log.Info("findBlocksCommand checkRange fastIndex", "err", 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 nil, nil
 	}
+	log.Info("findBlocksCommand checkRange", "startBlock", startBlock, "newFromBlock", newFromBlock.Number, "toBlockNumber", to, "noLimit", c.noLimit)
 
 	// TODO There should be transfers when either when we have found headers
 	// or when newFromBlock is different from fromBlock, but if I check for
@@ -202,10 +201,6 @@ func (c *findBlocksCommand) checkRange(parent context.Context, from *big.Int, to
 			// return err
 			return nil, nil
 		}
-
-		sort.SliceStable(foundHeaders, func(i, j int) bool {
-			return foundHeaders[i].Number.Cmp(foundHeaders[j].Number) == 1
-		})
 	}
 	// }
 
@@ -348,7 +343,6 @@ type loadBlocksAndTransfersCommand struct {
 	db            *Database
 	blockRangeDAO *BlockRangeSequentialDAO
 	blockDAO      *BlockDAO
-	erc20         *ERC20TransfersDownloader
 	chainClient   *chain.ClientWithFallback
 	feed          *event.Feed
 	balanceCache  *balanceCache

services/wallet/transfer/concurrent.go

@@ -139,7 +139,7 @@ func checkRangesWithStartBlock(parent context.Context, client BalanceReader, cac
 				return err
 			}
 			if lb.Cmp(hb) == 0 {
-				log.Debug("balances are equal", "from", from, "to", to)
+				log.Debug("balances are equal", "from", from, "to", to, "lb", lb, "hb", hb)
 
 				hn, err := cache.NonceAt(ctx, client, account, to)
 				if err != nil {
@@ -166,7 +166,7 @@ func checkRangesWithStartBlock(parent context.Context, client BalanceReader, cac
 					return err
 				}
 				if *ln == *hn {
-					log.Debug("transaction count is also equal", "from", from, "to", to)
+					log.Debug("transaction count is also equal", "from", from, "to", to, "ln", *ln, "hn", *hn)
 					return nil
 				}
 			}

services/wallet/transfer/reactor.go

@@ -20,7 +20,7 @@ const (
 	ReactorNotStarted string = "reactor not started"
 
 	NonArchivalNodeBlockChunkSize = 100
-	DefaultNodeBlockChunkSize     = 10000
+	DefaultNodeBlockChunkSize     = 100000
 )
 
 var errAlreadyRunning = errors.New("already running")

services/wallet/transfer/sequential_fetch_strategy.go

@@ -6,7 +6,6 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/status-im/status-go/rpc/chain"
@@ -42,14 +41,12 @@ type SequentialFetchStrategy struct {
 func (s *SequentialFetchStrategy) newCommand(chainClient *chain.ClientWithFallback,
 	accounts []common.Address) async.Commander {
 
-	signer := types.NewLondonSigner(chainClient.ToBigInt())
 	ctl := &loadBlocksAndTransfersCommand{
 		db:                 s.db,
 		chainClient:        chainClient,
 		accounts:           accounts,
 		blockRangeDAO:      &BlockRangeSequentialDAO{s.db.client},
 		blockDAO:           s.blockDAO,
-		erc20:              NewERC20TransfersDownloader(chainClient, accounts, signer),
 		feed:               s.feed,
 		errorsCount:        0,
 		transactionManager: s.transactionManager,