Replace grab/v3 with stdlib HTTP fetcher

Add multipart download, per-part retries, resume support,
context cancellation, and per-file pipe processing.
Remove cavaliergopher/grab/v3 dependency.

go.mod

@@ -3,7 +3,6 @@ module github.com/giongto35/cloud-game/v3
 go 1.26
 
 require (
-	github.com/cavaliergopher/grab/v3 v3.0.1
 	github.com/fsnotify/fsnotify v1.10.1
 	github.com/go-gst/go-gst v1.4.0
 	github.com/gofrs/flock v0.13.0
@@ -39,7 +38,7 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/philhofer/fwd v1.2.0 // indirect
 	github.com/pion/datachannel v1.6.2 // indirect
-	github.com/pion/dtls/v3 v3.1.4 // indirect
+	github.com/pion/dtls/v3 v3.1.5 // indirect
 	github.com/pion/mdns/v2 v2.1.0 // indirect
 	github.com/pion/randutil v0.1.0 // indirect
 	github.com/pion/rtcp v1.2.17 // indirect

go.sum

@@ -1,5 +1,3 @@
-github.com/cavaliergopher/grab/v3 v3.0.1 h1:4z7TkBfmPjmLAAmkkAZNX/6QJ1nNFdv3SdIHXju0Fr4=
-github.com/cavaliergopher/grab/v3 v3.0.1/go.mod h1:1U/KNnD+Ft6JJiYoYBAimKH2XrYptb8Kl3DFGmsjpq4=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -56,8 +54,8 @@ github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM=
 github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
 github.com/pion/datachannel v1.6.2 h1:7EXQ8TH3vTouBUdRWYbcX2edSx9Yj6k5zl5P+qyxEPc=
 github.com/pion/datachannel v1.6.2/go.mod h1:pzbdAZvyGtXbcHM1hBbsFaOTf40lZizU/dNlvVOak6E=
-github.com/pion/dtls/v3 v3.1.4 h1:QhvtMflMfu9Kf0RcDC5BJBle4caPskByrKQR6uuYqpY=
-github.com/pion/dtls/v3 v3.1.4/go.mod h1:cr/qotLISUw/9C1m83ZPNZtj9WnXkYLpfCptPqbkInc=
+github.com/pion/dtls/v3 v3.1.5 h1:9xJtVsHwMYeSjPp5Hh1FTis4DchnQWtnOa5o+6ygqfc=
+github.com/pion/dtls/v3 v3.1.5/go.mod h1:gz1K4jg6c+fq86oQMH4pilpCEOEPwmEr2jY+VcF/mkU=
 github.com/pion/ice/v4 v4.2.7 h1:zDEbC6MiEdhQpF8TxBOTws+NU6ZgGpveHrQq4Lc1kao=
 github.com/pion/ice/v4 v4.2.7/go.mod h1:9SNPaq0c7El/ki8leJzyCkK10zsskprR3zTNbO3monY=
 github.com/pion/interceptor v0.1.45 h1:6PUo/5829bIfRFIPPJQzuDn8EjxRTSB/CSD7QVCOaqo=

pkg/worker/caged/libretro/caged.go

@@ -28,9 +28,11 @@ func Cage(conf CagedConf, log *logger.Logger) Caged {
 }
 
 func (c *Caged) Init() error {
-	if err := manager.CheckCores(c.conf.Emulator, c.log); err != nil {
-		c.log.Warn().Err(err).Msgf("a Libretro cores sync fail")
-	}
+	go func() {
+		if err := manager.CheckCores(c.conf.Emulator, c.log); err != nil {
+			c.log.Warn().Err(err).Msgf("a Libretro cores sync fail")
+		}
+	}()
 
 	if c.conf.Emulator.FailFast {
 		if err := c.IsSupported(); err != nil {

pkg/worker/caged/libretro/manager/downloader.go

@@ -1,6 +1,7 @@
 package manager
 
 import (
+	"context"
 	"os"
 
 	"github.com/giongto35/cloud-game/v3/pkg/logger"
@@ -12,8 +13,15 @@ type Download struct {
 	Address string
 }
 
+type Result struct {
+	Key        string
+	Filename   string
+	Err        error
+	StatusCode int
+}
+
 type Client interface {
-	Request(dest string, urls ...Download) ([]string, []string)
+	Request(ctx context.Context, dest string, urls ...Download) <-chan Result
 }
 
 type Downloader struct {
@@ -30,7 +38,7 @@ type Process func(string, []string, *logger.Logger) []string
 
 func NewDefaultDownloader(log *logger.Logger) Downloader {
 	return Downloader{
-		backend: NewGrabDownloader(log),
+		backend: NewFetcher(log),
 		pipe:    []Process{unpackDelete},
 		log:     log,
 	}
@@ -40,11 +48,24 @@ func NewDefaultDownloader(log *logger.Logger) Downloader {
 // put them into the destination folder.
 // It will return a partial or full list of downloaded files,
 // a list of processed files if some pipe processing functions are set.
-func (d *Downloader) Download(dest string, urls ...Download) ([]string, []string) {
-	files, fails := d.backend.Request(dest, urls...)
-	for _, op := range d.pipe {
-		files = op(dest, files, d.log)
+func (d *Downloader) Download(ctx context.Context, dest string, urls ...Download) ([]string, []string) {
+	var files, fails []string
+
+	for r := range d.backend.Request(ctx, dest, urls...) {
+		if r.Err != nil {
+			if r.StatusCode == 404 {
+				fails = append(fails, r.Key)
+			}
+			continue
+		}
+
+		processed := []string{r.Filename}
+		for _, op := range d.pipe {
+			processed = op(dest, processed, d.log)
+		}
+		files = append(files, processed...)
 	}
+
 	return files, fails
 }
 

pkg/worker/caged/libretro/manager/fetcher.go

@@ -0,0 +1,302 @@
+package manager
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/giongto35/cloud-game/v3/pkg/logger"
+)
+
+const (
+	maxRetries   = 3
+	retryBackoff = 500 * time.Millisecond
+)
+
+type byteRange struct{ start, end int64 }
+
+type Fetcher struct {
+	Client      *http.Client
+	MinPartSize int64
+	MaxParts    int
+	UserAgent   string
+	Parallelism int
+	log         *logger.Logger
+}
+
+func NewFetcher(log *logger.Logger) Fetcher {
+	return Fetcher{
+		Client: &http.Client{
+			Timeout: 10 * time.Second,
+			Transport: &http.Transport{
+				Proxy:           http.ProxyFromEnvironment,
+				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+			},
+		},
+		Parallelism: 3,
+		MinPartSize: 10 << 20,
+		MaxParts:    3,
+		UserAgent:   "Cloud-Game/3.1",
+		log:         log,
+	}
+}
+
+func (f Fetcher) Request(ctx context.Context, dest string, urls ...Download) <-chan Result {
+	ch := make(chan Result)
+	go func() {
+		defer close(ch)
+		var wg sync.WaitGroup
+		sema := make(chan struct{}, f.Parallelism)
+		for _, dl := range urls {
+			if ctx.Err() != nil {
+				break
+			}
+			wg.Go(func() {
+				select {
+				case sema <- none:
+				case <-ctx.Done():
+					return
+				}
+				defer func() { <-sema }()
+				path, status, err := f.download(ctx, dest, dl)
+				if err != nil {
+					if ctx.Err() != nil {
+						return
+					}
+					f.log.Error().Err(err).Msgf("download [%s] %s has failed", dl.Key, dl.Address)
+				} else {
+					f.log.Info().Msgf("Downloaded [%s] -> %s", dl.Key, path)
+				}
+				ch <- Result{Key: dl.Key, Filename: path, Err: err, StatusCode: status}
+			})
+		}
+		wg.Wait()
+	}()
+	return ch
+}
+
+func (f Fetcher) download(ctx context.Context, dest string, dl Download) (string, int, error) {
+	var path string
+	var status int
+	_, err := retry(ctx, f.log, dl.Key, func() (struct{}, error) {
+		p, s, err := f.try(ctx, dest, dl)
+		status = s
+		if err == nil {
+			path = p
+			return none, nil
+		}
+		if s == http.StatusNotFound || s == http.StatusGone {
+			return none, nonRetryable{err}
+		}
+		return none, err
+	})
+	if err != nil {
+		return "", status, err
+	}
+	return path, status, nil
+}
+
+func (f Fetcher) try(ctx context.Context, dest string, dl Download) (string, int, error) {
+	size, ranges, err := f.head(ctx, dl)
+	if err != nil {
+		return f.downloadSingle(ctx, dest, dl, -1)
+	}
+	outPath := filepath.Join(dest, filename(dl))
+	if size > 0 {
+		if fi, err := os.Stat(outPath); err == nil && fi.Size() == size {
+			return outPath, http.StatusOK, nil
+		}
+	}
+	if ranges && size > int64(f.MinPartSize) {
+		return f.downloadMulti(ctx, dest, dl, size)
+	}
+	return f.downloadSingle(ctx, dest, dl, size)
+}
+
+func (f Fetcher) head(ctx context.Context, dl Download) (size int64, ranges bool, err error) {
+	req, _ := http.NewRequestWithContext(ctx, http.MethodHead, dl.Address, nil)
+	req.Header.Set("User-Agent", "Cloud-Game/3.0")
+	resp, err := f.Client.Do(req)
+	if err != nil {
+		return 0, false, err
+	}
+	resp.Body.Close()
+	return resp.ContentLength, resp.Header.Get("Accept-Ranges") == "bytes", nil
+}
+
+func (f Fetcher) downloadSingle(ctx context.Context, dest string, dl Download, knownSize int64) (string, int, error) {
+	outPath := filepath.Join(dest, filename(dl))
+	var resumeAt int64
+	if fi, err := os.Stat(outPath); err == nil {
+		if knownSize > 0 && fi.Size() == knownSize {
+			return outPath, http.StatusOK, nil
+		}
+		resumeAt = fi.Size()
+	}
+
+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, dl.Address, nil)
+	req.Header.Set("User-Agent", f.UserAgent)
+	if resumeAt > 0 {
+		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", resumeAt))
+	}
+	resp, err := f.Client.Do(req)
+	if err != nil {
+		return "", 0, fmt.Errorf("http get: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
+		return "", resp.StatusCode, fmt.Errorf("HTTP %d", resp.StatusCode)
+	}
+
+	flag := os.O_CREATE | os.O_WRONLY
+	if resumeAt > 0 && resp.StatusCode == http.StatusPartialContent {
+		flag |= os.O_APPEND
+	} else {
+		flag |= os.O_TRUNC
+	}
+	out, err := os.OpenFile(outPath, flag, 0644)
+	if err != nil {
+		return "", resp.StatusCode, fmt.Errorf("open file: %w", err)
+	}
+	defer out.Close()
+
+	if _, err := io.Copy(out, resp.Body); err != nil {
+		return "", resp.StatusCode, fmt.Errorf("write file: %w", err)
+	}
+	return outPath, resp.StatusCode, nil
+}
+
+func (f Fetcher) downloadMulti(ctx context.Context, dest string, dl Download, size int64) (string, int, error) {
+	partSize := max(size/int64(f.MaxParts), f.MinPartSize)
+	var ranges []byteRange
+	for start := int64(0); start < size; start += partSize {
+		end := start + partSize - 1
+		if end >= size {
+			end = size - 1
+		}
+		ranges = append(ranges, byteRange{start, end})
+		if end == size-1 {
+			break
+		}
+	}
+	if len(ranges) <= 1 {
+		return f.downloadSingle(ctx, dest, dl, size)
+	}
+
+	parts := make([][]byte, len(ranges))
+	var (
+		wg       sync.WaitGroup
+		mu       sync.Mutex
+		firstErr error
+	)
+	for i, r := range ranges {
+		wg.Go(func() {
+			data, err := retry(ctx, f.log, "", func() ([]byte, error) {
+				return f.fetchRange(ctx, dl, r.start, r.end)
+			})
+			mu.Lock()
+			if err != nil {
+				if firstErr == nil {
+					firstErr = err
+				}
+			} else {
+				parts[i] = data
+			}
+			mu.Unlock()
+		})
+	}
+	wg.Wait()
+	if firstErr != nil {
+		return "", 0, fmt.Errorf("multipart: %w", firstErr)
+	}
+	if err := writeFile(dest, dl, parts); err != nil {
+		return "", 0, err
+	}
+	return filepath.Join(dest, filename(dl)), http.StatusOK, nil
+}
+
+func (f Fetcher) fetchRange(ctx context.Context, dl Download, start, end int64) ([]byte, error) {
+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, dl.Address, nil)
+	req.Header.Set("User-Agent", "Cloud-Game/3.0")
+	req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", start, end))
+	resp, err := f.Client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("http get: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusPartialContent && resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("HTTP %d", resp.StatusCode)
+	}
+	var buf bytes.Buffer
+	if _, err := io.Copy(&buf, resp.Body); err != nil {
+		return nil, fmt.Errorf("read: %w", err)
+	}
+	return buf.Bytes(), nil
+}
+
+func writeFile(dest string, dl Download, parts [][]byte) error {
+	out, err := os.Create(filepath.Join(dest, filename(dl)))
+	if err != nil {
+		return fmt.Errorf("create file: %w", err)
+	}
+	defer out.Close()
+	for _, p := range parts {
+		if _, err := out.Write(p); err != nil {
+			return fmt.Errorf("write: %w", err)
+		}
+	}
+	return nil
+}
+
+func filename(dl Download) string {
+	name := filepath.Base(dl.Address)
+	if name == "" || name == "." || name == "/" {
+		return dl.Key
+	}
+	return name
+}
+
+func retry[T any](ctx context.Context, log *logger.Logger, label string, fn func() (T, error)) (T, error) {
+	var (
+		lastErr error
+		zero    T
+	)
+	for attempt := range maxRetries {
+		if ctx.Err() != nil {
+			return zero, ctx.Err()
+		}
+		if attempt > 0 {
+			select {
+			case <-time.After(retryBackoff * time.Duration(attempt)):
+			case <-ctx.Done():
+				return zero, ctx.Err()
+			}
+		}
+		val, err := fn()
+		if err == nil {
+			return val, nil
+		}
+		if nr, ok := err.(nonRetryable); ok {
+			return zero, nr.err
+		}
+		lastErr = err
+		if label != "" {
+			log.Warn().Msgf("%s attempt %d/%d failed: %v", label, attempt+1, maxRetries, err)
+		}
+	}
+	return zero, fmt.Errorf("after %d retries: %w", maxRetries, lastErr)
+}
+
+var none struct{}
+
+type nonRetryable struct{ err error }
+
+func (nr nonRetryable) Error() string { return nr.err.Error() }

pkg/worker/caged/libretro/manager/grab.go

@@ -1,63 +0,0 @@
-package manager
-
-import (
-	"crypto/tls"
-	"net/http"
-
-	"github.com/cavaliergopher/grab/v3"
-	"github.com/giongto35/cloud-game/v3/pkg/logger"
-)
-
-type GrabDownloader struct {
-	client      *grab.Client
-	parallelism int
-	log         *logger.Logger
-}
-
-func NewGrabDownloader(log *logger.Logger) GrabDownloader {
-	return GrabDownloader{
-		client: &grab.Client{
-			UserAgent: "Cloud-Game/3.0",
-			HTTPClient: &http.Client{
-				Transport: &http.Transport{
-					Proxy:           http.ProxyFromEnvironment,
-					TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-				},
-			},
-		},
-		parallelism: 5,
-		log:         log,
-	}
-}
-
-func (d GrabDownloader) Request(dest string, urls ...Download) (ok []string, nook []string) {
-	reqs := make([]*grab.Request, 0)
-	for _, url := range urls {
-		req, err := grab.NewRequest(dest, url.Address)
-		if err != nil {
-			d.log.Error().Err(err).Msgf("couldn't make request URL: %v, %v", url, err)
-		} else {
-			req.Label = url.Key
-			reqs = append(reqs, req)
-		}
-	}
-
-	// check each response
-	for resp := range d.client.DoBatch(d.parallelism, reqs...) {
-		r := resp.Request
-		if err := resp.Err(); err != nil {
-			d.log.Error().Err(err).Msgf("download [%s] %s has failed: %v", r.Label, r.URL(), err)
-			if resp.HTTPResponse != nil && resp.HTTPResponse.StatusCode == 404 {
-				nook = append(nook, resp.Request.Label)
-			}
-		} else {
-			status := ""
-			if resp.HTTPResponse != nil {
-				status = resp.HTTPResponse.Status
-			}
-			d.log.Info().Msgf("Downloaded [%v] [%s] -> %s", status, r.Label, resp.Filename)
-			ok = append(ok, resp.Filename)
-		}
-	}
-	return
-}

pkg/worker/caged/libretro/manager/http.go

@@ -1,6 +1,8 @@
 package manager
 
 import (
+	"context"
+
 	"github.com/giongto35/cloud-game/v3/pkg/config"
 	"github.com/giongto35/cloud-game/v3/pkg/logger"
 	"github.com/giongto35/cloud-game/v3/pkg/os"
@@ -60,13 +62,13 @@ func CheckCores(conf config.Emulator, log *logger.Logger) error {
 	if err := os.MakeDirAll(coreManager.Conf.GetCoresStorePath()); err != nil {
 		return err
 	}
-	if err := coreManager.Sync(); err != nil {
+	if err := coreManager.Sync(context.Background()); err != nil {
 		return err
 	}
 	return nil
 }
 
-func (m *Manager) Sync() error {
+func (m *Manager) Sync(ctx context.Context) error {
 	// IPC lock if multiple worker processes on the same machine
 	err := m.fmu.Lock()
 	if err != nil {
@@ -84,7 +86,7 @@ func (m *Manager) Sync() error {
 		return err
 	}
 	download := diff(m.Conf.GetCores(), installed)
-	if failed := m.download(download); len(failed) > 0 {
+	if failed := m.download(ctx, download); len(failed) > 0 {
 		m.log.Warn().Msgf("[core-dl] error: unable to download these cores: %v", failed)
 	}
 	return nil
@@ -97,7 +99,7 @@ func (m *Manager) getCoreUrls(names []string, repo Repository) (urls []Download)
 	return
 }
 
-func (m *Manager) download(cores []config.CoreInfo) (failed []string) {
+func (m *Manager) download(ctx context.Context, cores []config.CoreInfo) (failed []string) {
 	if len(cores) == 0 || m.repo == nil {
 		return
 	}
@@ -120,26 +122,26 @@ func (m *Manager) download(cores []config.CoreInfo) (failed []string) {
 	}
 
 	m.log.Info().Msgf("[core-dl] <<< download | main: %v | alt: %v", prime, second)
-	primeFails := m.down(prime, m.repo)
+	primeFails := m.down(ctx, prime, m.repo)
 	if len(primeFails) > 0 && m.altRepo != nil {
 		m.log.Warn().Msgf("[core-dl] error: unable to download some cores, trying 2nd repository")
-		failed = append(failed, m.down(primeFails, m.altRepo)...)
+		failed = append(failed, m.down(ctx, primeFails, m.altRepo)...)
 	}
 	if m.altRepo != nil {
-		altFails := m.down(second, m.altRepo)
+		altFails := m.down(ctx, second, m.altRepo)
 		if len(altFails) > 0 {
 			m.log.Error().Msgf("[core-dl] error: unable to download some cores, trying 1st repository")
-			failed = append(failed, m.down(altFails, m.repo)...)
+			failed = append(failed, m.down(ctx, altFails, m.repo)...)
 		}
 	}
 	return
 }
 
-func (m *Manager) down(cores []string, repo Repository) (failed []string) {
+func (m *Manager) down(ctx context.Context, cores []string, repo Repository) (failed []string) {
 	if len(cores) == 0 || repo == nil {
 		return
 	}
-	_, failed = m.client.Download(m.Conf.GetCoresStorePath(), m.getCoreUrls(cores, repo)...)
+	_, failed = m.client.Download(ctx, m.Conf.GetCoresStorePath(), m.getCoreUrls(cores, repo)...)
 	return
 }
 

pkg/worker/caged/libretro/manager/repository.go

@@ -1,7 +1,5 @@
 package manager
 
-import "strings"
-
 type ArchInfo struct {
 	Arch   string
 	Ext    string
@@ -29,16 +27,15 @@ func (r Repo) CoreUrl(_ string, _ ArchInfo) string { return r.Address }
 type Buildbot struct{ Repo }
 
 func (r Buildbot) CoreUrl(file string, info ArchInfo) string {
-	var sb strings.Builder
-	sb.WriteString(r.Address + "/")
+	url := r.Address
 	if info.Vendor != "" {
-		sb.WriteString(info.Vendor + "/")
+		url += "/" + info.Vendor
 	}
-	sb.WriteString(info.Os + "/" + info.Arch + "/latest/" + file + info.Ext)
+	url += "/" + info.Os + "/" + info.Arch + "/latest/" + file + info.Ext
 	if r.Compression != "" {
-		sb.WriteString("." + r.Compression)
+		url += "." + r.Compression
 	}
-	return sb.String()
+	return url
 }
 
 type Github struct{ Buildbot }