Compare commits

..
24 changed files with 621 additions and 304 deletions

No files matched your search

-4
View File
@@ -46,10 +46,6 @@ type Uploader interface {
WaitUpload() error WaitUpload() error
} }
type DownloadCounter interface {
ReceivedBytes() int64
}
// FetcherMeta defines the meta information of a fetcher. // FetcherMeta defines the meta information of a fetcher.
type FetcherMeta struct { type FetcherMeta struct {
Req *base.Request `json:"req"` Req *base.Request `json:"req"`
-76
View File
@@ -290,82 +290,6 @@ func TestBTDownloadSelectedFiles(t *testing.T) {
if f.Meta().Res.Size != 48*1024 { if f.Meta().Res.Size != 48*1024 {
t.Fatal("selected resource size is incorrect") t.Fatal("selected resource size is incorrect")
} }
if downloaded := f.Progress().TotalDownloaded(); downloaded != 48*1024 || f.ReceivedBytes() <= downloaded {
t.Fatalf("selected progress = %d, received traffic = %d; want 48 KiB of progress and traffic across the shared piece", downloaded, f.ReceivedBytes())
}
}
func TestBTProgressOnlyCountsVerifiedFileBytes(t *testing.T) {
const fileSize = 56 * 1024
fixture := makeBTFixtureWithPieceSize(t, map[string][]byte{
"a.bin": btPayload(fileSize, "verified-a"),
"b.bin": btPayload(fileSize, "verified-b"),
}, 16*1024, false, true, 64*1024)
f := btDownloadFetcher(t, nil)
btResolve(t, f, fixture.Magnet)
dir := t.TempDir()
f.Meta().Opts = &base.Options{Path: dir}
btStart(t, f)
deadline := time.Now().Add(8 * time.Second)
for f.torrent.Files()[0].BytesCompleted() == 0 && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
btPause(t, f)
if f.torrent.Files()[0].BytesCompleted() == 0 || f.torrent.Piece(0).State().Complete {
t.Fatal("expected some received bytes in a piece that is still incomplete")
}
if f.ReceivedBytes() == 0 {
t.Fatal("received traffic must be visible before a piece is verified")
}
if progress := f.Progress(); len(progress) != 2 || progress[0] != 0 {
t.Errorf("progress before the first file's piece is verified = %v, want zero for the first file", progress)
}
fixture.Limiter.SetLimit(rate.Inf)
btStart(t, f)
btAwaitDone(t, f)
fixture.AssertFiles(t, dir)
progress := f.Progress()
if len(progress) != 2 || progress[0] != fileSize || progress[1] != fileSize {
t.Fatalf("verified progress at completion = %v, want %d for each file", progress, fileSize)
}
}
func TestBTUploadedBytesSamplesLiveTraffic(t *testing.T) {
payload := btPayload(48*1024, "upload-counter")
fixture := newBTFixture(t, map[string][]byte{"": payload}, 0)
const saved = 1024
f := &Fetcher{
torrent: fixture.Torrent,
data: &fetcherData{SeedBytes: saved},
seedBaseBytes: saved,
}
f.torrentReady.Store(true)
f.torrentUpload.Store(true)
cfg := torrent.TestingConfig(t)
cfg.DataDir = t.TempDir()
cfg.DisableTrackers = true
cfg.DisableUTP = true
cfg.DisableIPv6 = true
peer, err := torrent.NewClient(cfg)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { peer.Close() })
download, _, err := peer.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(fixture.MetaInfo))
if err != nil {
t.Fatal(err)
}
download.AddPeers([]torrent.PeerInfo{{Addr: fixture.Seeder.ListenAddrs()[0]}})
download.DownloadAll()
select {
case <-download.Complete().On():
case <-time.After(10 * time.Second):
t.Fatal("loopback peer did not complete")
}
if got, want := f.UploadedBytes(), int64(saved+len(payload)); got != want {
t.Fatalf("uploaded bytes = %d, want %d including the saved bytes", got, want)
}
} }
func TestBTPauseAndResume(t *testing.T) { func TestBTPauseAndResume(t *testing.T) {
+10 -63
View File
@@ -10,7 +10,6 @@ import (
"os" "os"
"path/filepath" "path/filepath"
"runtime" "runtime"
"sort"
"strings" "strings"
"sync" "sync"
"sync/atomic" "sync/atomic"
@@ -43,10 +42,9 @@ type Fetcher struct {
ctl *controller.Controller ctl *controller.Controller
config *config config *config
torrent *torrent.Torrent torrent *torrent.Torrent
meta *fetcher.FetcherMeta meta *fetcher.FetcherMeta
data *fetcherData data *fetcherData
seedBaseBytes int64
torrentReady atomic.Bool torrentReady atomic.Bool
torrentUpload atomic.Bool torrentUpload atomic.Bool
@@ -465,46 +463,14 @@ func (f *Fetcher) Progress() fetcher.Progress {
if !f.torrentReady.Load() { if !f.torrentReady.Load() {
return append(fetcher.Progress(nil), f.data.Progress...) return append(fetcher.Progress(nil), f.data.Progress...)
} }
pieceSize := f.torrent.Info().PieceLength
var offset int64
var verified []verifiedByteRange
for _, run := range f.torrent.PieceStateRuns() {
end := offset + int64(run.Length)*pieceSize
if pieceVerified(run.PieceState) {
verified = append(verified, verifiedByteRange{start: offset, end: end})
}
offset = end
}
files := f.torrent.Files()
for i := range f.data.Progress { for i := range f.data.Progress {
selectIndex := f.meta.Opts.SelectFiles[i] selectIndex := f.meta.Opts.SelectFiles[i]
file := files[selectIndex] file := f.torrent.Files()[selectIndex]
f.data.Progress[i] = verifiedFileBytes(verified, file.Offset(), file.Length()) f.data.Progress[i] = file.BytesCompleted()
} }
return append(fetcher.Progress(nil), f.data.Progress...) return append(fetcher.Progress(nil), f.data.Progress...)
} }
type verifiedByteRange struct {
start, end int64
}
func verifiedFileBytes(verified []verifiedByteRange, offset, length int64) int64 {
first := sort.Search(len(verified), func(i int) bool { return verified[i].end > offset })
fileEnd := offset + length
var completed int64
for _, run := range verified[first:] {
if run.start >= fileEnd {
break
}
completed += min(fileEnd, run.end) - max(offset, run.start)
}
return completed
}
func pieceVerified(state torrent.PieceState) bool {
return state.Ok && state.Complete && !state.Hashing && !state.QueuedForHash
}
func (f *Fetcher) Wait() error { func (f *Fetcher) Wait() error {
f.mu.Lock() f.mu.Lock()
var closed <-chan struct{} var closed <-chan struct{}
@@ -621,7 +587,7 @@ func (f *Fetcher) isDoneLocked() bool {
file := f.torrent.Files()[index] file := f.torrent.Files()[index]
for i := file.BeginPieceIndex(); i < file.EndPieceIndex(); i++ { for i := file.BeginPieceIndex(); i < file.EndPieceIndex(); i++ {
state := f.torrent.Piece(i).State() state := f.torrent.Piece(i).State()
if !pieceVerified(state) { if !state.Ok || !state.Complete || state.Hashing || state.QueuedForHash {
return false return false
} }
} }
@@ -719,13 +685,11 @@ func (f *Fetcher) Upload() (err error) {
} }
func (f *Fetcher) doUpload(fromUpload bool) { func (f *Fetcher) doUpload(fromUpload bool) {
f.mu.Lock()
if !f.torrentUpload.CompareAndSwap(false, true) { if !f.torrentUpload.CompareAndSwap(false, true) {
f.mu.Unlock()
return return
} }
f.seedBaseBytes = f.data.SeedBytes f.mu.Lock()
initialTime := f.data.SeedTime initialBytes, initialTime := f.data.SeedBytes, f.data.SeedTime
torrentClosed := f.torrent.Closed() torrentClosed := f.torrent.Closed()
f.mu.Unlock() f.mu.Unlock()
var doneTime int64 var doneTime int64
@@ -743,7 +707,8 @@ func (f *Fetcher) doUpload(fromUpload bool) {
f.mu.Unlock() f.mu.Unlock()
continue continue
} }
f.refreshSeedBytesLocked() stats := f.torrentStats()
f.data.SeedBytes = initialBytes + stats.BytesWrittenData.Int64()
if !fromUpload && !f.isDoneLocked() { if !fromUpload && !f.isDoneLocked() {
f.mu.Unlock() f.mu.Unlock()
continue continue
@@ -776,27 +741,9 @@ func (f *Fetcher) torrentStats() torrent.TorrentStats {
func (f *Fetcher) UploadedBytes() int64 { func (f *Fetcher) UploadedBytes() int64 {
f.mu.Lock() f.mu.Lock()
defer f.mu.Unlock() defer f.mu.Unlock()
if f.torrentReady.Load() && f.torrentUpload.Load() {
f.refreshSeedBytesLocked()
}
return f.data.SeedBytes return f.data.SeedBytes
} }
func (f *Fetcher) refreshSeedBytesLocked() {
stats := f.torrentStats()
f.data.SeedBytes = max(f.data.SeedBytes, f.seedBaseBytes+stats.BytesWrittenData.Int64())
}
func (f *Fetcher) ReceivedBytes() int64 {
f.mu.Lock()
defer f.mu.Unlock()
if !f.torrentReady.Load() {
return 0
}
stats := f.torrentStats()
return stats.BytesReadUsefulData.Int64()
}
func (f *Fetcher) WaitUpload() (err error) { func (f *Fetcher) WaitUpload() (err error) {
<-f.uploadDoneCh <-f.uploadDoneCh
return nil return nil
+1 -6
View File
@@ -46,11 +46,6 @@ func newBTHybridFixture(t testing.TB, payload []byte) *btFixture {
} }
func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool) *btFixture { func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool) *btFixture {
t.Helper()
return makeBTFixtureWithPieceSize(t, files, bytesPerSecond, v2, private, 16*1024)
}
func makeBTFixtureWithPieceSize(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool, pieceSize int64) *btFixture {
t.Helper() t.Helper()
source := t.TempDir() source := t.TempDir()
name := "payload.bin" name := "payload.bin"
@@ -69,7 +64,7 @@ func makeBTFixtureWithPieceSize(t testing.TB, files map[string][]byte, bytesPerS
} }
expected[filepath.Join(name, filepath.FromSlash(path))] = append([]byte(nil), data...) expected[filepath.Join(name, filepath.FromSlash(path))] = append([]byte(nil), data...)
} }
info := metainfo.Info{PieceLength: pieceSize} info := metainfo.Info{PieceLength: 16 * 1024}
if err := info.BuildFromFilePath(root); err != nil { if err := info.BuildFromFilePath(root); err != nil {
t.Fatal(err) t.Fatal(err)
} }
+12
View File
@@ -418,6 +418,16 @@ func (f *Fetcher) updateMaxConnTime(d time.Duration) {
} }
func (f *Fetcher) Resolve(req *base.Request) error { func (f *Fetcher) Resolve(req *base.Request) error {
return f.ResolveContext(context.Background(), req)
}
// ResolveContext cancels the initial request while waiting for headers. Once
// Resolve has returned, the retained response belongs to the fetcher and its
// existing Pause/Close lifecycle controls the body and prefetch.
func (f *Fetcher) ResolveContext(ctx context.Context, req *base.Request) error {
if err := ctx.Err(); err != nil {
return err
}
if err := base.ParseReqExtra[fhttp.ReqExtra](req); err != nil { if err := base.ParseReqExtra[fhttp.ReqExtra](req); err != nil {
return err return err
} }
@@ -429,6 +439,8 @@ func (f *Fetcher) Resolve(req *base.Request) error {
// This allows the response to be reused for downloading (important for one-time URLs) // This allows the response to be reused for downloading (important for one-time URLs)
// The body outlives Resolve, so the read timeout gets its own cancel func. // The body outlives Resolve, so the read timeout gets its own cancel func.
reqCtx, reqCancel := context.WithCancel(context.Background()) reqCtx, reqCancel := context.WithCancel(context.Background())
stopCancel := context.AfterFunc(ctx, reqCancel)
defer stopCancel()
httpReq, err := f.buildRequest(reqCtx, req) httpReq, err := f.buildRequest(reqCtx, req)
if err != nil { if err != nil {
reqCancel() reqCancel()
+36
View File
@@ -0,0 +1,36 @@
package api
import (
"net/http"
"path/filepath"
"testing"
"github.com/GopeedLab/gopeed/pkg/base"
"github.com/GopeedLab/gopeed/pkg/download"
restmodel "github.com/GopeedLab/gopeed/pkg/rest/model"
)
func TestCancelledTorrentPreviewCanBeResolvedAgain(t *testing.T) {
d := download.NewDownloader(&download.DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir()})
if err := d.Setup(); err != nil {
t.Fatal(err)
}
defer d.Close()
service, err := NewService(d)
if err != nil {
t.Fatal(err)
}
path, err := filepath.Abs("../../internal/protocol/bt/testdata/test.torrent")
if err != nil {
t.Fatal(err)
}
req := &restmodel.ResolveTask{Req: &base.Request{URL: path}}
first := serviceTestRequest[*download.ResolveResult](t, service, http.MethodPost, "/api/v1/resolve", req)
serviceTestRequest[any](t, service, http.MethodDelete, "/api/v1/resolve/"+first.ID, nil)
// Release is idempotent, including an old window disposing after Create.
serviceTestRequest[any](t, service, http.MethodDelete, "/api/v1/resolve/"+first.ID, nil)
second := serviceTestRequest[*download.ResolveResult](t, service, http.MethodPost, "/api/v1/resolve", req)
if second.Res.Hash != first.Res.Hash {
t.Fatalf("different torrent after release: %v", second)
}
}
+9 -1
View File
@@ -51,6 +51,7 @@ func NewService(downloader *download.Downloader) (*Service, error) {
service.routes = []RouteSpec{ service.routes = []RouteSpec{
newRoute(http.MethodGet, "/api/v1/info", service.info), newRoute(http.MethodGet, "/api/v1/info", service.info),
newRoute(http.MethodPost, "/api/v1/resolve", service.resolve), newRoute(http.MethodPost, "/api/v1/resolve", service.resolve),
newRoute(http.MethodDelete, "/api/v1/resolve/{id}", service.releaseResolve),
newRoute(http.MethodPost, "/api/v1/tasks", service.createTask), newRoute(http.MethodPost, "/api/v1/tasks", service.createTask),
newRoute(http.MethodPost, "/api/v1/tasks/batch", service.createTaskBatch), newRoute(http.MethodPost, "/api/v1/tasks/batch", service.createTaskBatch),
newRoute(http.MethodPatch, "/api/v1/tasks/{id}", service.patchTask), newRoute(http.MethodPatch, "/api/v1/tasks/{id}", service.patchTask),
@@ -143,13 +144,20 @@ func (s *Service) resolve(ctx *Context) *Response {
if response := decodeRequest(ctx.Body, &request); response != nil { if response := decodeRequest(ctx.Body, &request); response != nil {
return response return response
} }
result, err := s.Downloader.Resolve(request.Req) result, err := s.Downloader.ResolveWithID(request.ID, request.Req)
if err != nil { if err != nil {
return newErrorResponse(err.Error()) return newErrorResponse(err.Error())
} }
return newOKResponse(result) return newOKResponse(result)
} }
func (s *Service) releaseResolve(ctx *Context) *Response {
if err := s.Downloader.ReleaseResolve(ctx.PathParam("id")); err != nil {
return newErrorResponse(err.Error())
}
return newOKResponse(nil)
}
func (s *Service) createTask(ctx *Context) *Response { func (s *Service) createTask(ctx *Context) *Response {
var request restmodel.CreateTask var request restmodel.CreateTask
if response := decodeRequest(ctx.Body, &request); response != nil { if response := decodeRequest(ctx.Body, &request); response != nil {
+169 -47
View File
@@ -116,6 +116,7 @@ type Downloader struct {
cfg *DownloaderConfig cfg *DownloaderConfig
fetcherCache map[string]fetcher.Fetcher fetcherCache map[string]fetcher.Fetcher
resolveStates map[string]*resolveState
storage Storage storage Storage
tasks []*Task tasks []*Task
waitTasks []*Task waitTasks []*Task
@@ -153,6 +154,7 @@ func NewDownloader(cfg *DownloaderConfig) *Downloader {
cancelBackground: cancelBackground, cancelBackground: cancelBackground,
cfg: cfg, cfg: cfg,
fetcherCache: make(map[string]fetcher.Fetcher), fetcherCache: make(map[string]fetcher.Fetcher),
resolveStates: make(map[string]*resolveState),
waitTasks: make([]*Task, 0), waitTasks: make([]*Task, 0),
groups: make(map[string]*TaskGroup), groups: make(map[string]*TaskGroup),
creatingGroups: make(map[string]struct{}), creatingGroups: make(map[string]struct{}),
@@ -345,12 +347,7 @@ func (d *Downloader) Setup() error {
} }
current := task.fetcher.Progress().TotalDownloaded() current := task.fetcher.Progress().TotalDownloaded()
delta := max(0, current-task.Progress.Downloaded) delta := current - task.Progress.Downloaded
if counter, ok := task.fetcher.(fetcher.DownloadCounter); ok && task.Status == base.DownloadStatusRunning {
received := counter.ReceivedBytes()
delta = received - task.receivedBytes
task.receivedBytes = received
}
if task.Meta != nil && task.Meta.Req != nil { if task.Meta != nil && task.Meta.Req != nil {
if producer := d.blob.Production(task.Meta.Req.URL); producer != nil { if producer := d.blob.Production(task.Meta.Req.URL); producer != nil {
downloaded, received := producer.Progress() downloaded, received := producer.Progress()
@@ -358,20 +355,19 @@ func (d *Downloader) Setup() error {
task.producer = producer task.producer = producer
task.producerReceived = 0 task.producerReceived = 0
task.speedArr = nil task.speedArr = nil
task.downloadSampledAt = time.Time{}
} }
current = downloaded current = downloaded
delta = max(0, received-task.producerReceived) delta = max(0, received-task.producerReceived)
task.producerReceived = received task.producerReceived = received
} }
} }
tick := float64(d.cfg.RefreshInterval) / 1000
downloadDataChanged := false downloadDataChanged := false
uploading := task.Uploading uploading := task.Uploading
if task.Status == base.DownloadStatusRunning { if task.Status == base.DownloadStatusRunning {
downloadDataChanged = current != task.Progress.Downloaded downloadDataChanged = current != task.Progress.Downloaded
task.Progress.Used = task.timer.Used() task.Progress.Used = task.timer.Used()
elapsed := sampleSeconds(&task.downloadSampledAt, time.Now()) task.Progress.Speed = task.updateSpeed(max(0, delta), tick)
task.Progress.Speed = task.updateSpeed(delta, elapsed)
task.Progress.Downloaded = current task.Progress.Downloaded = current
} }
@@ -380,8 +376,7 @@ func (d *Downloader) Setup() error {
uploader := task.fetcher.(fetcher.Uploader) uploader := task.fetcher.(fetcher.Uploader)
currentUploaded := uploader.UploadedBytes() currentUploaded := uploader.UploadedBytes()
uploadDataChanged = currentUploaded != task.Progress.Uploaded uploadDataChanged = currentUploaded != task.Progress.Uploaded
elapsed := sampleSeconds(&task.uploadSampledAt, time.Now()) task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, tick)
task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, elapsed)
task.Progress.Uploaded = currentUploaded task.Progress.Uploaded = currentUploaded
} }
task.statusLock.Unlock() task.statusLock.Unlock()
@@ -566,42 +561,156 @@ func ensureResourceRequestRawURLs(parentReq *base.Request, res *base.Resource) {
} }
} }
func (d *Downloader) Resolve(req *base.Request) (rr *ResolveResult, err error) { // resolveState owns a Fetcher until Create transfers it to a task.
rrId, err := gonanoid.New() // State transitions are protected by fetcherMapLock; closeOnce serializes cleanup.
if err != nil { type contextResolver interface {
return ResolveContext(context.Context, *base.Request) error
} }
ensureRequestRawURL(req)
type resolveState struct {
ctx context.Context
cancel context.CancelFunc
fetcher fetcher.Fetcher
cancelled bool
resolved bool
claimed bool
closeOnce sync.Once
closeErr error
}
func (s *resolveState) close(f fetcher.Fetcher) error {
if f == nil {
return nil
}
s.closeOnce.Do(func() { s.closeErr = f.Close() })
return s.closeErr
}
func (d *Downloader) Resolve(req *base.Request) (*ResolveResult, error) {
return d.ResolveWithID("", req)
}
// ResolveWithID lets a preview cancel before metadata is available. Clients
// that do not supply an ID retain the server-generated resource ID behavior.
func (d *Downloader) ResolveWithID(id string, req *base.Request) (rr *ResolveResult, err error) {
if id == "" {
id, err = gonanoid.New()
if err != nil {
return nil, err
}
}
if req == nil {
return nil, errors.New("resolve request is required")
}
resolveCtx, cancel := context.WithCancel(d.backgroundCtx)
defer cancel()
state := &resolveState{ctx: resolveCtx, cancel: cancel}
d.fetcherMapLock.Lock()
if d.closed.Load() {
d.fetcherMapLock.Unlock()
return nil, errors.New("downloader is closed")
}
if d.resolveStates[id] != nil || d.fetcherCache[id] != nil {
d.fetcherMapLock.Unlock()
return nil, errors.New("resource id is already in use")
}
d.resolveStates[id] = state
d.fetcherMapLock.Unlock()
defer func() {
if err != nil || state.fetcher == nil {
d.fetcherMapLock.Lock()
if d.resolveStates[id] == state {
delete(d.resolveStates, id)
}
d.fetcherMapLock.Unlock()
_ = state.close(state.fetcher)
}
}()
ensureRequestRawURL(req)
res, err := d.triggerOnResolve(req) res, err := d.triggerOnResolve(req)
if err != nil { if err != nil {
return return nil, err
} }
if res != nil && len(res.Files) > 0 { if res != nil && len(res.Files) > 0 {
rr = &ResolveResult{ return &ResolveResult{Res: res, SuggestedDownloadDir: d.suggestedDownloadDir(res)}, nil
Res: res,
SuggestedDownloadDir: d.suggestedDownloadDir(res),
}
return
} }
f, err := d.buildFetcher(req.URL)
fetcher, err := d.buildFetcher(req.URL)
if err != nil { if err != nil {
return return nil, err
}
err = fetcher.Resolve(req)
if err != nil {
return
} }
d.fetcherMapLock.Lock() d.fetcherMapLock.Lock()
d.fetcherCache[rrId] = fetcher state.fetcher = f
cancelled := state.cancelled || d.closed.Load()
d.fetcherMapLock.Unlock() d.fetcherMapLock.Unlock()
rr = &ResolveResult{ if cancelled {
ID: rrId, return nil, errors.New("resource resolution cancelled")
Res: fetcher.Meta().Res, }
SuggestedDownloadDir: d.suggestedDownloadDir(fetcher.Meta().Res), if resolver, ok := f.(contextResolver); ok {
err = resolver.ResolveContext(state.ctx, req)
} else {
err = f.Resolve(req)
}
if err != nil {
return nil, err
}
d.fetcherMapLock.Lock()
defer d.fetcherMapLock.Unlock()
if state.cancelled || d.closed.Load() {
return nil, errors.New("resource resolution cancelled")
}
state.resolved = true
d.fetcherCache[id] = f
return &ResolveResult{ID: id, Res: f.Meta().Res, SuggestedDownloadDir: d.suggestedDownloadDir(f.Meta().Res)}, nil
}
// ReleaseResolve is idempotent. A resource already transferred to a task is
// never closed by a late preview cancellation.
func (d *Downloader) ReleaseResolve(id string) error {
d.fetcherMapLock.Lock()
state := d.resolveStates[id]
f := d.fetcherCache[id]
delete(d.fetcherCache, id)
delete(d.resolveStates, id)
if state != nil {
state.cancelled = true
state.cancel()
if state.claimed {
f = nil
} else {
f = state.fetcher
if _, cancellable := f.(contextResolver); cancellable && !state.resolved {
// Resolve owns unpublished response/prefetch state. Its context
// cancels the request; the returning resolver closes it once
// publication has stopped, including a racing response.
f = nil
}
}
}
d.fetcherMapLock.Unlock()
if state != nil {
return state.close(f)
}
if f != nil {
return f.Close()
}
return nil
}
func (d *Downloader) closeResolvedResources() {
d.fetcherMapLock.Lock()
ids := make(map[string]struct{}, len(d.resolveStates)+len(d.fetcherCache))
for id := range d.resolveStates {
ids[id] = struct{}{}
}
for id := range d.fetcherCache {
ids[id] = struct{}{}
}
d.fetcherMapLock.Unlock()
for id := range ids {
if err := d.ReleaseResolve(id); err != nil {
d.Logger.Warn().Err(err).Msg("close resolved resource failed")
}
} }
return
} }
func (d *Downloader) notifyRunning() { func (d *Downloader) notifyRunning() {
@@ -875,18 +984,34 @@ func (d *Downloader) CreateWithOptions(rrId string, opts *base.Options) (string,
func (d *Downloader) createResolved(rrId string, opts *base.Options, groupID string) (taskId string, err error) { func (d *Downloader) createResolved(rrId string, opts *base.Options, groupID string) (taskId string, err error) {
d.fetcherMapLock.Lock() d.fetcherMapLock.Lock()
fetcher, ok := d.fetcherCache[rrId] fetcher, ok := d.fetcherCache[rrId]
state := d.resolveStates[rrId]
if ok && state != nil {
state.claimed = true
}
delete(d.fetcherCache, rrId) delete(d.fetcherCache, rrId)
d.fetcherMapLock.Unlock() d.fetcherMapLock.Unlock()
if !ok { if !ok {
return "", errors.New("invalid resource id") return "", errors.New("invalid resource id")
} }
// Only one creator may use a resolved fetcher. Failed creation makes it // Failed creation returns ownership to the preview unless it was cancelled.
// available again; a persisted task consumes it even if starting fails.
defer func() { defer func() {
if taskId == "" { d.fetcherMapLock.Lock()
d.fetcherMapLock.Lock() cancelled := d.closed.Load() || (state != nil && state.cancelled)
if taskId == "" && !cancelled {
d.fetcherCache[rrId] = fetcher d.fetcherCache[rrId] = fetcher
d.fetcherMapLock.Unlock() if state != nil {
state.claimed = false
}
} else if d.resolveStates[rrId] == state {
delete(d.resolveStates, rrId)
}
d.fetcherMapLock.Unlock()
if taskId == "" && cancelled {
if state != nil {
_ = state.close(fetcher)
} else {
_ = fetcher.Close()
}
} }
}() }()
if opts == nil { if opts == nil {
@@ -1569,6 +1694,7 @@ func (d *Downloader) Close() error {
return nil return nil
} }
d.cancelBackground() d.cancelBackground()
d.closeResolvedResources()
d.waitForExtensionGit() d.waitForExtensionGit()
if d.protocolJobs != nil { if d.protocolJobs != nil {
d.protocolJobs.close() d.protocolJobs.close()
@@ -2460,8 +2586,6 @@ func (d *Downloader) doStart(task *Task) (err error) {
return nil return nil
} }
task.Progress.Speed = 0 task.Progress.Speed = 0
task.speedArr = nil
task.downloadSampledAt = time.Time{}
task.timer.Start() task.timer.Start()
task.statusLock.Unlock() task.statusLock.Unlock()
if err := task.fetcher.Start(); err != nil { if err := task.fetcher.Start(); err != nil {
@@ -3070,10 +3194,8 @@ func initTask(task *Task) {
task.lock = &sync.Mutex{} task.lock = &sync.Mutex{}
task.statsLock = &sync.Mutex{} task.statsLock = &sync.Mutex{}
task.blobRefLock = &sync.Mutex{} task.blobRefLock = &sync.Mutex{}
task.speedArr = nil task.speedArr = make([]int64, 0)
task.uploadSpeedArr = nil task.uploadSpeedArr = make([]int64, 0)
task.downloadSampledAt = time.Time{}
task.uploadSampledAt = time.Time{}
} }
func (d *Downloader) releaseBlobTask(task *Task) { func (d *Downloader) releaseBlobTask(task *Task) {
+26 -43
View File
@@ -75,23 +75,20 @@ type Task struct {
producer production.Source producer production.Source
producerReceived int64 producerReceived int64
fetcherManager fetcher.FetcherManager fetcherManager fetcher.FetcherManager
fetcher fetcher.Fetcher fetcher fetcher.Fetcher
timer *util.Timer timer *util.Timer
statusLock *sync.Mutex statusLock *sync.Mutex
lock *sync.Mutex lock *sync.Mutex
statsLock *sync.Mutex statsLock *sync.Mutex
statsLoaded bool statsLoaded bool
stats *fetcher.Stats stats *fetcher.Stats
statsSnapshot json.RawMessage statsSnapshot json.RawMessage
blobRefLock *sync.Mutex blobRefLock *sync.Mutex
blobURL string blobURL string
runGeneration uint64 runGeneration uint64
speedArr []speedSample speedArr []int64
uploadSpeedArr []speedSample uploadSpeedArr []int64
downloadSampledAt time.Time
uploadSampledAt time.Time
receivedBytes int64
} }
// TaskRuntimeStatus is the lightweight, frequently refreshed view of a task. // TaskRuntimeStatus is the lightweight, frequently refreshed view of a task.
@@ -198,21 +195,7 @@ func (t *Task) updateUploadSpeed(downloaded int64, usedTime float64) int64 {
return calcSpeed(&t.uploadSpeedArr, downloaded, usedTime) return calcSpeed(&t.uploadSpeedArr, downloaded, usedTime)
} }
type speedSample struct { func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
bytes int64
seconds float64
}
func sampleSeconds(lastSample *time.Time, now time.Time) float64 {
previous := *lastSample
*lastSample = now
if previous.IsZero() {
return 0
}
return now.Sub(previous).Seconds()
}
func calcSpeed(speedArr *[]speedSample, downloaded int64, usedTime float64) int64 {
if usedTime <= 0 { if usedTime <= 0 {
return 0 return 0
} }
@@ -221,18 +204,18 @@ func calcSpeed(speedArr *[]speedSample, downloaded int64, usedTime float64) int6
return 0 return 0
} }
*speedArr = append(*speedArr, speedSample{bytes: downloaded, seconds: usedTime}) *speedArr = append(*speedArr, downloaded)
var total int64 // Record last 5 seconds of download speed to calculate the average speed
var elapsed float64 if len(*speedArr) > int(5.0/usedTime) {
first := len(*speedArr) *speedArr = (*speedArr)[1:]
for first > 0 && elapsed < 5 {
first--
sample := (*speedArr)[first]
total += sample.bytes
elapsed += sample.seconds
} }
*speedArr = (*speedArr)[first:]
return int64(float64(total) / elapsed) var total int64
for _, v := range *speedArr {
total += v
}
return int64(float64(total) / float64(len(*speedArr)) / usedTime)
} }
type TaskFilter struct { type TaskFilter struct {
+2 -39
View File
@@ -1,32 +1,9 @@
package download package download
import ( import "testing"
"testing"
"time"
)
func TestTransferSpeedWithUnevenSamples(t *testing.T) {
var task Task
for _, sample := range []struct {
bytes int64
seconds float64
}{
{1000, 1},
{2000, 2},
{4000, 4},
{6000, 6},
} {
if got := task.updateSpeed(sample.bytes, sample.seconds); got != 1000 {
t.Errorf("download speed after %v seconds = %d, want 1000", sample.seconds, got)
}
if got := task.updateUploadSpeed(sample.bytes, sample.seconds); got != 1000 {
t.Errorf("upload speed after %v seconds = %d, want 1000", sample.seconds, got)
}
}
}
func TestCalcSpeedResetOnRollback(t *testing.T) { func TestCalcSpeedResetOnRollback(t *testing.T) {
speedArr := []speedSample{{1024, 1}, {2048, 1}, {4096, 1}} speedArr := []int64{1024, 2048, 4096}
if got := calcSpeed(&speedArr, -512, 1); got != 0 { if got := calcSpeed(&speedArr, -512, 1); got != 0 {
t.Fatalf("calcSpeed() = %d, want 0 after rollback", got) t.Fatalf("calcSpeed() = %d, want 0 after rollback", got)
@@ -39,17 +16,3 @@ func TestCalcSpeedResetOnRollback(t *testing.T) {
t.Fatalf("calcSpeed() = %d, want 1024 after reset", got) t.Fatalf("calcSpeed() = %d, want 1024 after reset", got)
} }
} }
func TestSampleSecondsUsesThePreviousSuccessfulSample(t *testing.T) {
start := time.Now()
var last time.Time
if got := sampleSeconds(&last, start); got != 0 {
t.Fatalf("first sample elapsed = %v, want 0", got)
}
if got := sampleSeconds(&last, start.Add(1500*time.Millisecond)); got != 1.5 {
t.Fatalf("elapsed after a delayed sample = %v, want 1.5", got)
}
if got := sampleSeconds(&last, start.Add(2*time.Second)); got != 0.5 {
t.Fatalf("elapsed after the next sample = %v, want 0.5", got)
}
}
+224
View File
@@ -0,0 +1,224 @@
package download
import (
"errors"
"fmt"
"github.com/GopeedLab/gopeed/internal/fetcher"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/GopeedLab/gopeed/pkg/base"
)
func TestReleaseResolveCancelsPendingHTTPHeaders(t *testing.T) {
entered := make(chan struct{})
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { close(entered); <-r.Context().Done() }))
defer server.Close()
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir()})
if err := d.Setup(); err != nil {
t.Fatal(err)
}
defer d.Close()
done := make(chan error, 1)
go func() {
_, err := d.ResolveWithID("pending-http", &base.Request{URL: server.URL + "/file"})
done <- err
}()
select {
case <-entered:
case <-time.After(time.Second):
t.Fatal("request did not start")
}
if err := d.ReleaseResolve("pending-http"); err != nil {
t.Fatal(err)
}
select {
case err := <-done:
if err == nil {
t.Fatal("cancelled Resolve succeeded")
}
case <-time.After(300 * time.Millisecond):
server.CloseClientConnections()
t.Fatal("ReleaseResolve did not cancel the request waiting for headers")
}
if len(d.fetcherCache) != 0 || len(d.resolveStates) != 0 {
t.Fatal("cancelled operation retained resources")
}
}
func TestReleaseResolveAfterCreateDoesNotCloseTask(t *testing.T) {
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: nil})
if err := d.Setup(); err != nil {
t.Fatal(err)
}
defer d.Close()
d.cfg.MaxRunning = 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) }))
defer server.Close()
rr, err := d.Resolve(&base.Request{URL: server.URL})
if err != nil {
t.Fatal(err)
}
id, err := d.CreateWithOptions(rr.ID, &base.Options{Path: t.TempDir()})
if err != nil {
t.Fatal(err)
}
if err := d.ReleaseResolve(rr.ID); err != nil {
t.Fatal(err)
}
task := d.GetTask(id)
if task == nil {
t.Fatal("late release removed task")
}
if err := d.doStart(task); err != nil {
t.Fatal(err)
}
waitForTaskStatus(t, d, id, base.DownloadStatusDone, 3*time.Second)
}
func TestResolveRejectsClosedDownloader(t *testing.T) {
d := NewDownloader(&DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir()})
if err := d.Setup(); err != nil {
t.Fatal(err)
}
d.Close()
if _, err := d.ResolveWithID("closed", &base.Request{URL: "http://127.0.0.1/"}); err == nil {
t.Fatalf("closed downloader accepted Resolve: %v", err)
}
}
func TestReleaseResolveCancelsPendingMagnetMetadata(t *testing.T) {
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir()})
if err := d.Setup(); err != nil {
t.Fatal(err)
}
defer d.Close()
done := make(chan error, 1)
go func() {
_, err := d.ResolveWithID("pending-magnet", &base.Request{URL: "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567"})
done <- err
}()
deadline := time.Now().Add(2 * time.Second)
for {
d.fetcherMapLock.Lock()
state := d.resolveStates["pending-magnet"]
ready := state != nil && state.fetcher != nil
d.fetcherMapLock.Unlock()
if ready {
break
}
if time.Now().After(deadline) {
t.Fatal("magnet resolve did not start")
}
time.Sleep(time.Millisecond)
}
if err := d.ReleaseResolve("pending-magnet"); err != nil {
t.Fatal(err)
}
select {
case err := <-done:
if err == nil {
t.Fatal("cancelled magnet Resolve succeeded")
}
case <-time.After(time.Second):
t.Fatal("cancelled magnet remained in metadata wait")
}
if len(d.fetcherCache) != 0 || len(d.resolveStates) != 0 {
t.Fatal("cancelled magnet retained its resource")
}
}
func TestCloseDrainsResolvedPreviewCache(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) }))
defer server.Close()
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir()})
if err := d.Setup(); err != nil {
t.Fatal(err)
}
if _, err := d.Resolve(&base.Request{URL: server.URL}); err != nil {
t.Fatal(err)
}
if err := d.Close(); err != nil {
t.Fatal(err)
}
if len(d.fetcherCache) != 0 || len(d.resolveStates) != 0 {
t.Fatal("shutdown retained preview resources")
}
}
type cancelledCreationFetcher struct {
fetcher.Fetcher
entered chan struct{}
release chan struct{}
err error
}
func (f *cancelledCreationFetcher) Patch(*base.Request, *base.Options) error {
close(f.entered)
<-f.release
return f.err
}
func TestReleaseResolveDuringCreationTransfersOrClosesOwnership(t *testing.T) {
for _, fail := range []bool{false, true} {
t.Run(fmt.Sprintf("fail=%v", fail), func(t *testing.T) {
d := NewDownloader(&DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir()})
if err := d.Setup(); err != nil {
t.Fatal(err)
}
defer d.Close()
d.cfg.MaxRunning = 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) }))
defer server.Close()
rr, err := d.Resolve(&base.Request{URL: server.URL})
if err != nil {
t.Fatal(err)
}
f := &cancelledCreationFetcher{Fetcher: d.fetcherCache[rr.ID], entered: make(chan struct{}), release: make(chan struct{})}
if fail {
f.err = errors.New("creation failed")
}
d.fetcherMapLock.Lock()
d.fetcherCache[rr.ID] = f
d.fetcherMapLock.Unlock()
done := make(chan struct{})
var id string
var createErr error
go func() {
defer close(done)
id, createErr = d.CreateWithOptions(rr.ID, &base.Options{Path: t.TempDir(), Size: 2})
}()
select {
case <-f.entered:
case <-time.After(time.Second):
t.Fatal("creation did not claim resource")
}
if err := d.ReleaseResolve(rr.ID); err != nil {
t.Fatal(err)
}
close(f.release)
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("creation did not return")
}
if len(d.fetcherCache) != 0 || len(d.resolveStates) != 0 {
t.Fatal("cancelled creation returned ownership to preview cache")
}
if fail {
if createErr == nil {
t.Fatal("expected creation failure")
}
return
}
if createErr != nil {
t.Fatal(createErr)
}
// The protocol manager persists its concrete Fetcher type.
task := d.GetTask(id)
task.fetcher = f.Fetcher
if err := d.doStart(task); err != nil {
t.Fatal(err)
}
waitForTaskStatus(t, d, id, base.DownloadStatusDone, 3*time.Second)
})
}
}
+1 -1
View File
@@ -24,7 +24,7 @@ func Info(w http.ResponseWriter, r *http.Request) {
func Resolve(w http.ResponseWriter, r *http.Request) { func Resolve(w http.ResponseWriter, r *http.Request) {
var req model.ResolveTask var req model.ResolveTask
if ReadJson(r, w, &req) { if ReadJson(r, w, &req) {
rr, err := Downloader.Resolve(req.Req) rr, err := Downloader.ResolveWithID(req.ID, req.Req)
if err != nil { if err != nil {
WriteJson(w, model.NewErrorResult(err.Error())) WriteJson(w, model.NewErrorResult(err.Error()))
return return
+1
View File
@@ -3,6 +3,7 @@ package model
import "github.com/GopeedLab/gopeed/pkg/base" import "github.com/GopeedLab/gopeed/pkg/base"
type ResolveTask struct { type ResolveTask struct {
ID string `json:"id,omitempty"`
Req *base.Request `json:"req"` Req *base.Request `json:"req"`
} }
+3
View File
@@ -81,6 +81,9 @@ Future<ResolveResult> resolve(ResolveTask resolveTask) {
); );
} }
Future<void> releaseResolve(String id) =>
_parse<void>(() => _request('api/v1/resolve/${Uri.encodeComponent(id)}', method: 'DELETE'), null);
Future<String> createTask(CreateTask createTask) { Future<String> createTask(CreateTask createTask) {
return _parse<String>(() => _request('api/v1/tasks', method: 'POST', data: createTask), (data) => data as String); return _parse<String>(() => _request('api/v1/tasks', method: 'POST', data: createTask), (data) => data as String);
} }
+4 -1
View File
@@ -8,7 +8,10 @@ part 'resolve_task.g.dart';
class ResolveTask { class ResolveTask {
Request? req; Request? req;
ResolveTask({this.req}); @JsonKey(includeIfNull: false)
String? id;
ResolveTask({this.req, this.id});
factory ResolveTask.fromJson(Map<String, dynamic> json) => _$ResolveTaskFromJson(json); factory ResolveTask.fromJson(Map<String, dynamic> json) => _$ResolveTaskFromJson(json);
Map<String, dynamic> toJson() => _$ResolveTaskToJson(this); Map<String, dynamic> toJson() => _$ResolveTaskToJson(this);
+2 -1
View File
@@ -10,7 +10,8 @@ ResolveTask _$ResolveTaskFromJson(Map<String, dynamic> json) => ResolveTask(
req: json['req'] == null req: json['req'] == null
? null ? null
: Request.fromJson(json['req'] as Map<String, dynamic>), : Request.fromJson(json['req'] as Map<String, dynamic>),
id: json['id'] as String?,
); );
Map<String, dynamic> _$ResolveTaskToJson(ResolveTask instance) => Map<String, dynamic> _$ResolveTaskToJson(ResolveTask instance) =>
<String, dynamic>{'req': ?instance.req}; <String, dynamic>{'req': ?instance.req, 'id': ?instance.id};
@@ -53,7 +53,6 @@ class AppRuntimeState {
class AppRuntimeController extends AsyncNotifier<AppRuntimeState> { class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
static StartConfig? _defaultStartConfig; static StartConfig? _defaultStartConfig;
Future<void> _preferenceWrites = Future<void>.value(); Future<void> _preferenceWrites = Future<void>.value();
bool _analyticsStarted = false;
@override @override
Future<AppRuntimeState> build() async { Future<AppRuntimeState> build() async {
@@ -74,7 +73,6 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
startConfig startConfig
..apiEnable = true ..apiEnable = true
..mcpEnable = config.api.mcpEnable; ..mcpEnable = config.api.mcpEnable;
await _initAnalytics(config);
} else { } else {
startConfig startConfig
..apiEnable = config.api.enable ..apiEnable = config.api.enable
@@ -255,8 +253,7 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
} }
Future<void> _initAnalytics(DownloaderConfig config) async { Future<void> _initAnalytics(DownloaderConfig config) async {
if (_analyticsStarted || !AnalyticsConfig.isConfigured || !config.extra.analyticsEnabled) return; if (!AnalyticsConfig.isConfigured || !config.extra.analyticsEnabled) return;
_analyticsStarted = true;
try { try {
final generatedClientId = await Analytics.instance.init(fallbackClientId: config.extra.deviceId); final generatedClientId = await Analytics.instance.init(fallbackClientId: config.extra.deviceId);
if (generatedClientId != null) { if (generatedClientId != null) {
@@ -265,7 +262,6 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
} }
unawaited(Analytics.instance.logAppOpen()); unawaited(Analytics.instance.logAppOpen());
} catch (error, stackTrace) { } catch (error, stackTrace) {
_analyticsStarted = false;
logger.w('GA4 init failed', error, stackTrace); logger.w('GA4 init failed', error, stackTrace);
} }
} }
@@ -19,7 +19,7 @@ Future<DownloaderConfig> loadDownloaderConfig({
config = await load(); config = await load();
} catch (error, stackTrace) { } catch (error, stackTrace) {
onLoadError?.call(error, stackTrace); onLoadError?.call(error, stackTrace);
config = DownloaderConfig()..extra.analyticsEnabled = false; config = DownloaderConfig();
loaded = false; loaded = false;
} }
final changed = await initDownloaderConfig(config, defaultDownloadDir: defaultDownloadDir); final changed = await initDownloaderConfig(config, defaultDownloadDir: defaultDownloadDir);
@@ -93,6 +93,10 @@ void _bindGopeed(CapabilityRegistry registry, _ConfigStore configStore) {
(p) => api.getMirrorCandidates(p['provider'] as String, p['url'] as String), (p) => api.getMirrorCandidates(p['provider'] as String, p['url'] as String),
) )
..bind(GopeedMethods.resolve, api.resolve) ..bind(GopeedMethods.resolve, api.resolve)
..bind(GopeedMethods.releaseResolve, (id) async {
await api.releaseResolve(id);
return const RpcUnit();
})
..bind(GopeedMethods.createTask, api.createTask) ..bind(GopeedMethods.createTask, api.createTask)
..bind(GopeedMethods.createTaskBatch, api.createTaskBatch) ..bind(GopeedMethods.createTaskBatch, api.createTaskBatch)
..bind(GopeedMethods.patchTask, (params) async { ..bind(GopeedMethods.patchTask, (params) async {
@@ -16,6 +16,7 @@ abstract final class GopeedMethods {
static const triggerProtocolJob = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.protocol.job.trigger'); static const triggerProtocolJob = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.protocol.job.trigger');
static const getMirrorCandidates = RpcMethod<Map<String, dynamic>, List<String>>('gopeed.mirrors.candidates'); static const getMirrorCandidates = RpcMethod<Map<String, dynamic>, List<String>>('gopeed.mirrors.candidates');
static const resolve = RpcMethod<ResolveTask, ResolveResult>('gopeed.resolve'); static const resolve = RpcMethod<ResolveTask, ResolveResult>('gopeed.resolve');
static const releaseResolve = RpcMethod<String, RpcUnit>('gopeed.releaseResolve');
static const createTask = RpcMethod<CreateTask, String>('gopeed.task.create'); static const createTask = RpcMethod<CreateTask, String>('gopeed.task.create');
static const createTaskBatch = RpcMethod<CreateTaskBatch, CreateTaskBatchResult>('gopeed.task.createBatch'); static const createTaskBatch = RpcMethod<CreateTaskBatch, CreateTaskBatchResult>('gopeed.task.createBatch');
static const patchTask = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.task.patch'); static const patchTask = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.task.patch');
@@ -50,6 +51,7 @@ class GopeedService {
Future<List<String>> getMirrorCandidates(String provider, String url) => Future<List<String>> getMirrorCandidates(String provider, String url) =>
_invoker.invoke(GopeedMethods.getMirrorCandidates, {'provider': provider, 'url': url}); _invoker.invoke(GopeedMethods.getMirrorCandidates, {'provider': provider, 'url': url});
Future<ResolveResult> resolve(ResolveTask request) => _invoker.invoke(GopeedMethods.resolve, request); Future<ResolveResult> resolve(ResolveTask request) => _invoker.invoke(GopeedMethods.resolve, request);
Future<void> releaseResolve(String id) async => _invoker.invoke(GopeedMethods.releaseResolve, id);
Future<String> createTask(CreateTask request) => _invoker.invoke(GopeedMethods.createTask, request); Future<String> createTask(CreateTask request) => _invoker.invoke(GopeedMethods.createTask, request);
@@ -1,3 +1,5 @@
import 'dart:developer' as developer;
import 'package:uuid/uuid.dart';
import 'package:path/path.dart' as path; import 'package:path/path.dart' as path;
import '../../../api/model/create_task_batch.dart'; import '../../../api/model/create_task_batch.dart';
@@ -16,6 +18,8 @@ class PendingDownload {
final Request request; final Request request;
final Options options; final Options options;
ResolveResult? result; ResolveResult? result;
String? _resolveID;
bool _cancelled = false;
String? resolveError; String? resolveError;
String? createError; String? createError;
final Set<int> selected = {}; final Set<int> selected = {};
@@ -50,11 +54,16 @@ class PendingDownload {
bool get hasSelection => !created && result != null && selected.isNotEmpty; bool get hasSelection => !created && result != null && selected.isNotEmpty;
Future<void> resolve(GopeedService service) async { Future<void> resolve(GopeedService service) async {
if (result != null || resolving) return; if (_cancelled || result != null || resolving) return;
resolving = true; resolving = true;
resolveError = null; resolveError = null;
try { try {
result = await service.resolve(ResolveTask(req: request)); _resolveID = const Uuid().v4();
result = await service.resolve(ResolveTask(req: request, id: _resolveID));
if (_cancelled) {
await _release(service, result!.id);
return;
}
selected.addAll(List.generate(files.isEmpty ? 1 : files.length, (index) => index)); selected.addAll(List.generate(files.isEmpty ? 1 : files.length, (index) => index));
} catch (error) { } catch (error) {
resolveError = error.toString(); resolveError = error.toString();
@@ -63,6 +72,23 @@ class PendingDownload {
} }
} }
/// Stop an outstanding preview and release any resource it already owns.
/// A response arriving after cancellation is released by resolve() as well.
Future<void> cancel(GopeedService service) async {
if (_cancelled || created) return;
_cancelled = true;
await _release(service, result?.id.isNotEmpty == true ? result!.id : _resolveID);
}
Future<void> _release(GopeedService service, String? id) async {
if (id == null || id.isEmpty) return;
try {
await service.releaseResolve(id);
} catch (error, stack) {
developer.log('Failed to release resolved resource', name: 'gopeed.resolve', error: error, stackTrace: stack);
}
}
List<CreateTaskBatchItem> creationItems() { List<CreateTaskBatchItem> creationItems() {
final indexes = selected.toList()..sort(); final indexes = selected.toList()..sort();
if (result!.id.isNotEmpty || files.isEmpty) { if (result!.id.isNotEmpty || files.isEmpty) {
@@ -1,3 +1,5 @@
import 'dart:async';
import 'package:shadcn_flutter/shadcn_flutter.dart'; import 'package:shadcn_flutter/shadcn_flutter.dart';
import 'package:flutter/widgets.dart' as flutter show Flexible; import 'package:flutter/widgets.dart' as flutter show Flexible;
@@ -49,6 +51,7 @@ class _ResolveDownloadsDialogState extends State<_ResolveDownloadsDialog> {
@override @override
void dispose() { void dispose() {
_cancelled = true; _cancelled = true;
_releasePreviews();
super.dispose(); super.dispose();
} }
@@ -64,8 +67,15 @@ class _ResolveDownloadsDialogState extends State<_ResolveDownloadsDialog> {
} }
} }
void _releasePreviews() {
for (final source in widget.sources) {
unawaited(source.cancel(widget.service));
}
}
void _cancel() { void _cancel() {
_cancelled = true; _cancelled = true;
_releasePreviews();
closeOverlay(context, false); closeOverlay(context, false);
} }
@@ -32,22 +32,8 @@ void main() {
// The in-memory fallback still gets usable defaults. // The in-memory fallback still gets usable defaults.
expect(config.downloadDir, '/downloads'); expect(config.downloadDir, '/downloads');
expect(config.categories, hasLength(4)); expect(config.categories, hasLength(4));
expect(config.extra.analyticsEnabled, isFalse);
}); });
for (final analyticsEnabled in [false, true]) {
test('preserves the saved analytics preference ($analyticsEnabled)', () async {
final stored = configWith()..extra.analyticsEnabled = analyticsEnabled;
final config = await loadDownloaderConfig(
load: () async => stored,
save: (_) async {},
defaultDownloadDir: () async => '/downloads',
);
expect(config.extra.analyticsEnabled, analyticsEnabled);
});
}
test('skips persist when init changed nothing', () async { test('skips persist when init changed nothing', () async {
var saved = 0; var saved = 0;
await loadDownloaderConfig( await loadDownloaderConfig(
@@ -64,6 +64,81 @@ Future<void> open(
} }
void main() { void main() {
testWidgets('cancel releases a ready preview resource', (tester) async {
final released = <String>[];
final registry = CapabilityRegistry(createAppCapabilityCodecs())
..bind(GopeedMethods.resolve, (_) => resolved('ready-rid'))
..bind(const RpcMethod<String, RpcUnit>('gopeed.releaseResolve'), (id) {
released.add(id);
return const RpcUnit();
});
await open(tester, [pending('torrent')], registry);
await tester.pumpAndSettle();
await tester.tap(find.byKey(const ValueKey('resolve-cancel-button')));
await tester.pumpAndSettle();
expect(released, contains('ready-rid'));
});
testWidgets('cancel releases a late preview result', (tester) async {
final response = Completer<ResolveResult>();
final released = <String>[];
final registry = CapabilityRegistry(createAppCapabilityCodecs())
..bind(GopeedMethods.resolve, (_) => response.future)
..bind(const RpcMethod<String, RpcUnit>('gopeed.releaseResolve'), (id) {
released.add(id);
return const RpcUnit();
});
await open(tester, [pending('torrent')], registry);
await tester.tap(find.byKey(const ValueKey('resolve-close-button')));
await tester.pumpAndSettle();
response.complete(resolved('late-rid'));
await tester.pumpAndSettle();
expect(released, contains('late-rid'));
});
testWidgets('cancel sends the request ID before a preview completes', (tester) async {
final response = Completer<ResolveResult>();
final released = <String>[];
String? requestedID;
final registry = CapabilityRegistry(createAppCapabilityCodecs())
..bind(GopeedMethods.resolve, (request) {
requestedID = request.id;
return response.future;
})
..bind(GopeedMethods.releaseResolve, (id) {
released.add(id);
return const RpcUnit();
});
await open(tester, [pending('torrent')], registry);
await tester.tap(find.byKey(const ValueKey('resolve-close-button')));
await tester.pumpAndSettle();
expect(requestedID, isNotEmpty);
expect(released, contains(requestedID));
response.complete(
ResolveResult(
id: requestedID!,
res: Resource(files: [FileInfo(name: 'late.txt')]),
),
);
await tester.pumpAndSettle();
expect(released.where((id) => id == requestedID), hasLength(2));
});
testWidgets('successful creation keeps task-owned resources', (tester) async {
final released = <String>[];
final registry = CapabilityRegistry(createAppCapabilityCodecs())
..bind(GopeedMethods.resolve, (_) => resolved('task-owned'))
..bind(GopeedMethods.createTaskBatch, (_) => CreateTaskBatchResult(taskIds: ['task']))
..bind(GopeedMethods.releaseResolve, (id) {
released.add(id);
return const RpcUnit();
});
await open(tester, [pending('torrent')], registry);
await tester.pumpAndSettle();
await tester.tap(find.byKey(const ValueKey('resolve-create-button')));
await tester.pumpAndSettle();
expect(released, isEmpty);
});
for (final width in [390.0, 700.0, 1024.0]) { for (final width in [390.0, 700.0, 1024.0]) {
testWidgets('file list fits and exposes ready selections at width $width', (tester) async { testWidgets('file list fits and exposes ready selections at width $width', (tester) async {
final registry = CapabilityRegistry(createAppCapabilityCodecs()) final registry = CapabilityRegistry(createAppCapabilityCodecs())