Fetching metadata with workers

cmd/metadata/config/config.go

@@ -114,6 +114,7 @@ type Settings struct {
 	MaxRetryCountOnError uint64   `yaml:"max_retry_count_on_error"`
 	Index                []string `yaml:"index"`
 	AWS                  AWS      `yaml:"aws"`
+	MaxCPU               uint64   `yaml:"max_cpu,omitempty"`
 }
 
 // Validate -
@@ -124,6 +125,9 @@ func (cfg *Settings) Validate() error {
 	if cfg.HTTPTimeout == 0 {
 		cfg.HTTPTimeout = 10
 	}
+	if cfg.MaxCPU == 0 {
+		cfg.MaxCPU = 4
+	}
 	if cfg.MaxRetryCountOnError == 0 {
 		cfg.MaxRetryCountOnError = 3
 	}

cmd/metadata/contract.go

@@ -78,25 +78,16 @@ func (indexer *Indexer) resolveContractMetadata(ctx context.Context, cm *models.
 	indexer.incrementCounter("contract", cm.Status)
 }
 
-func (indexer *Indexer) onContractTick(ctx context.Context) error {
-	uresolved, err := indexer.db.GetContractMetadata(models.StatusNew, 15, 0)
-	if err != nil {
-		return err
-	}
-	for i := range uresolved {
-		resolveCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
-		defer cancel()
+func (indexer *Indexer) contractWorker(ctx context.Context, contract models.ContractMetadata) error {
+	resolveCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
+	defer cancel()
 
-		indexer.resolveContractMetadata(resolveCtx, &uresolved[i])
+	indexer.resolveContractMetadata(resolveCtx, &contract)
 
-		if err := indexer.db.UpdateContractMetadata(&uresolved[i], map[string]interface{}{
-			"status":      uresolved[i].Status,
-			"metadata":    uresolved[i].Metadata,
-			"retry_count": uresolved[i].RetryCount,
-			"update_id":   uresolved[i].UpdateID,
-		}); err != nil {
-			return err
-		}
-	}
-	return nil
+	return indexer.db.UpdateContractMetadata(&contract, map[string]interface{}{
+		"status":      contract.Status,
+		"metadata":    contract.Metadata,
+		"retry_count": contract.RetryCount,
+		"update_id":   contract.UpdateID,
+	})
 }

cmd/metadata/indexer.go

@@ -34,8 +34,8 @@ type Indexer struct {
 	scanner   *tzkt.Scanner
 	prom      *prometheus.Service
 	ctx       *internalContext.Context
-	contracts *service.Service
-	tokens    *service.Service
+	contracts *service.ContractService
+	tokens    *service.TokenService
 	thumbnail *thumbnail.Service
 	settings  config.Settings
 
@@ -71,8 +71,8 @@ func NewIndexer(ctx context.Context, network string, indexerConfig *config.Index
 	if aws := storage.NewAWS(settings.AWS.AccessKey, settings.AWS.Secret, settings.AWS.Region, settings.AWS.BucketName); aws != nil {
 		indexer.thumbnail = thumbnail.New(aws, db, prom, network, settings.IPFSGateways, 10)
 	}
-	indexer.contracts = service.New(indexer.onContractTick, service.WithName("contracts"))
-	indexer.tokens = service.New(indexer.onTokenTick, service.WithName("tokens"))
+	indexer.contracts = service.NewContractService(db, indexer.contractWorker)
+	indexer.tokens = service.NewTokenService(db, indexer.tokenWorker)
 
 	return indexer, nil
 }

cmd/metadata/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"os"
 	"os/signal"
+	"runtime"
 	"sync"
 	"syscall"
 	"time"
@@ -44,6 +45,7 @@ func main() {
 		log.Error(err)
 		return
 	}
+	runtime.GOMAXPROCS(int(cfg.Metadata.Settings.MaxCPU))
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

cmd/metadata/service/contract.go

@@ -0,0 +1,75 @@
+package service
+
+import (
+	"context"
+	"sync"
+
+	"github.com/dipdup-net/metadata/cmd/metadata/models"
+	log "github.com/sirupsen/logrus"
+)
+
+// ContractService -
+type ContractService struct {
+	repo    models.ContractRepository
+	handler func(ctx context.Context, contract models.ContractMetadata) error
+	workers chan struct{}
+	wg      sync.WaitGroup
+}
+
+// NewContractService -
+func NewContractService(repo models.ContractRepository, handler func(context.Context, models.ContractMetadata) error) *ContractService {
+	return &ContractService{
+		repo:    repo,
+		handler: handler,
+		workers: make(chan struct{}, 10),
+	}
+}
+
+// Start -
+func (s *ContractService) Start(ctx context.Context) {
+	s.wg.Add(1)
+	go s.manager(ctx)
+}
+
+// Close -
+func (s *ContractService) Close() error {
+	s.wg.Wait()
+
+	close(s.workers)
+	return nil
+}
+
+func (s *ContractService) manager(ctx context.Context) {
+	defer s.wg.Done()
+
+	if s.handler == nil {
+		log.Warn("processor contract metadata service: service without handler")
+		return
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			unresolved, err := s.repo.GetContractMetadata(models.StatusNew, 15, 0)
+			if err != nil {
+				log.Error(err)
+				continue
+			}
+			for i := range unresolved {
+				s.workers <- struct{}{}
+				s.wg.Add(1)
+				go func(contract models.ContractMetadata) {
+					defer func() {
+						<-s.workers
+						s.wg.Done()
+					}()
+					if err := s.handler(ctx, contract); err != nil {
+						log.Error(err)
+					}
+				}(unresolved[i])
+			}
+		}
+	}
+}

cmd/metadata/service/token.go

@@ -0,0 +1,75 @@
+package service
+
+import (
+	"context"
+	"sync"
+
+	"github.com/dipdup-net/metadata/cmd/metadata/models"
+	log "github.com/sirupsen/logrus"
+)
+
+// TokenService -
+type TokenService struct {
+	repo    models.TokenRepository
+	handler func(ctx context.Context, contract models.TokenMetadata) error
+	workers chan struct{}
+	wg      sync.WaitGroup
+}
+
+// NewContractService -
+func NewTokenService(repo models.TokenRepository, handler func(context.Context, models.TokenMetadata) error) *TokenService {
+	return &TokenService{
+		repo:    repo,
+		handler: handler,
+		workers: make(chan struct{}, 10),
+	}
+}
+
+// Start -
+func (s *TokenService) Start(ctx context.Context) {
+	s.wg.Add(1)
+	go s.manager(ctx)
+}
+
+// Close -
+func (s *TokenService) Close() error {
+	s.wg.Wait()
+
+	close(s.workers)
+	return nil
+}
+
+func (s *TokenService) manager(ctx context.Context) {
+	defer s.wg.Done()
+
+	if s.handler == nil {
+		log.Warn("processor token metadata service: service without handler")
+		return
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			unresolved, err := s.repo.GetTokenMetadata(models.StatusNew, 15, 0)
+			if err != nil {
+				log.Error(err)
+				continue
+			}
+			for i := range unresolved {
+				s.workers <- struct{}{}
+				s.wg.Add(1)
+				go func(contract models.TokenMetadata) {
+					defer func() {
+						<-s.workers
+						s.wg.Done()
+					}()
+					if err := s.handler(ctx, contract); err != nil {
+						log.Error(err)
+					}
+				}(unresolved[i])
+			}
+		}
+	}
+}

cmd/metadata/token.go

@@ -183,27 +183,17 @@ func mergeTokenMetadata(src, got []byte) ([]byte, error) {
 	return json.Marshal(srcMap)
 }
 
-func (indexer *Indexer) onTokenTick(ctx context.Context) error {
-	uresolved, err := indexer.db.GetTokenMetadata(models.StatusNew, 15, 0)
-	if err != nil {
-		return err
-	}
+func (indexer *Indexer) tokenWorker(ctx context.Context, token models.TokenMetadata) error {
+	resolveCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
 
-	for i := range uresolved {
-		resolveCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
-		defer cancel()
-
-		if err := indexer.resolveTokenMetadata(resolveCtx, &uresolved[i]); err != nil {
-			return err
-		}
-		if err := indexer.db.UpdateTokenMetadata(&uresolved[i], map[string]interface{}{
-			"status":      uresolved[i].Status,
-			"metadata":    uresolved[i].Metadata,
-			"retry_count": uresolved[i].RetryCount,
-			"update_id":   uresolved[i].UpdateID,
-		}); err != nil {
-			return err
-		}
+	if err := indexer.resolveTokenMetadata(resolveCtx, &token); err != nil {
+		return err
 	}
-	return nil
+	return indexer.db.UpdateTokenMetadata(&token, map[string]interface{}{
+		"status":      token.Status,
+		"metadata":    token.Metadata,
+		"retry_count": token.RetryCount,
+		"update_id":   token.UpdateID,
+	})
 }