From 93ef82083e27978c0ecd8757b53fe0209edd154a Mon Sep 17 00:00:00 2001 From: sergystepanov Date: Mon, 6 Jul 2026 09:44:45 +0300 Subject: [PATCH] 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 +- go.sum | 6 +- pkg/worker/caged/libretro/caged.go | 8 +- .../caged/libretro/manager/downloader.go | 33 +- pkg/worker/caged/libretro/manager/fetcher.go | 302 ++++++++++++++++++ pkg/worker/caged/libretro/manager/grab.go | 63 ---- pkg/worker/caged/libretro/manager/http.go | 22 +- .../caged/libretro/manager/repository.go | 13 +- 8 files changed, 354 insertions(+), 96 deletions(-) create mode 100644 pkg/worker/caged/libretro/manager/fetcher.go delete mode 100644 pkg/worker/caged/libretro/manager/grab.go diff --git a/go.mod b/go.mod index e37735c5..e81358d3 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 440d3e55..9dde1c8c 100644 --- a/go.sum +++ b/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= diff --git a/pkg/worker/caged/libretro/caged.go b/pkg/worker/caged/libretro/caged.go index f3553b1a..3a8d674b 100644 --- a/pkg/worker/caged/libretro/caged.go +++ b/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 { diff --git a/pkg/worker/caged/libretro/manager/downloader.go b/pkg/worker/caged/libretro/manager/downloader.go index d3e97409..a3239a63 100644 --- a/pkg/worker/caged/libretro/manager/downloader.go +++ b/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 } diff --git a/pkg/worker/caged/libretro/manager/fetcher.go b/pkg/worker/caged/libretro/manager/fetcher.go new file mode 100644 index 00000000..a0d9bbab --- /dev/null +++ b/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() } diff --git a/pkg/worker/caged/libretro/manager/grab.go b/pkg/worker/caged/libretro/manager/grab.go deleted file mode 100644 index f7d40a05..00000000 --- a/pkg/worker/caged/libretro/manager/grab.go +++ /dev/null @@ -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 -} diff --git a/pkg/worker/caged/libretro/manager/http.go b/pkg/worker/caged/libretro/manager/http.go index ff57a2de..fef26902 100644 --- a/pkg/worker/caged/libretro/manager/http.go +++ b/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 } diff --git a/pkg/worker/caged/libretro/manager/repository.go b/pkg/worker/caged/libretro/manager/repository.go index 3dbe0686..8b63d7aa 100644 --- a/pkg/worker/caged/libretro/manager/repository.go +++ b/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 }