Feature: receiver timeout in config

dipdup-io · Sep 6, 2024 · 24c517c · 24c517c
1 parent ebf8354
commit 24c517c
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 5 deletions.
diff --git a/build/dipdup.yml b/build/dipdup.yml
@@ -96,6 +96,7 @@ metadata:
     workers_count: ${RECEIVER_WORKERS_COUNT:-10}
     max_attempts: ${MAX_RETRY_COUNT:-5}
     delay: 10
+    timeout: ${RECEIVER_TIMEOUT:-10}
 
   http_timeout: 5
   max_retry_count_on_error: ${MAX_RETRY_COUNT:-5}

diff --git a/cmd/metadata/config.go b/cmd/metadata/config.go
@@ -80,7 +80,8 @@ type FillerConfig struct {
 
 // ReceiverConfig -
 type ReceiverConfig struct {
-	WorkersCount int `yaml:"workers_count" validate:"required,min=1"`
-	MaxAttempts  int `yaml:"max_attempts" validate:"omitempty,min=1"`
-	Delay        int `yaml:"delay" validate:"omitempty,min=1"`
+	WorkersCount int    `yaml:"workers_count" validate:"required,min=1"`
+	MaxAttempts  int    `yaml:"max_attempts" validate:"omitempty,min=1"`
+	Delay        int    `yaml:"delay" validate:"omitempty,min=1"`
+	Timeout      uint64 `yaml:"timeout" validate:"min=1"`
 }
diff --git a/cmd/metadata/receiver.go b/cmd/metadata/receiver.go
@@ -30,6 +30,7 @@ type Receiver struct {
 	maxAttempts  int
 	delay        int
 	workersCount int
+	timeout      time.Duration
 
 	wg *sync.WaitGroup
 }
@@ -40,6 +41,7 @@ func NewReceiver(cfg ReceiverConfig, tm storage.ITokenMetadata, ipfsNode *ipfs.N
 		workersCount = 10
 		maxAttempts  = 5
 		delay        = 10
+		timeout      = time.Second * 30
 	)
 
 	if cfg.WorkersCount > 0 {
@@ -51,6 +53,9 @@ func NewReceiver(cfg ReceiverConfig, tm storage.ITokenMetadata, ipfsNode *ipfs.N
 	if cfg.Delay > 0 {
 		delay = cfg.Delay
 	}
+	if cfg.Timeout > 0 {
+		timeout = time.Second * time.Duration(cfg.Timeout)
+	}
 
 	t := http.DefaultTransport.(*http.Transport).Clone()
 	t.MaxIdleConns = 100
@@ -67,6 +72,7 @@ func NewReceiver(cfg ReceiverConfig, tm storage.ITokenMetadata, ipfsNode *ipfs.N
 		maxAttempts:  maxAttempts,
 		delay:        delay,
 		workersCount: workersCount,
+		timeout:      timeout,
 		wg:           new(sync.WaitGroup),
 	}
 
@@ -143,7 +149,7 @@ func (r *Receiver) worker(ctx context.Context, task storage.TokenMetadata) {
 		Str("uri", *task.Uri).
 		Msg("try to receive token metadata")
 
-	timeoutCtx, cancel := context.WithTimeout(ctx, time.Second*30)
+	timeoutCtx, cancel := context.WithTimeout(ctx, r.timeout)
 	defer cancel()
 
 	var err error
@@ -160,7 +166,7 @@ func (r *Receiver) worker(ctx context.Context, task storage.TokenMetadata) {
 	if err != nil {
 		eStr := err.Error()
 		task.Error = &eStr
-		log.Err(ErrInvalidUri).Str("uri", *task.Uri).Uint("attempt", task.Attempts).Msg("fail to receive metadata")
+		log.Err(err).Str("uri", *task.Uri).Uint("attempt", task.Attempts).Msg("fail to receive metadata")
 
 		if task.Attempts == uint(r.maxAttempts) {
 			task.Status = storage.StatusFailed