Merge pull request #1758 from 0chain/feat/retry-refpath

Add retries for refpath request

Author: dabasov

zboxcore/sdk/allocation.go

@@ -422,7 +422,7 @@ func SetDownloadWorkerCount(count int) {
 
 // InitAllocation initializes the allocation.
 func (a *Allocation) InitAllocation() {
-	a.downloadChan = make(chan *DownloadRequest, 100)
+	a.downloadChan = make(chan *DownloadRequest, 400)
 	a.repairChan = make(chan *RepairRequest, 1)
 	a.ctx, a.ctxCancelF = context.WithCancel(context.Background())
 	a.downloadProgressMap = make(map[string]*DownloadRequest)

zboxcore/sdk/commitworker.go

@@ -471,6 +471,7 @@ func (commitReq *CommitRequestV2) processCommit() {
 		return
 	}
 	if commitReq.commitMask.CountOnes() < commitReq.consensusThresh {
+		commitReq.commitMask = zboxutil.NewUint128(0)
 		commitReq.result = ErrorCommitResult("Failed to get reference path")
 		return
 	}
@@ -624,41 +625,78 @@ func getReferencePathV2(blobber *blockchain.StorageNode, allocationID, allocatio
 		return trie, nil
 	}
 	now := time.Now()
-	req, err := zboxutil.NewReferencePathRequestV2(blobber.Baseurl, allocationID, allocationTx, sig, paths, false)
-	if err != nil {
-		l.Logger.Error("Creating ref path req", err)
-		return nil, err
-	}
-	var lR ReferencePathResultV2
-	ctx, cncl := context.WithTimeout(context.Background(), (time.Second * 30))
-	err = zboxutil.HttpDo(ctx, cncl, req, func(resp *http.Response, err error) error {
-		if err != nil {
-			l.Logger.Error("Ref path error:", err)
-			return err
-		}
-		defer resp.Body.Close()
-		respBody, err := io.ReadAll(resp.Body)
-		if err != nil {
-			l.Logger.Error("Ref path: Resp", err)
-			return err
-		}
-		if resp.StatusCode != http.StatusOK {
-			return errors.New(
-				strconv.Itoa(resp.StatusCode),
-				fmt.Sprintf("Reference path error response: Status: %d - %s ",
-					resp.StatusCode, string(respBody)))
-		}
-		err = json.Unmarshal(respBody, &lR)
-		if err != nil {
-			l.Logger.Error("Reference path json decode error: ", err)
-			return err
-		}
-		return nil
-	})
+	var (
+		shouldContinue bool
+		err            error
+		lR             ReferencePathResultV2
+	)
+	for retries := 0; retries < 3; retries++ {
+		err, shouldContinue = func() (err error, shouldContinue bool) {
+			var req *http.Request
+			req, err = zboxutil.NewReferencePathRequestV2(blobber.Baseurl, allocationID, allocationTx, sig, paths, false)
+			if err != nil {
+				l.Logger.Error("Creating ref path req", err)
+				return
+			}
 
+			ctx, cncl := context.WithTimeout(context.Background(), (time.Second * 30))
+			err = zboxutil.HttpDo(ctx, cncl, req, func(resp *http.Response, err error) error {
+				if err != nil {
+					l.Logger.Error("Ref path error:", err)
+					if errors.Is(err, http.ErrServerClosed) || strings.Contains(err.Error(), "GOAWAY") {
+						shouldContinue = true
+					}
+					return err
+				}
+				defer resp.Body.Close()
+				respBody, err := io.ReadAll(resp.Body)
+				if err != nil {
+					l.Logger.Error("Ref path: Resp", err)
+					if strings.Contains(err.Error(), "GOAWAY") || errors.Is(err, io.ErrUnexpectedEOF) {
+						shouldContinue = true
+					}
+					return err
+				}
+				if resp.StatusCode != http.StatusOK {
+					if resp.StatusCode == http.StatusTooManyRequests {
+						logger.Logger.Debug(blobber.Baseurl,
+							" got too many request error. Retrying")
+
+						var r int
+						r, err = zboxutil.GetRateLimitValue(resp)
+						if err != nil {
+							logger.Logger.Error(err)
+							return err
+						}
+
+						time.Sleep(time.Duration(r) * time.Second)
+						shouldContinue = true
+						return err
+					}
+					return errors.New(
+						strconv.Itoa(resp.StatusCode),
+						fmt.Sprintf("Reference path error response: Status: %d - %s ",
+							resp.StatusCode, string(respBody)))
+				}
+				err = json.Unmarshal(respBody, &lR)
+				if err != nil {
+					l.Logger.Error("Reference path json decode error: ", err)
+					return err
+				}
+				return nil
+			})
+			return
+		}()
+		if shouldContinue {
+			continue
+		} else {
+			break
+		}
+	}
 	if err != nil {
 		return nil, err
 	}
+
 	elapsedRefPath := time.Since(now)
 	mu.Lock()
 	defer mu.Unlock()
@@ -725,6 +763,9 @@ func submitWriteMarker(wmData, metaData []byte, blobber *blockchain.StorageNode,
 			respBody, err = io.ReadAll(resp.Body)
 			if err != nil {
 				logger.Logger.Error("Response read: ", err)
+				if errors.Is(err, io.ErrUnexpectedEOF) {
+					shouldContinue = true
+				}
 				return
 			}
 			if resp.StatusCode == http.StatusOK {
@@ -764,6 +805,13 @@ func submitWriteMarker(wmData, metaData []byte, blobber *blockchain.StorageNode,
 				return
 			}
 
+			trimmed := strings.TrimSpace(string(respBody))
+			if strings.HasPrefix(trimmed, "<html>") {
+				time.Sleep(5 * time.Second)
+				shouldContinue = true
+				return
+			}
+
 			err = thrown.New("commit_error",
 				fmt.Sprintf("Got error response %s with status %d", respBody, resp.StatusCode))
 			return

zboxcore/sdk/downloadworker.go

@@ -737,7 +737,7 @@ func (req *DownloadRequest) processDownload() {
 			if startBlock+int64(j)*numBlocks+numBlocks > endBlock {
 				blocksToDownload = endBlock - (startBlock + int64(j)*numBlocks)
 			}
-			data, err := req.getBlocksData(startBlock+int64(j)*numBlocks, blocksToDownload, j == 0)
+			data, err := req.getBlocksData(startBlock+int64(j)*numBlocks, blocksToDownload, j == 0 && shouldTimeRequest)
 			if req.isDownloadCanceled {
 				return errors.New("download_abort", "Download aborted by user")
 			}
@@ -792,7 +792,7 @@ func (req *DownloadRequest) processDownload() {
 	wg.Wait()
 	// req.fileHandler.Sync() //nolint
 	elapsedGetBlocksAndWrite := time.Since(now) - elapsedInitEC - elapsedInitEncryption
-	l.Logger.Debug(fmt.Sprintf("[processDownload] Timings:\n allocation_id: %s,\n remotefilepath: %s,\n initEC: %d ms,\n initEncryption: %d ms,\n getBlocks and writes: %d ms",
+	l.Logger.Info(fmt.Sprintf("[processDownload] Timings:\n allocation_id: %s,\n remotefilepath: %s,\n initEC: %d ms,\n initEncryption: %d ms,\n getBlocks and writes: %d ms",
 		req.allocationID,
 		req.remotefilepath,
 		elapsedInitEC.Milliseconds(),

zboxcore/sdk/sdk.go

@@ -54,6 +54,7 @@ var (
 	networkWorkerTimerInHours = 1 //nolint:unused
 	singleClientMode          = false
 	shouldVerifyHash          = true
+	shouldTimeRequest         = true
 )
 
 func SetSingleClietnMode(mode bool) {
@@ -64,6 +65,10 @@ func SetShouldVerifyHash(verify bool) {
 	shouldVerifyHash = verify
 }
 
+func SetShouldTimeRequest(timeRequest bool) {
+	shouldTimeRequest = timeRequest
+}
+
 func SetSaveProgress(save bool) {
 	shouldSaveProgress = save
 }

zboxcore/zboxutil/download_buffer.go

@@ -105,6 +105,10 @@ func (r *DownloadBufferWithMask) RequestChunk(ctx context.Context, num int) []by
 		default:
 		}
 		r.mu.Lock()
+		if r.downloadBuf == nil {
+			r.mu.Unlock()
+			return nil
+		}
 		isSet := r.mask & (1 << num)
 		// already assigned
 		if isSet == 0 {
@@ -137,9 +141,14 @@ func (r *DownloadBufferWithMask) ReleaseChunk(num int) {
 }
 
 func (r *DownloadBufferWithMask) ClearBuffer() {
-	for _, buff := range r.downloadBuf {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for num, buff := range r.downloadBuf {
 		if buff != nil {
-			BufferPool.Put(buff)
+			isSet := r.mask & (1 << num)
+			if isSet != 0 {
+				BufferPool.Put(buff)
+			}
 		}
 	}
 	r.downloadBuf = nil