fix: http retry strategy (#240)

.github/workflows/build.yml

@@ -44,7 +44,7 @@ jobs:
           go build -tags nosqlite -ldflags="-w -s" -buildmode=c-shared -o ui/flutter/windows/libgopeed.dll github.com/GopeedLab/gopeed/bind/desktop
           cd ui/flutter
           flutter build windows
-          $mingw = "C:\ProgramData\Chocolatey\lib\mingw\tools\install\mingw64\bin"
+          $mingw = "C:\Program Files\Git\mingw64\bin"
           $system = "C:\Windows\System32"
           $release = "build\windows\runner\Release\"
           cp $mingw\libstdc++-6.dll $release

internal/protocol/http/fetcher.go

@@ -18,6 +18,7 @@ import (
 	"os"
 	"path"
 	"strconv"
+	"time"
 )
 
 type RequestError struct {
@@ -37,6 +38,8 @@ type chunk struct {
 	Begin      int64
 	End        int64
 	Downloaded int64
+
+	retryTimes int
 }
 
 func newChunk(begin int64, end int64) *chunk {
@@ -255,24 +258,43 @@ func (f *Fetcher) fetch() {
 
 func (f *Fetcher) fetchChunk(index int, ctx context.Context) (err error) {
 	chunk := f.chunks[index]
+	chunk.retryTimes = 0
 
 	httpReq, err := buildRequest(ctx, f.meta.Req)
 	if err != nil {
 		return err
 	}
 	var (
-		client = buildClient()
-		buf    = make([]byte, 8192)
+		client     = buildClient()
+		buf        = make([]byte, 8192)
+		maxRetries = 3
 	)
-	// 重试5次
-	for i := 0; i < 5; i++ {
+	// retry until all remain chunks failed
+	for true {
 		// if chunk is completed, return
 		if f.meta.Res.Range && chunk.Downloaded >= chunk.End-chunk.Begin+1 {
 			return
 		}
+
+		if chunk.retryTimes >= maxRetries {
+			if !f.meta.Res.Range {
+				return
+			}
+			// check if all failed
+			allFailed := true
+			for _, c := range f.chunks {
+				if chunk.Downloaded < chunk.End-chunk.Begin+1 && c.retryTimes < maxRetries {
+					allFailed = false
+					break
+				}
+			}
+			if allFailed {
+				return
+			}
+		}
+
 		var (
-			resp  *http.Response
-			retry bool
+			resp *http.Response
 		)
 		if f.meta.Res.Range {
 			httpReq.Header.Set(base.HttpHeaderRange,
@@ -292,35 +314,41 @@ func (f *Fetcher) fetchChunk(index int, ctx context.Context) (err error) {
 			return nil
 		}()
 		if err != nil {
-			// retry request
+			// retry request after 1 second
+			chunk.retryTimes = chunk.retryTimes + 1
+			time.Sleep(time.Second)
 			continue
 		}
-		// 请求成功就重置错误次数，连续失败5次才终止
-		i = 0
-		retry, err = func() (bool, error) {
+		// Http request success, reset retry times
+		chunk.retryTimes = 0
+		err = func() error {
 			defer resp.Body.Close()
 			for {
 				n, err := resp.Body.Read(buf)
 				if n > 0 {
 					_, err := f.file.WriteAt(buf[:n], chunk.Begin+chunk.Downloaded)
 					if err != nil {
-						return false, err
+						return err
 					}
 					chunk.Downloaded += int64(n)
 				}
 				if err != nil {
-					if err != io.EOF {
-						return true, err
+					if err == io.EOF {
+						return nil
 					}
-					break
+					return err
 				}
 			}
-			return false, nil
 		}()
-		if !retry {
-			// 下载成功，跳出重试
-			break
+		// if chunk is completed, reset other not complete chunks retry times
+		if err == nil {
+			for _, c := range f.chunks {
+				if c != chunk && c.retryTimes > 0 {
+					c.retryTimes = 0
+				}
+			}
 		}
+		break
 	}
 	return
 }

internal/protocol/http/fetcher_test.go

@@ -76,7 +76,7 @@ func testResolve(startTestServer func() net.Listener, path string, want *base.Re
 func TestFetcher_DownloadNormal(t *testing.T) {
 	listener := test.StartTestFileServer()
 	defer listener.Close()
-	// 正常下载
+
 	downloadNormal(listener, 1, t)
 	downloadNormal(listener, 5, t)
 	downloadNormal(listener, 8, t)
@@ -86,7 +86,7 @@ func TestFetcher_DownloadNormal(t *testing.T) {
 func TestFetcher_DownloadContinue(t *testing.T) {
 	listener := test.StartTestFileServer()
 	defer listener.Close()
-	// 暂停继续
+
 	downloadContinue(listener, 1, t)
 	downloadContinue(listener, 5, t)
 	downloadContinue(listener, 8, t)
@@ -96,7 +96,7 @@ func TestFetcher_DownloadContinue(t *testing.T) {
 func TestFetcher_DownloadChunked(t *testing.T) {
 	listener := test.StartTestCustomServer()
 	defer listener.Close()
-	// chunked编码下载
+
 	downloadNormal(listener, 1, t)
 	downloadNormal(listener, 2, t)
 }
@@ -104,24 +104,33 @@ func TestFetcher_DownloadChunked(t *testing.T) {
 func TestFetcher_DownloadPost(t *testing.T) {
 	listener := test.StartTestPostServer()
 	defer listener.Close()
-	// post下载
+
 	downloadPost(listener, 1, t)
 }
 
 func TestFetcher_DownloadRetry(t *testing.T) {
 	listener := test.StartTestRetryServer()
 	defer listener.Close()
-	// chunked编码下载
+
 	downloadNormal(listener, 1, t)
 }
 
 func TestFetcher_DownloadError(t *testing.T) {
 	listener := test.StartTestErrorServer()
 	defer listener.Close()
-	// chunked编码下载
+
 	downloadError(listener, 1, t)
 }
 
+func TestFetcher_DownloadLimit(t *testing.T) {
+	listener := test.StartTestLimitServer()
+	defer listener.Close()
+
+	downloadNormal(listener, 1, t)
+	downloadNormal(listener, 2, t)
+	downloadNormal(listener, 8, t)
+}
+
 func TestFetcher_DownloadResume(t *testing.T) {
 	listener := test.StartTestFileServer()
 	defer listener.Close()

internal/test/httptest.go

@@ -9,6 +9,9 @@ import (
 	"net"
 	"net/http"
 	"os"
+	"strconv"
+	"strings"
+	"sync/atomic"
 	"time"
 )
 
@@ -81,7 +84,7 @@ func StartTestRetryServer() net.Listener {
 		mux := http.NewServeMux()
 		mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) {
 			counter++
-			if counter != 1 && counter < 5 {
+			if counter != 1 && counter < 2 {
 				writer.WriteHeader(500)
 				return
 			}
@@ -134,6 +137,79 @@ func StartTestErrorServer() net.Listener {
 	})
 }
 
+// StartTestLimitServer connections limit server
+func StartTestLimitServer() net.Listener {
+	var connections atomic.Int32
+
+	return startTestServer(func() http.Handler {
+		mux := http.NewServeMux()
+		mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) {
+			defer func() {
+				connections.Add(-1)
+			}()
+			connections.Add(1)
+			if connections.Load() > 4 {
+				writer.WriteHeader(403)
+				return
+			}
+
+			r := request.Header.Get("Range")
+			if r == "" {
+				writer.Header().Set("Content-Length", fmt.Sprintf("%d", BuildSize))
+				writer.WriteHeader(200)
+				file, err := os.Open(BuildFile)
+				if err != nil {
+					panic(err)
+				}
+				defer file.Close()
+				io.Copy(writer, file)
+			} else {
+				// split range
+				s := strings.Split(r, "=")
+				if len(s) != 2 {
+					writer.WriteHeader(400)
+					return
+				}
+				s = strings.Split(s[1], "-")
+				if len(s) != 2 {
+					writer.WriteHeader(400)
+					return
+				}
+				start, err := strconv.ParseInt(s[0], 10, 64)
+				if err != nil {
+					writer.WriteHeader(400)
+					return
+				}
+				end, err := strconv.ParseInt(s[1], 10, 64)
+				if err != nil {
+					writer.WriteHeader(400)
+					return
+				}
+				if start < 0 || end < 0 || start > end {
+					writer.WriteHeader(400)
+					return
+				}
+				if end >= BuildSize {
+					end = BuildSize - 1
+				}
+				writer.Header().Set("Content-Length", fmt.Sprintf("%d", end-start+1))
+				writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, BuildSize))
+				writer.Header().Set("Accept-Ranges", "bytes")
+				writer.WriteHeader(206)
+				file, err := os.Open(BuildFile)
+				if err != nil {
+					writer.WriteHeader(500)
+					return
+				}
+				defer file.Close()
+				file.Seek(start, 0)
+				io.CopyN(writer, file, end-start+1)
+			}
+		})
+		return mux
+	})
+}
+
 func startTestServer(serverHandle func() http.Handler) net.Listener {
 	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {