fix(resolve): release cancelled previews and preserve task ownership

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 {

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()

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)
+		})
+	}
+}

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

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"`
 }