From 8161042b938ee380d908acf2c2e4c8d4e1283884 Mon Sep 17 00:00:00 2001 From: Levi Date: Tue, 10 Nov 2026 09:39:07 +0800 Subject: [PATCH] fix(resolve): release cancelled previews and preserve task ownership --- internal/protocol/http/fetcher.go | 12 + pkg/api/service.go | 10 +- pkg/download/downloader.go | 193 ++++++++++++--- pkg/download/resolve_lifecycle_test.go | 224 ++++++++++++++++++ pkg/rest/api.go | 2 +- pkg/rest/model/task.go | 1 + ui/flutter/lib/api/api.dart | 3 + ui/flutter/lib/api/model/resolve_task.dart | 5 +- ui/flutter/lib/api/model/resolve_task.g.dart | 3 +- .../core/capabilities/app_capabilities.dart | 4 + .../core/capabilities/gopeed_capability.dart | 2 + .../tasks/application/pending_download.dart | 30 ++- .../widgets/resolve_download_dialog.dart | 10 + .../resolve_download_dialog_test.dart | 43 ++++ 14 files changed, 506 insertions(+), 36 deletions(-) create mode 100644 pkg/download/resolve_lifecycle_test.go diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index cfc4f9f8..e7ba4b2f 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -418,6 +418,16 @@ func (f *Fetcher) updateMaxConnTime(d time.Duration) { } 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 { 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) // The body outlives Resolve, so the read timeout gets its own cancel func. reqCtx, reqCancel := context.WithCancel(context.Background()) + stopCancel := context.AfterFunc(ctx, reqCancel) + defer stopCancel() httpReq, err := f.buildRequest(reqCtx, req) if err != nil { reqCancel() diff --git a/pkg/api/service.go b/pkg/api/service.go index 182f96f1..4c8856a5 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -51,6 +51,7 @@ func NewService(downloader *download.Downloader) (*Service, error) { service.routes = []RouteSpec{ newRoute(http.MethodGet, "/api/v1/info", service.info), 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/batch", service.createTaskBatch), 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 { return response } - result, err := s.Downloader.Resolve(request.Req) + result, err := s.Downloader.ResolveWithID(request.ID, request.Req) if err != nil { return newErrorResponse(err.Error()) } 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 { var request restmodel.CreateTask if response := decodeRequest(ctx.Body, &request); response != nil { diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index e7da2caf..6fcd0419 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -116,6 +116,7 @@ type Downloader struct { cfg *DownloaderConfig fetcherCache map[string]fetcher.Fetcher + resolveStates map[string]*resolveState storage Storage tasks []*Task waitTasks []*Task @@ -153,6 +154,7 @@ func NewDownloader(cfg *DownloaderConfig) *Downloader { cancelBackground: cancelBackground, cfg: cfg, fetcherCache: make(map[string]fetcher.Fetcher), + resolveStates: make(map[string]*resolveState), waitTasks: make([]*Task, 0), groups: make(map[string]*TaskGroup), creatingGroups: make(map[string]struct{}), @@ -559,42 +561,156 @@ func ensureResourceRequestRawURLs(parentReq *base.Request, res *base.Resource) { } } -func (d *Downloader) Resolve(req *base.Request) (rr *ResolveResult, err error) { - rrId, err := gonanoid.New() - if err != nil { - return - } - ensureRequestRawURL(req) +// resolveState owns a Fetcher until Create transfers it to a task. +// State transitions are protected by fetcherMapLock; closeOnce serializes cleanup. +type contextResolver interface { + ResolveContext(context.Context, *base.Request) error +} +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) if err != nil { - return + return nil, err } if res != nil && len(res.Files) > 0 { - rr = &ResolveResult{ - Res: res, - SuggestedDownloadDir: d.suggestedDownloadDir(res), - } - return + return &ResolveResult{Res: res, SuggestedDownloadDir: d.suggestedDownloadDir(res)}, nil } - - fetcher, err := d.buildFetcher(req.URL) + f, err := d.buildFetcher(req.URL) if err != nil { - return - } - err = fetcher.Resolve(req) - if err != nil { - return + return nil, err } d.fetcherMapLock.Lock() - d.fetcherCache[rrId] = fetcher + state.fetcher = f + cancelled := state.cancelled || d.closed.Load() d.fetcherMapLock.Unlock() - rr = &ResolveResult{ - ID: rrId, - Res: fetcher.Meta().Res, - SuggestedDownloadDir: d.suggestedDownloadDir(fetcher.Meta().Res), + if cancelled { + return nil, errors.New("resource resolution cancelled") + } + 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() { @@ -868,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) { d.fetcherMapLock.Lock() fetcher, ok := d.fetcherCache[rrId] + state := d.resolveStates[rrId] + if ok && state != nil { + state.claimed = true + } delete(d.fetcherCache, rrId) d.fetcherMapLock.Unlock() if !ok { return "", errors.New("invalid resource id") } - // Only one creator may use a resolved fetcher. Failed creation makes it - // available again; a persisted task consumes it even if starting fails. + // Failed creation returns ownership to the preview unless it was cancelled. 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.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 { @@ -1562,6 +1694,7 @@ func (d *Downloader) Close() error { return nil } d.cancelBackground() + d.closeResolvedResources() d.waitForExtensionGit() if d.protocolJobs != nil { d.protocolJobs.close() diff --git a/pkg/download/resolve_lifecycle_test.go b/pkg/download/resolve_lifecycle_test.go new file mode 100644 index 00000000..553738b2 --- /dev/null +++ b/pkg/download/resolve_lifecycle_test.go @@ -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) + }) + } +} diff --git a/pkg/rest/api.go b/pkg/rest/api.go index 4d119276..cafa1664 100644 --- a/pkg/rest/api.go +++ b/pkg/rest/api.go @@ -24,7 +24,7 @@ func Info(w http.ResponseWriter, r *http.Request) { func Resolve(w http.ResponseWriter, r *http.Request) { var req model.ResolveTask if ReadJson(r, w, &req) { - rr, err := Downloader.Resolve(req.Req) + rr, err := Downloader.ResolveWithID(req.ID, req.Req) if err != nil { WriteJson(w, model.NewErrorResult(err.Error())) return diff --git a/pkg/rest/model/task.go b/pkg/rest/model/task.go index 4091e977..43fd50bb 100644 --- a/pkg/rest/model/task.go +++ b/pkg/rest/model/task.go @@ -3,6 +3,7 @@ package model import "github.com/GopeedLab/gopeed/pkg/base" type ResolveTask struct { + ID string `json:"id,omitempty"` Req *base.Request `json:"req"` } diff --git a/ui/flutter/lib/api/api.dart b/ui/flutter/lib/api/api.dart index c2bf519a..f924e359 100644 --- a/ui/flutter/lib/api/api.dart +++ b/ui/flutter/lib/api/api.dart @@ -81,6 +81,9 @@ Future resolve(ResolveTask resolveTask) { ); } +Future releaseResolve(String id) => + _parse(() => _request('api/v1/resolve/${Uri.encodeComponent(id)}', method: 'DELETE'), null); + Future createTask(CreateTask createTask) { return _parse(() => _request('api/v1/tasks', method: 'POST', data: createTask), (data) => data as String); } diff --git a/ui/flutter/lib/api/model/resolve_task.dart b/ui/flutter/lib/api/model/resolve_task.dart index 5fc8cc1b..bdc3b3de 100644 --- a/ui/flutter/lib/api/model/resolve_task.dart +++ b/ui/flutter/lib/api/model/resolve_task.dart @@ -8,7 +8,10 @@ part 'resolve_task.g.dart'; class ResolveTask { Request? req; - ResolveTask({this.req}); + @JsonKey(includeIfNull: false) + String? id; + + ResolveTask({this.req, this.id}); factory ResolveTask.fromJson(Map json) => _$ResolveTaskFromJson(json); Map toJson() => _$ResolveTaskToJson(this); diff --git a/ui/flutter/lib/api/model/resolve_task.g.dart b/ui/flutter/lib/api/model/resolve_task.g.dart index 0305902b..3e0e9f18 100644 --- a/ui/flutter/lib/api/model/resolve_task.g.dart +++ b/ui/flutter/lib/api/model/resolve_task.g.dart @@ -10,7 +10,8 @@ ResolveTask _$ResolveTaskFromJson(Map json) => ResolveTask( req: json['req'] == null ? null : Request.fromJson(json['req'] as Map), + id: json['id'] as String?, ); Map _$ResolveTaskToJson(ResolveTask instance) => - {'req': ?instance.req}; + {'req': ?instance.req, 'id': ?instance.id}; diff --git a/ui/flutter/lib/core/capabilities/app_capabilities.dart b/ui/flutter/lib/core/capabilities/app_capabilities.dart index a760dcfa..b7121253 100644 --- a/ui/flutter/lib/core/capabilities/app_capabilities.dart +++ b/ui/flutter/lib/core/capabilities/app_capabilities.dart @@ -93,6 +93,10 @@ void _bindGopeed(CapabilityRegistry registry, _ConfigStore configStore) { (p) => api.getMirrorCandidates(p['provider'] as String, p['url'] as String), ) ..bind(GopeedMethods.resolve, api.resolve) + ..bind(GopeedMethods.releaseResolve, (id) async { + await api.releaseResolve(id); + return const RpcUnit(); + }) ..bind(GopeedMethods.createTask, api.createTask) ..bind(GopeedMethods.createTaskBatch, api.createTaskBatch) ..bind(GopeedMethods.patchTask, (params) async { diff --git a/ui/flutter/lib/core/capabilities/gopeed_capability.dart b/ui/flutter/lib/core/capabilities/gopeed_capability.dart index 6254040f..80292a1e 100644 --- a/ui/flutter/lib/core/capabilities/gopeed_capability.dart +++ b/ui/flutter/lib/core/capabilities/gopeed_capability.dart @@ -16,6 +16,7 @@ abstract final class GopeedMethods { static const triggerProtocolJob = RpcMethod, RpcUnit>('gopeed.protocol.job.trigger'); static const getMirrorCandidates = RpcMethod, List>('gopeed.mirrors.candidates'); static const resolve = RpcMethod('gopeed.resolve'); + static const releaseResolve = RpcMethod('gopeed.releaseResolve'); static const createTask = RpcMethod('gopeed.task.create'); static const createTaskBatch = RpcMethod('gopeed.task.createBatch'); static const patchTask = RpcMethod, RpcUnit>('gopeed.task.patch'); @@ -50,6 +51,7 @@ class GopeedService { Future> getMirrorCandidates(String provider, String url) => _invoker.invoke(GopeedMethods.getMirrorCandidates, {'provider': provider, 'url': url}); Future resolve(ResolveTask request) => _invoker.invoke(GopeedMethods.resolve, request); + Future releaseResolve(String id) async => _invoker.invoke(GopeedMethods.releaseResolve, id); Future createTask(CreateTask request) => _invoker.invoke(GopeedMethods.createTask, request); diff --git a/ui/flutter/lib/features/tasks/application/pending_download.dart b/ui/flutter/lib/features/tasks/application/pending_download.dart index a23e75d2..58dfe139 100644 --- a/ui/flutter/lib/features/tasks/application/pending_download.dart +++ b/ui/flutter/lib/features/tasks/application/pending_download.dart @@ -1,3 +1,5 @@ +import 'dart:developer' as developer; +import 'package:uuid/uuid.dart'; import 'package:path/path.dart' as path; import '../../../api/model/create_task_batch.dart'; @@ -16,6 +18,8 @@ class PendingDownload { final Request request; final Options options; ResolveResult? result; + String? _resolveID; + bool _cancelled = false; String? resolveError; String? createError; final Set selected = {}; @@ -50,11 +54,16 @@ class PendingDownload { bool get hasSelection => !created && result != null && selected.isNotEmpty; Future resolve(GopeedService service) async { - if (result != null || resolving) return; + if (_cancelled || result != null || resolving) return; resolving = true; resolveError = null; 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)); } catch (error) { 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 cancel(GopeedService service) async { + if (_cancelled || created) return; + _cancelled = true; + await _release(service, result?.id.isNotEmpty == true ? result!.id : _resolveID); + } + + Future _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 creationItems() { final indexes = selected.toList()..sort(); if (result!.id.isNotEmpty || files.isEmpty) { diff --git a/ui/flutter/lib/features/tasks/presentation/widgets/resolve_download_dialog.dart b/ui/flutter/lib/features/tasks/presentation/widgets/resolve_download_dialog.dart index 92a83d92..060b0552 100644 --- a/ui/flutter/lib/features/tasks/presentation/widgets/resolve_download_dialog.dart +++ b/ui/flutter/lib/features/tasks/presentation/widgets/resolve_download_dialog.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:shadcn_flutter/shadcn_flutter.dart'; import 'package:flutter/widgets.dart' as flutter show Flexible; @@ -49,6 +51,7 @@ class _ResolveDownloadsDialogState extends State<_ResolveDownloadsDialog> { @override void dispose() { _cancelled = true; + _releasePreviews(); 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() { _cancelled = true; + _releasePreviews(); closeOverlay(context, false); } diff --git a/ui/flutter/test/features/tasks/presentation/resolve_download_dialog_test.dart b/ui/flutter/test/features/tasks/presentation/resolve_download_dialog_test.dart index e4c2abfe..5e259a5e 100644 --- a/ui/flutter/test/features/tasks/presentation/resolve_download_dialog_test.dart +++ b/ui/flutter/test/features/tasks/presentation/resolve_download_dialog_test.dart @@ -96,6 +96,49 @@ void main() { expect(released, contains('late-rid')); }); + testWidgets('cancel sends the request ID before a preview completes', (tester) async { + final response = Completer(); + final released = []; + 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 = []; + 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]) { testWidgets('file list fits and exposes ready selections at width $width', (tester) async { final registry = CapabilityRegistry(createAppCapabilityCodecs())