new 'next' method for updater to get the next token to update in each…

… iteration an support sorted scan

cmd/census3/main.go

@@ -209,7 +209,7 @@ func main() {
  log.Fatal(err)
  }
  // start the token updater with the database and the provider manager
- updater := scanner.NewUpdater(database, w3p, pm, filtersDB)
+ updater := scanner.NewUpdater(database, w3p, pm, filtersDB, config.scannerCoolDown)
  // start the holder scanner with the database and the provider manager
  hc := scanner.NewScanner(database, updater, w3p, pm, config.scannerCoolDown)
  // if the admin token is not defined, generate a random one

scanner/providers/web3/erc20_provider.go

@@ -159,6 +159,7 @@ func (p *ERC20HolderProvider) HoldersBalances(ctx context.Context, _ []byte, fro
  alreadyProcessedLogs := uint64(0)
  balances := make(map[common.Address]*big.Int)
  // iterate the logs and update the balances
+ log.Infow("parsing logs", "address", p.address, "type", p.TypeName(), "count", len(logs))
  for _, currentLog := range logs {
  // skip the log if it has been removed
  if currentLog.Removed {
@@ -208,7 +209,7 @@ func (p *ERC20HolderProvider) HoldersBalances(ctx context.Context, _ []byte, fro
  balances[logData.From] = new(big.Int).Neg(logData.Value)
  }
  }
- log.Infow("saving blocks",
+ log.Infow("logs parsed",
  "count", len(balances),
  "new_logs", newTransfers,
  "already_processed_logs", alreadyProcessedLogs,

scanner/providers/web3/erc721_provider.go

@@ -152,7 +152,6 @@ func (p *ERC721HolderProvider) HoldersBalances(ctx context.Context, _ []byte, fr
  if errors.Is(err, ErrTooManyRequests) {
  log.Warnf("too many requests, the provider will continue in the next iteration from block %d", lastBlock)
  }
- log.Warnw("logs received", "number_of_logs", len(logs), "last_block", lastBlock)
  // encode the number of new transfers
  newTransfers := uint64(0)
  alreadyProcessedLogs := uint64(0)
@@ -399,6 +398,6 @@ func (p *ERC721HolderProvider) isLogAlreadyProcessed(l types.Log) (bool, error)
  if _, err := hashFn.Write([]byte(transferID)); err != nil {
  return false, err
  }
- hID := hashFn.Sum(nil)
+ hID := hashFn.Sum(nil)[:8]
  return p.filter.TestAndAdd(hID, nil)
 }

scanner/providers/web3/erc777_provider.go

@@ -152,7 +152,6 @@ func (p *ERC777HolderProvider) HoldersBalances(ctx context.Context, _ []byte, fr
  if errors.Is(err, ErrTooManyRequests) {
  log.Warnf("too many requests, the provider will continue in the next iteration from block %d", lastBlock)
  }
- log.Warnw("logs received", "number_of_logs", len(logs), "last_block", lastBlock)
  // encode the number of new transfers
  newTransfers := uint64(0)
  alreadyProcessedLogs := uint64(0)
@@ -399,6 +398,6 @@ func (p *ERC777HolderProvider) isLogAlreadyProcessed(l types.Log) (bool, error)
  if _, err := hashFn.Write([]byte(transferID)); err != nil {
  return false, err
  }
- hID := hashFn.Sum(nil)
+ hID := hashFn.Sum(nil)[:8]
  return p.filter.TestAndAdd(hID, nil)
 }

scanner/scanner.go

@@ -109,7 +109,6 @@ func (s *Scanner) Start(ctx context.Context) {
  continue
  }
  }
-
  log.Infow("checking token in the updater queue",
  "address", token.Address.Hex(),
  "chainID", token.ChainID,
@@ -435,6 +434,7 @@ func (s *Scanner) prepareToken(token *ScannerToken) error {
  return err
  }
  token.CreationBlock = creationBlock
+ token.LastBlock = creationBlock
  token.Ready = true
  }
  return nil

scanner/updater.go

@@ -7,6 +7,7 @@ import (
  "fmt"
  "math/big"
  "sync"
+ "sync/atomic"
  "time"
 
  "github.com/ethereum/go-ethereum/common"
@@ -51,62 +52,73 @@ type Updater struct {
  ctx context.Context
  cancel context.CancelFunc
 
- db *db.DB
- networks *web3.Web3Pool
- providers *manager.ProviderManager
- queue map[string]*UpdateRequest
- queueMtx sync.Mutex
- processing sync.Map
- waiter sync.WaitGroup
- kvdb dvotedb.Database
+ db *db.DB
+ networks *web3.Web3Pool
+ providers *manager.ProviderManager
+ sortedQueue []string
+ queue map[string]*UpdateRequest
+ queueMtx sync.Mutex
+ processing sync.Map
+ nextReq atomic.Uint64
+ waiter sync.WaitGroup
+ kvdb dvotedb.Database
+ coolDown time.Duration
 }
 
 // NewUpdater creates a new instance of Updater.
 func NewUpdater(db *db.DB, networks *web3.Web3Pool, pm *manager.ProviderManager,
- kvdb dvotedb.Database,
+ kvdb dvotedb.Database, coolDown time.Duration,
 ) *Updater {
  return &Updater{
- db: db,
- networks: networks,
- providers: pm,
- queue: make(map[string]*UpdateRequest),
- kvdb: kvdb,
+ db: db,
+ networks: networks,
+ providers: pm,
+ sortedQueue: []string{},
+ queue: make(map[string]*UpdateRequest),
+ kvdb: kvdb,
+ coolDown: coolDown,
  }
 }
 
 // Start starts the updater process in a goroutine.
 func (u *Updater) Start(ctx context.Context, concurrentTokens int) {
  u.ctx, u.cancel = context.WithCancel(ctx)
+ sem := make(chan struct{}, concurrentTokens)
+ defer close(sem)
  for {
  select {
  case <-u.ctx.Done():
  return
  default:
- pending := u.pendingRequests()
- if len(pending) == 0 {
- time.Sleep(coolDown)
+ req, id := u.next()
+ if req == nil {
+ log.Info("no more requests to process, sleeping...")
+ time.Sleep(u.coolDown)
  continue
  }
- sem := make(chan struct{}, concurrentTokens)
- defer close(sem)
- for id, req := range u.pendingRequests() {
- u.processing.Store(id, true)
- sem <- struct{}{}
- go func(id string, req *UpdateRequest) {
- defer func() {
- <-sem
- u.processing.Store(id, false)
- }()
- if err := u.process(req); err != nil {
- log.Errorf("Error processing update request: %v", err)
- return
- }
- // update the request in the queue
- u.queueMtx.Lock()
- u.queue[id] = req
- u.queueMtx.Unlock()
- }(id, req)
- }
+ sem <- struct{}{}
+ u.waiter.Add(1)
+ go func(id string, req UpdateRequest) {
+ defer func() {
+ u.waiter.Done()
+ <-sem
+ }()
+ log.Infow("processing token",
+ "address", req.Address.Hex(),
+ "from", req.CreationBlock,
+ "to", req.EndBlock,
+ "current", req.LastBlock)
+ res, err := u.process(id, req)
+ if err != nil {
+ log.Errorf("error processing update request: %v", err)
+ return
+ }
+ // update the request in the queue
+ log.Infow("updating request in the queue", "lastBlock", req.LastBlock, "done", req.Done)
+ if err := u.SetRequest(id, &res); err != nil {
+ log.Errorf("error updating request in the queue: %v", err)
+ }
+ }(id, *req)
  }
  }
 }
@@ -127,12 +139,20 @@ func (u *Updater) RequestStatus(id string, deleteOnDone bool) *UpdateRequest {
  if !ok {
  return nil
  }
- res := *req
  if deleteOnDone && req.Done {
+ // remove the request from the processing map
  u.processing.Delete(id)
+ // remove the request from the queue
  delete(u.queue, id)
+ // remove the request from the sorted queue
+ for i, v := range u.sortedQueue {
+ if v == id {
+ u.sortedQueue = append(u.sortedQueue[:i], u.sortedQueue[i+1:]...)
+ break
+ }
+ }
  }
- return &res
+ return req
 }
 
 // SetRequest adds a new request to the queue. It will return an error if the
@@ -157,8 +177,10 @@ func (u *Updater) SetRequest(id string, req *UpdateRequest) error {
  }
  u.queueMtx.Lock()
  defer u.queueMtx.Unlock()
+ if _, exists := u.queue[id]; !exists {
+ u.sortedQueue = append(u.sortedQueue, id)
+ }
  u.queue[id] = req
- u.processing.Store(id, false)
  return nil
 }
 
@@ -195,48 +217,71 @@ func RequestID(address common.Address, chainID uint64, externalID string) (strin
  return hex.EncodeToString(bHash[:4]), nil
 }
 
-func (u *Updater) pendingRequests() map[string]*UpdateRequest {
+func (u *Updater) next() (*UpdateRequest, string) {
  u.queueMtx.Lock()
  defer u.queueMtx.Unlock()
- queue := map[string]*UpdateRequest{}
- for k, v := range u.queue {
- if processing, ok := u.processing.Load(k); v.Done || !ok || processing.(bool) {
- continue
+ // if the queue is empty return nil
+ if len(u.sortedQueue) == 0 {
+ return nil, ""
+ }
+ // get the next request in the queue, if the next request is out of the
+ // range of the sorted queue, return nil and set the next request index to 0
+ i := u.nextReq.Load()
+ max := uint64(len(u.sortedQueue))
+ if i >= max {
+ u.nextReq.Store(0)
+ return nil, ""
+ }
+ // iterate over the sorted queue to find the next request that is not being
+ // processed or already done
+ for ; i < max; i++ {
+ id := u.sortedQueue[i]
+ req, exists := u.queue[id]
+ if !exists {
+ // if the request is not found, remove the ID from the sorted queue and
+ // return nil setting the next request index to 0
+ u.sortedQueue = append(u.sortedQueue[:i], u.sortedQueue[i+1:]...)
+ u.nextReq.Store(0)
+ return nil, ""
+ }
+ // if request is not done and not being processed, return it
+ if isProcessing, ok := u.processing.Load(id); !req.Done && (!ok || !isProcessing.(bool)) {
+ u.nextReq.Store(i + 1)
+ return req, id
  }
- queue[k] = v
  }
- return queue
+ // if the next request is not found, set the next request index to 0
+ u.nextReq.Store(0)
+ return nil, ""
 }
 
 // process iterates over the current queue items, getting the token holders
 // balances and saving them in the database until the last block is greater or
 // equal to the end block. It updates th status of the request in the queue. It
 // will return an error if the provider is not found, the token is external or
 // there is an error getting the token holders balances.
-func (u *Updater) process(req *UpdateRequest) error {
- // log the start of the process
- log.Infow("rescanning token",
- "address", req.Address.Hex(),
- "from", req.CreationBlock,
- "to", req.EndBlock,
- "current", req.LastBlock)
+func (u *Updater) process(id string, req UpdateRequest) (UpdateRequest, error) {
+ // set the request as processing and defer to set it as not processing
+ u.processing.Store(id, true)
+ defer u.processing.Store(id, false)
+ // create a context with a timeout to avoid blocking the process
  ctx, cancel := context.WithTimeout(u.ctx, UPDATE_TIMEOUT)
  defer cancel()
  // get the provider by token type
  provider, err := u.providers.GetProvider(ctx, req.Type)
  if err != nil {
- return fmt.Errorf("error getting provider for token: %v", err)
+ return req, fmt.Errorf("error getting provider for token: %v", err)
  }
  // if the token is a external token, return an error
  if !provider.IsExternal() {
  chainAddress, ok := u.networks.ChainAddress(req.ChainID, req.Address.Hex())
  if !ok {
- return fmt.Errorf("error getting chain address for token: %v", err)
+ return req, fmt.Errorf("error getting chain address for token: %v", err)
  }
  // load filter of the token from the database
  filter, err := treedb.LoadTree(u.kvdb, chainAddress)
  if err != nil {
- return err
+ return req, err
  }
  // set the reference of the token to update in the provider
  if err := provider.SetRef(web3provider.Web3ProviderRef{
@@ -245,7 +290,7 @@ func (u *Updater) process(req *UpdateRequest) error {
  CreationBlock: req.CreationBlock,
  Filter: filter,
  }); err != nil {
- return fmt.Errorf("error setting provider reference: %v", err)
+ return req, fmt.Errorf("error setting provider reference: %v", err)
  }
  }
  // update the last block number of the provider to the last block of
@@ -257,7 +302,7 @@ func (u *Updater) process(req *UpdateRequest) error {
  ChainID: req.ChainID,
  })
  if err != nil {
- return fmt.Errorf("error getting token holders from database: %v", err)
+ return req, fmt.Errorf("error getting token holders from database: %v", err)
  }
  currentHolders := map[common.Address]*big.Int{}
  for _, holder := range results {
@@ -275,7 +320,7 @@ func (u *Updater) process(req *UpdateRequest) error {
  }
  // set the current holders in the provider
  if err := provider.SetLastBalances(ctx, nil, currentHolders, req.LastBlock); err != nil {
- return fmt.Errorf("error setting last balances in provider: %v", err)
+ return req, fmt.Errorf("error setting last balances in provider: %v", err)
  }
  // get range balances from the provider, it will check itereate again
  // over transfers logs, checking if there are new transfers using the
@@ -296,7 +341,7 @@ func (u *Updater) process(req *UpdateRequest) error {
  }
  }
  if err != nil {
- return fmt.Errorf("error getting token holders balances: %v", err)
+ return req, fmt.Errorf("error getting token holders balances: %v", err)
  }
  log.Debugw("new logs received",
  "address", req.Address.Hex(),
@@ -311,13 +356,12 @@ func (u *Updater) process(req *UpdateRequest) error {
  ChainID: req.ChainID,
  }, balances, delta.NewLogsCount, delta.Block, delta.Synced, delta.TotalSupply)
  if err != nil {
- return fmt.Errorf("error saving token holders balances: %v", err)
+ return req, fmt.Errorf("error saving token holders balances: %v", err)
  }
  log.Debugw("token holders balances updated",
  "token", req.Address.Hex(),
  "chainID", req.ChainID,
  "created", created,
  "updated", updated)
- log.Infow("updating request in the queue", "lastBlock", req.LastBlock, "done", req.Done)
- return nil
+ return req, nil
 }