From de65149b29c8be4dbdccd865918208dd82c35b19 Mon Sep 17 00:00:00 2001 From: Levi Date: Mon, 19 Oct 2026 21:40:57 +0800 Subject: [PATCH] refactor(downloader): remove opts parameter from Resolve (#1595) --- cmd/extmock/main.go | 4 +- internal/fetcher/fetcher.go | 2 +- internal/protocol/bt/download_test.go | 448 ++++++++++++++++++ internal/protocol/bt/fetcher.go | 325 +++++++++---- internal/protocol/bt/fetcher_test.go | 36 +- internal/protocol/bt/fixture_test.go | 212 +++++++++ internal/protocol/bt/selection_test.go | 80 +++- internal/protocol/bt/storage.go | 116 +++++ internal/protocol/ed2k/fetcher.go | 7 +- internal/protocol/ed2k/fetcher_test.go | 3 +- internal/protocol/hls/fetcher.go | 8 +- internal/protocol/hls/fetcher_test.go | 24 +- internal/protocol/hls/media_validate_test.go | 9 +- internal/protocol/hls/staging_test.go | 22 +- internal/protocol/http/fetcher.go | 13 +- .../http/fetcher_checksum_retry_test.go | 9 +- .../protocol/http/fetcher_checksum_test.go | 53 ++- internal/protocol/http/fetcher_test.go | 173 ++++--- internal/protocol/http/resolve_range_test.go | 6 +- pkg/api/service.go | 4 +- pkg/api/service_test.go | 82 ++++ pkg/download/downloader.go | 70 +-- pkg/download/downloader_test.go | 46 +- pkg/download/extension.go | 4 + pkg/download/extension_extra_test.go | 4 +- pkg/download/extension_test.go | 48 +- .../extension_unnamed_resource_test.go | 2 +- pkg/download/hls_live_test.go | 2 +- pkg/download/hls_protocol_test.go | 2 +- pkg/download/selection_test.go | 8 +- pkg/mcpserver/server.go | 5 +- pkg/mcpserver/server_test.go | 7 + pkg/rest/api.go | 4 +- pkg/rest/model/task.go | 4 + pkg/rest/server_test.go | 10 +- ui/flutter/lib/api/api.dart | 3 +- ui/flutter/lib/api/model/patch_task.dart | 16 + ui/flutter/lib/api/model/resolve_task.dart | 4 +- ui/flutter/lib/api/model/resolve_task.g.dart | 11 +- .../core/capabilities/app_capabilities.dart | 3 +- .../core/capabilities/gopeed_capability.dart | 3 +- .../tasks/application/tasks_controller.dart | 4 +- .../pages/create_task_window_page.dart | 3 +- ui/flutter/test/api/resolve_test.dart | 39 +- .../capabilities/capability_rpc_test.dart | 35 ++ 45 files changed, 1589 insertions(+), 384 deletions(-) create mode 100644 internal/protocol/bt/download_test.go create mode 100644 internal/protocol/bt/fixture_test.go create mode 100644 internal/protocol/bt/storage.go create mode 100644 pkg/api/service_test.go create mode 100644 ui/flutter/lib/api/model/patch_task.dart diff --git a/cmd/extmock/main.go b/cmd/extmock/main.go index 92903e04..9ed19177 100644 --- a/cmd/extmock/main.go +++ b/cmd/extmock/main.go @@ -166,7 +166,7 @@ func runResolve(args []string) error { result, err := d.Resolve(&base.Request{ URL: url, Labels: map[string]string(labels), - }, nil) + }) if err != nil { return err } @@ -250,7 +250,7 @@ func runDownload(args []string) error { result, err := d.Resolve(&base.Request{ URL: url, Labels: map[string]string(labels), - }, nil) + }) if err != nil { return err } diff --git a/internal/fetcher/fetcher.go b/internal/fetcher/fetcher.go index 8d3a7821..50319a13 100644 --- a/internal/fetcher/fetcher.go +++ b/internal/fetcher/fetcher.go @@ -12,7 +12,7 @@ import ( // Each download task will have a corresponding Fetcher instance for the management of the download task type Fetcher interface { Setup(ctl *controller.Controller) - Resolve(req *base.Request, opts *base.Options) error + Resolve(req *base.Request) error Start() error // Patch modifies task-specific data based on the protocol. // For HTTP: can modify Request info (URL, headers, etc.) diff --git a/internal/protocol/bt/download_test.go b/internal/protocol/bt/download_test.go new file mode 100644 index 00000000..8ae3a9a8 --- /dev/null +++ b/internal/protocol/bt/download_test.go @@ -0,0 +1,448 @@ +package bt + +import ( + "encoding/json" + "net/url" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/GopeedLab/gopeed/internal/controller" + "github.com/GopeedLab/gopeed/internal/fetcher" + "github.com/GopeedLab/gopeed/pkg/base" + btprotocol "github.com/GopeedLab/gopeed/pkg/protocol/bt" + "github.com/anacrolix/torrent" + pp "github.com/anacrolix/torrent/peer_protocol" + "golang.org/x/time/rate" +) + +type btDownloadHarness struct { + *Fetcher + pendingRequests atomic.Int64 +} + +// These tests exercise the BT fetcher, native storage and real peer protocol. +// All content is generated locally and served by a loopback seeder. +func btDownloadFetcher(t *testing.T, f fetcher.Fetcher) *btDownloadHarness { + t.Helper() + if f == nil { + f = new(FetcherManager).Build() + } + ctl := controller.NewController() + ctl.GetConfig = func(v any) { + encoded, err := json.Marshal(new(FetcherManager).DefaultConfig()) + if err != nil { + t.Fatal(err) + } + if err := json.Unmarshal(encoded, v); err != nil { + t.Fatal(err) + } + } + f.Setup(ctl) + t.Cleanup(func() { + if err := f.Close(); err != nil { + t.Error(err) + } + if err := closeClient(); err != nil { + t.Error(err) + } + }) + h := &btDownloadHarness{Fetcher: f.(*Fetcher)} + // Keep the test client on loopback and observe real cancellation replies. + // Waiting for them avoids the pinned library's immediate cancel/resume + // request scheduling bug, without relying on arbitrary sleep durations. + cfg = torrent.TestingConfig(t) + cfg.DisableTrackers = false + cfg.Seed = true + cfg.DisableUTP = true + cfg.DisableIPv6 = true + cfg.DisableWebtorrent = true + cfg.KeepAliveTimeout = time.Minute + cfg.MaxAllocPeerRequestDataPerConn = 1 << 20 + cfg.Callbacks.SentRequest = []func(torrent.PeerRequestEvent){func(torrent.PeerRequestEvent) { + h.pendingRequests.Add(1) + }} + cfg.Callbacks.ReadMessage = func(_ *torrent.PeerConn, msg *pp.Message) { + if !msg.Keepalive && (msg.Type == pp.Piece || msg.Type == pp.Reject) { + h.pendingRequests.Add(-1) + } + } + var err error + client, err = torrent.NewClient(cfg) + if err != nil { + t.Fatal(err) + } + return h +} + +func btResolve(t *testing.T, f *btDownloadHarness, source string) { + t.Helper() + if err := f.Resolve(&base.Request{URL: source}); err != nil { + t.Fatal(err) + } +} + +func btStart(t *testing.T, f *btDownloadHarness) { + t.Helper() + if err := f.Start(); err != nil { + t.Fatal(err) + } +} + +func btAwaitDone(t *testing.T, f *btDownloadHarness) { + t.Helper() + done := make(chan error, 1) + go func() { done <- f.Wait() }() + timeout := time.NewTimer(20 * time.Second) + defer timeout.Stop() + select { + case err := <-done: + if err != nil { + t.Fatalf("BT download failed: %v", err) + } + if !f.isDone() { + t.Fatal("Wait returned before selected files were verified") + } + case <-timeout.C: + var status strings.Builder + client.WriteStatus(&status) + t.Log(status.String()) + t.Fatalf("BT download did not finish: progress=%v stats=%+v pieces=%+v peers=%+v", f.Progress(), f.torrent.Stats(), f.torrent.PieceStateRuns(), f.Stats().Runtime) + } +} + +func btAwaitPartial(t *testing.T, f *btDownloadHarness) { + t.Helper() + timeout := time.NewTimer(15 * time.Second) + defer timeout.Stop() + ticker := time.NewTicker(20 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-ticker.C: + progress := f.Progress().TotalDownloaded() + if progress >= 32*1024 && progress < f.Meta().Res.Size { + return + } + case <-timeout.C: + t.Fatal("BT transfer did not reach partial progress") + } + } +} + +func btPause(t *testing.T, f *btDownloadHarness) { + t.Helper() + if err := f.Pause(); err != nil { + t.Fatal(err) + } + timeout := time.NewTimer(5 * time.Second) + defer timeout.Stop() + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for { + if f.pendingRequests.Load() == 0 { + // The message callback runs under the client lock. Acquiring that + // lock ensures the final cancellation reply has finished processing. + client.Stats() + return + } + select { + case <-ticker.C: + case <-timeout.C: + t.Fatal("peer did not acknowledge cancelled requests") + } + } +} + +func TestBTDownloadSources(t *testing.T) { + for _, kind := range []string{"magnet", "torrent", "file URI", "data URI"} { + t.Run(kind, func(t *testing.T) { + fixture := newBTFixture(t, map[string][]byte{"": btPayload(96*1024, kind)}, 0) + source := fixture.Magnet + switch kind { + case "torrent": + source = fixture.TorrentFile + case "file URI": + source = fixture.FileURI() + case "data URI": + source = fixture.DataURI + } + f := btDownloadFetcher(t, nil) + btResolve(t, f, source) + if f.Meta().Opts != nil { + t.Fatal("Resolve assigned download options") + } + f.storage.mu.RLock() + bound := f.storage.owner != nil + f.storage.mu.RUnlock() + if bound { + t.Fatal("Resolve bound download storage") + } + resolvedTorrent := f.torrent + dir := filepath.Join(t.TempDir(), "chosen-after-resolve") + f.Meta().Opts = &base.Options{Path: dir} + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("output directory exists before Start: %v", err) + } + btStart(t, f) + if f.torrent != resolvedTorrent { + t.Fatal("Start replaced the resolved torrent") + } + btAwaitDone(t, f) + fixture.AssertFiles(t, dir) + }) + } +} + +func TestBTDirectDownloadWithExtraTrackers(t *testing.T) { + for _, source := range []string{"request", "config"} { + t.Run(source, func(t *testing.T) { + fixture := makeBTFixture(t, map[string][]byte{ + "a.bin": btPayload(20*1024, "direct-a"), + "nested/b.bin": btPayload(48*1024, "direct-b"), + }, 0, false, false) + magnet, err := url.Parse(fixture.Magnet) + if err != nil { + t.Fatal(err) + } + query := magnet.Query() + query.Del("tr") + query.Del("x.pe") + magnet.RawQuery = query.Encode() + req := &base.Request{URL: magnet.String()} + f := btDownloadFetcher(t, nil) + if source == "request" { + req.Extra = &btprotocol.ReqExtra{Trackers: []string{fixture.TrackerURL}} + } else { + f.config.Trackers = []string{fixture.TrackerURL} + } + // CreateDirect supplies options before Downloader resolves the task. + dir := filepath.Join(t.TempDir(), "direct-download") + opts := &base.Options{Path: dir, SelectFiles: []int{}} + f.Meta().Opts = opts + resolved := make(chan error, 1) + go func() { resolved <- f.Resolve(req) }() + select { + case err := <-resolved: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + f.torrentDropFunc() + <-resolved + t.Fatal("direct magnet download did not obtain metadata from its extra tracker") + } + if f.Meta().Opts != opts || len(opts.SelectFiles) != len(fixture.Files) { + t.Fatal("resolution lost direct download options or did not select all files") + } + btStart(t, f) + btAwaitDone(t, f) + fixture.AssertFiles(t, dir) + }) + } +} + +func TestBTDownloadSelectedFiles(t *testing.T) { + fixture := newBTFixture(t, map[string][]byte{ + "a.bin": btPayload(20*1024, "a"), "nested/b.bin": btPayload(48*1024, "b"), + }, 0) + f := btDownloadFetcher(t, nil) + dir := t.TempDir() + btResolve(t, f, fixture.Magnet) + selected := -1 + for i, file := range f.Meta().Res.Files { + if file.Name == "b.bin" { + selected = i + } + } + if selected < 0 { + t.Fatal("selected file absent from metadata") + } + f.Meta().Opts = &base.Options{Path: dir} + if err := f.Patch(nil, &base.Options{SelectFiles: []int{selected}}); err != nil { + t.Fatal(err) + } + btStart(t, f) + btAwaitDone(t, f) + fixture.AssertFiles(t, dir, filepath.Join("bundle", "nested", "b.bin")) + if _, err := os.Stat(filepath.Join(dir, "bundle", "a.bin")); !os.IsNotExist(err) { + t.Fatalf("unselected output remains: %v", err) + } + if f.Meta().Res.Size != 48*1024 { + t.Fatal("selected resource size is incorrect") + } +} + +func TestBTPauseAndResume(t *testing.T) { + fixture := newBTFixture(t, map[string][]byte{"": btPayload(512*1024, "pause")}, 64*1024) + f := btDownloadFetcher(t, nil) + dir := t.TempDir() + btResolve(t, f, fixture.Magnet) + f.Meta().Opts = &base.Options{Path: dir} + btStart(t, f) + btAwaitPartial(t, f) + btPause(t, f) + resolvedTorrent := f.torrent + fixture.Limiter.SetLimit(rate.Inf) + // Once cancelled requests have settled, no more data should arrive. + stats := f.torrent.Stats() + readWhilePaused := stats.BytesReadData.Int64() + time.Sleep(200 * time.Millisecond) + stats = f.torrent.Stats() + if stats.BytesReadData.Int64() != readWhilePaused || f.isDone() { + t.Fatal("data download continued while paused") + } + btStart(t, f) + if f.torrent != resolvedTorrent { + t.Fatal("resume replaced the resolved torrent") + } + btAwaitDone(t, f) + fixture.AssertFiles(t, dir) +} + +type btFetcherCheckpoint struct { + Meta *fetcher.FetcherMeta `json:"meta"` + Data json.RawMessage `json:"data"` +} + +func btStore(t *testing.T, f *btDownloadHarness) []byte { + t.Helper() + data, err := new(FetcherManager).Store(f.Fetcher) + if err != nil { + t.Fatal(err) + } + encodedData, err := json.Marshal(data) + if err != nil { + t.Fatal(err) + } + encoded, err := json.Marshal(btFetcherCheckpoint{Meta: f.Meta(), Data: encodedData}) + if err != nil { + t.Fatal(err) + } + return encoded +} + +func btRestore(t *testing.T, encoded []byte) *btDownloadHarness { + t.Helper() + var checkpoint btFetcherCheckpoint + if err := json.Unmarshal(encoded, &checkpoint); err != nil { + t.Fatal(err) + } + data, restore := new(FetcherManager).Restore() + if err := json.Unmarshal(checkpoint.Data, data); err != nil { + t.Fatal(err) + } + return btDownloadFetcher(t, restore(checkpoint.Meta, data)) +} + +func btCloseForRestart(t *testing.T, f *btDownloadHarness) { + t.Helper() + if err := f.Close(); err != nil { + t.Fatal(err) + } + if err := new(FetcherManager).Close(); err != nil { + t.Fatal(err) + } +} + +func TestBTColdStartResume(t *testing.T) { + for _, kind := range []string{"v1", "hybrid"} { + t.Run(kind, func(t *testing.T) { + var fixture *btFixture + if kind == "hybrid" { + fixture = newBTHybridFixture(t, btPayload(512*1024, "restart-hybrid")) + fixture.Limiter.SetLimit(64 * 1024) + } else { + fixture = newBTFixture(t, map[string][]byte{"": btPayload(512*1024, "restart")}, 64*1024) + } + f := btDownloadFetcher(t, nil) + dir := t.TempDir() + btResolve(t, f, fixture.Magnet) + f.Meta().Opts = &base.Options{Path: dir} + btStart(t, f) + btAwaitPartial(t, f) + btPause(t, f) + checkpoint := btStore(t, f) + hash := f.Meta().Res.Hash + btCloseForRestart(t, f) + seedStats := fixture.Torrent.Stats() + sentBeforeRestart := seedStats.BytesWrittenData.Int64() + restored := btRestore(t, checkpoint) + if restored.Meta().Opts.Path != dir || restored.Meta().Res.Hash != hash { + t.Fatal("download directory or hash was lost") + } + fixture.Limiter.SetLimit(rate.Inf) + btStart(t, restored) + btAwaitDone(t, restored) + fixture.AssertFiles(t, dir) + seedStats = fixture.Torrent.Stats() + if sent := seedStats.BytesWrittenData.Int64() - sentBeforeRestart; sent >= 512*1024 { + t.Fatalf("restart redownloaded the entire file: %d bytes", sent) + } + }) + } +} + +func TestBTDownloadHybrid(t *testing.T) { + fixture := newBTHybridFixture(t, btPayload(80*1024, "hybrid-v2")) + f := btDownloadFetcher(t, nil) + dir := t.TempDir() + btResolve(t, f, fixture.Magnet) + f.Meta().Opts = &base.Options{Path: dir} + btStart(t, f) + btAwaitDone(t, f) + fixture.AssertFiles(t, dir) +} + +func TestBTRepairCorruptedExistingFile(t *testing.T) { + fixture := newBTFixture(t, map[string][]byte{"": btPayload(128*1024, "repair")}, 0) + dir := t.TempDir() + damaged := append([]byte(nil), fixture.Files["payload.bin"]...) + damaged[20000] ^= 0xff + if err := os.WriteFile(filepath.Join(dir, "payload.bin"), damaged, 0600); err != nil { + t.Fatal(err) + } + f := btDownloadFetcher(t, nil) + btResolve(t, f, fixture.Magnet) + f.Meta().Opts = &base.Options{Path: dir} + btStart(t, f) + btAwaitDone(t, f) + fixture.AssertFiles(t, dir) +} + +func TestBTPatchSelectionWhilePaused(t *testing.T) { + fixture := newBTFixture(t, map[string][]byte{"a.bin": btPayload(256*1024, "patch-a"), "b.bin": btPayload(256*1024, "patch-b")}, 64*1024) + f := btDownloadFetcher(t, nil) + dir := t.TempDir() + btResolve(t, f, fixture.Magnet) + f.Meta().Opts = &base.Options{Path: dir, SelectFiles: []int{0}} + btStart(t, f) + btAwaitPartial(t, f) + btPause(t, f) + if err := f.Patch(nil, &base.Options{SelectFiles: []int{1}}); err != nil { + t.Fatal(err) + } + fixture.Limiter.SetLimit(rate.Inf) + btStart(t, f) + btAwaitDone(t, f) + fixture.AssertFiles(t, dir, filepath.Join("bundle", "b.bin")) + for _, suffix := range []string{"", ".part"} { + if _, err := os.Stat(filepath.Join(dir, "bundle", "a.bin") + suffix); !os.IsNotExist(err) { + t.Fatalf("unselected output remains: %v", err) + } + } +} + +func TestBTDownloadMultiFileWithEmptyFile(t *testing.T) { + fixture := newBTFixture(t, map[string][]byte{"empty.txt": {}, "nested/data.bin": btPayload(48*1024, "empty-file")}, 0) + f := btDownloadFetcher(t, nil) + dir := t.TempDir() + btResolve(t, f, fixture.Magnet) + f.Meta().Opts = &base.Options{Path: dir} + btStart(t, f) + btAwaitDone(t, f) + fixture.AssertFiles(t, dir) +} diff --git a/internal/protocol/bt/fetcher.go b/internal/protocol/bt/fetcher.go index e22c403a..874c9a62 100644 --- a/internal/protocol/bt/fetcher.go +++ b/internal/protocol/bt/fetcher.go @@ -3,11 +3,13 @@ package bt import ( "bytes" "context" + "errors" "fmt" "io" "net/url" "os" "path/filepath" + "runtime" "strings" "sync" "sync/atomic" @@ -33,6 +35,10 @@ var ( ) type Fetcher struct { + mu sync.Mutex + closeOnce sync.Once + storage *deferredFileStorage + ctl *controller.Controller config *config @@ -94,13 +100,17 @@ func (f *Fetcher) initClient() (err error) { return } -func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) error { - f.meta.Req = req - f.meta.Opts = opts - if f.meta.Opts == nil { - f.meta.Opts = &base.Options{} +func (f *Fetcher) Resolve(req *base.Request) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.torrent != nil { + f.safeDrop() + f.torrent = nil + f.storage = nil + f.torrentReady.Store(false) } - if err := f.addTorrent(req, false); err != nil { + f.meta.Req = req + if err := f.addTorrent(req); err != nil { return err } f.updateRes() @@ -108,12 +118,72 @@ func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) error { } func (f *Fetcher) Start() (err error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.meta.Opts == nil || f.meta.Opts.Path == "" { + return errors.New("download directory is required when starting a torrent") + } + if err := f.torrentDropCtx.Err(); err != nil { + return err + } if !f.torrentReady.Load() { - if err = f.addTorrent(f.meta.Req, false); err != nil { + if err = f.addTorrent(f.meta.Req); err != nil { return } } + path := f.meta.Opts.Path + if err := os.MkdirAll(path, 0755); err != nil { + return err + } + hashes, err := f.storage.bind(storage.NewFileClientOpts{ + ClientBaseDir: path, + FilePathMaker: func(opts storage.FilePathMakerOpts) string { + _, name := torrentFileLayout(opts.Info, *opts.File) + return name + }, + }) + if err != nil { + return err + } + if hashes != nil { + // File storage infers completion from final file names and sizes. Verify + // before publishing that state, or peers can see an unverified file as + // complete and disconnect before a damaged piece is discovered. + // Late v2 hashes schedule their own initial checks through the library. + changes := f.torrent.SubscribePieceStateChanges() + defer changes.Close() + for i, known := range hashes { + if !known { + continue + } + if err := f.torrentDropCtx.Err(); err != nil { + return err + } + if err := f.torrent.Piece(i).VerifyDataContext(f.torrentDropCtx); err != nil { + return err + } + // VerifyDataContext can return while the result is still being + // written to storage. Wait for it before refreshing completion. + for { + state := f.torrent.Piece(i).State() + if !state.Hashing && !state.QueuedForHash && !state.Marking { + break + } + select { + case <-changes.Values: + case <-f.torrentDropCtx.Done(): + return f.torrentDropCtx.Err() + case <-f.torrent.Closed(): + return errors.New("torrent closed during verification") + } + } + } + for i := range hashes { + f.torrent.Piece(i).UpdateCompletion() + } + } + files := f.torrent.Files() // If the user does not specify the file to download, all files will be downloaded by default if f.data.Progress == nil { @@ -134,21 +204,29 @@ func (f *Fetcher) Start() (err error) { } } f.torrent.AllowDataDownload() + go f.doUpload(false) return } func (f *Fetcher) Pause() (err error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.torrent == nil { + return nil + } f.torrent.DisallowDataDownload() return } func (f *Fetcher) Close() (err error) { - f.safeDrop() - f.torrentDropFunc() - f.uploadDoneCh <- nil - if len(client.Torrents()) == 0 { - err = closeClient() - } + f.closeOnce.Do(func() { + f.torrentDropFunc() + f.mu.Lock() + defer f.mu.Unlock() + f.safeDrop() + f.torrentReady.Store(false) + close(f.uploadDoneCh) + }) return nil } @@ -166,6 +244,8 @@ func (f *Fetcher) Meta() *fetcher.FetcherMeta { } func (f *Fetcher) Stats() *fetcher.Stats { + f.mu.Lock() + defer f.mu.Unlock() var stats torrent.TorrentStats var pieceMap *base.PieceMap pieceMapReady := false @@ -372,48 +452,65 @@ func normalizeBTTransport(network string) string { } func (f *Fetcher) Progress() fetcher.Progress { + f.mu.Lock() + defer f.mu.Unlock() if !f.torrentReady.Load() { - return f.data.Progress + return append(fetcher.Progress(nil), f.data.Progress...) } for i := range f.data.Progress { selectIndex := f.meta.Opts.SelectFiles[i] file := f.torrent.Files()[selectIndex] f.data.Progress[i] = file.BytesCompleted() } - return f.data.Progress + return append(fetcher.Progress(nil), f.data.Progress...) } -func (f *Fetcher) Wait() (err error) { +func (f *Fetcher) Wait() error { + f.mu.Lock() + var closed <-chan struct{} + if f.torrent != nil { + closed = f.torrent.Closed() + } + f.mu.Unlock() + ticker := time.NewTicker(time.Second) + defer ticker.Stop() for { select { case <-f.torrentDropCtx.Done(): - return - case <-time.After(time.Second): - if f.torrentReady.Load() && len(f.meta.Opts.SelectFiles) > 0 { - if f.isDone() { - // remove unselected files - for i, file := range f.torrent.Files() { - selected := false - for _, selectIndex := range f.meta.Opts.SelectFiles { - if i == selectIndex { - selected = true - break - } - } - if !selected { - root, name := torrentFileLayout(f.torrent.Info(), file.FileInfo()) - if err := removeUnselectedFile(f.meta.Opts.Path, root, name); err != nil { - return err - } - } - } - return - } + return nil + case <-closed: + return nil + case <-ticker.C: + done, err := f.finishDownload() + if done || err != nil { + return err } } } } +func (f *Fetcher) finishDownload() (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + if !f.torrentReady.Load() || f.meta.Opts == nil || len(f.meta.Opts.SelectFiles) == 0 || !f.isDoneLocked() { + return false, nil + } + selected := make(map[int]bool, len(f.meta.Opts.SelectFiles)) + for _, i := range f.meta.Opts.SelectFiles { + selected[i] = true + } + for i, file := range f.torrent.Files() { + if selected[i] { + continue + } + root, name := torrentFileLayout(f.torrent.Info(), file.FileInfo()) + if err := removeUnselectedFile(f.meta.Opts.Path, root, name); err != nil { + return false, err + } + } + return true, nil +} + // Match the file storage layout, including its NoName convention for rootless // torrents. File.Path() always prepends BestName, even when storage omits it. func torrentFileLayout(info *metainfo.Info, file metainfo.FileInfo) (root, name string) { @@ -471,13 +568,22 @@ func removeUnselectedFile(downloadDir, torrentRoot, torrentPath string) error { } func (f *Fetcher) isDone() bool { - if f.meta.Opts == nil { + f.mu.Lock() + defer f.mu.Unlock() + return f.isDoneLocked() +} + +func (f *Fetcher) isDoneLocked() bool { + if f.meta.Opts == nil || f.torrent == nil { return false } - for _, selectIndex := range f.meta.Opts.SelectFiles { - file := f.torrent.Files()[selectIndex] - if file.BytesCompleted() < file.Length() { - return false + for _, index := range f.meta.Opts.SelectFiles { + file := f.torrent.Files()[index] + for i := file.BeginPieceIndex(); i < file.EndPieceIndex(); i++ { + state := f.torrent.Piece(i).State() + if !state.Ok || !state.Complete || state.Hashing || state.QueuedForHash { + return false + } } } return true @@ -486,6 +592,11 @@ func (f *Fetcher) isDone() bool { // Patch modifies the BT task settings. // Invalid file indices are silently ignored. func (f *Fetcher) Patch(req *base.Request, opts *base.Options) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.meta.Opts == nil { + f.meta.Opts = &base.Options{} + } if opts == nil { return nil } @@ -542,9 +653,9 @@ func (f *Fetcher) updateRes() { Files: make([]*base.FileInfo, len(f.torrent.Files())), Hash: f.torrent.InfoHash().String(), } - // Directory torrent - if f.torrent.Info().Length == 0 { - res.Name = f.torrent.Name() + // Use the same root classification as storage (v2 single files have Length == 0). + if files := f.torrent.Files(); len(files) > 0 { + res.Name, _ = torrentFileLayout(f.torrent.Info(), files[0].FileInfo()) } for i, file := range f.torrent.Files() { res.Files[i] = &base.FileInfo{ @@ -561,61 +672,50 @@ func (f *Fetcher) updateRes() { } func (f *Fetcher) Upload() (err error) { - return f.addTorrent(f.meta.Req, true) + if err = f.Start(); err != nil { + return err + } + return nil } func (f *Fetcher) doUpload(fromUpload bool) { if !f.torrentUpload.CompareAndSwap(false, true) { return } - - // Check and update seed data - lastData := &fetcherData{ - SeedBytes: f.data.SeedBytes, - SeedTime: f.data.SeedTime, - } - var doneTime int64 = 0 + f.mu.Lock() + initialBytes, initialTime := f.data.SeedBytes, f.data.SeedTime + torrentClosed := f.torrent.Closed() + f.mu.Unlock() + var doneTime int64 + ticker := time.NewTicker(time.Second) + defer ticker.Stop() for { select { case <-f.torrentDropCtx.Done(): return - case <-time.After(time.Second): + case <-torrentClosed: + return + case <-ticker.C: + f.mu.Lock() if !f.torrentReady.Load() { + f.mu.Unlock() continue } - stats := f.torrentStats() - f.data.SeedBytes = lastData.SeedBytes + stats.BytesWrittenData.Int64() - - // Check is download complete, if not don't check and stop seeding - if !fromUpload && !f.isDone() { + f.data.SeedBytes = initialBytes + stats.BytesWrittenData.Int64() + if !fromUpload && !f.isDoneLocked() { + f.mu.Unlock() continue } if doneTime == 0 { doneTime = time.Now().Unix() } - f.data.SeedTime = lastData.SeedTime + time.Now().Unix() - doneTime - - // If the seed forever is true, keep seeding - if f.config.SeedKeep { - continue - } - - // If the seed ratio is reached, stop seeding - if f.config.SeedRatio > 0 { - seedRadio := f.seedRadio() - if seedRadio >= f.config.SeedRatio { - f.Close() - break - } - } - - // If the seed time is reached, stop seeding - if f.config.SeedTime > 0 { - if f.data.SeedTime >= f.config.SeedTime { - f.Close() - break - } + f.data.SeedTime = initialTime + time.Now().Unix() - doneTime + stop := !f.config.SeedKeep && ((f.config.SeedRatio > 0 && f.seedRadio() >= f.config.SeedRatio) || (f.config.SeedTime > 0 && f.data.SeedTime >= f.config.SeedTime)) + f.mu.Unlock() + if stop { + f.Close() + return } } } @@ -633,6 +733,8 @@ func (f *Fetcher) torrentStats() torrent.TorrentStats { } func (f *Fetcher) UploadedBytes() int64 { + f.mu.Lock() + defer f.mu.Unlock() return f.data.SeedBytes } @@ -641,7 +743,7 @@ func (f *Fetcher) WaitUpload() (err error) { return nil } -func (f *Fetcher) addTorrent(req *base.Request, fromUpload bool) (err error) { +func (f *Fetcher) addTorrent(req *base.Request) (err error) { if err = base.ParseReqExtra[bt.ReqExtra](req); err != nil { return } @@ -660,11 +762,16 @@ func (f *Fetcher) addTorrent(req *base.Request, fromUpload bool) (err error) { var reader io.Reader if schema == "FILE" { fileUrl, _ := url.Parse(req.URL) - filePath := fileUrl.Path[1:] + filePath := fileUrl.Path + if runtime.GOOS == "windows" && len(filePath) > 2 && filePath[0] == '/' && filePath[2] == ':' { + filePath = filePath[1:] + } + filePath = filepath.FromSlash(filePath) reader, err = os.Open(filePath) if err != nil { return } + defer reader.(io.Closer).Close() } else if schema == "DATA" { _, data := util.ParseDataUri(req.URL) reader = bytes.NewBuffer(data) @@ -697,21 +804,6 @@ func (f *Fetcher) addTorrent(req *base.Request, fromUpload bool) (err error) { return } } - spec.Storage = storage.NewFileOpts(storage.NewFileClientOpts{ - ClientBaseDir: cfg.DataDir, - FilePathMaker: func(opts storage.FilePathMakerOpts) string { - _, name := torrentFileLayout(opts.Info, *opts.File) - return name - }, - TorrentDirMaker: func(baseDir string, info *metainfo.Info, infoHash metainfo.Hash) string { - return f.meta.Opts.Path - }, - }) - f.torrent, _, err = client.AddTorrentSpec(spec) - if err != nil { - return - } - // Do not add external tracker to a private torrent. if !privateTorrent { // use map to deduplicate @@ -730,17 +822,36 @@ func (f *Fetcher) addTorrent(req *base.Request, fromUpload bool) (err error) { } } if len(trackers) > 0 { - announceList := make([][]string, 0) for tracker := range trackers { - announceList = append(announceList, []string{tracker}) + spec.Trackers = append(spec.Trackers, []string{tracker}) } - f.torrent.AddTrackers(announceList) } } - <-f.torrent.GotInfo() + s := &deferredFileStorage{} + spec.Storage = s + var added bool + f.torrent, added, err = client.AddTorrentSpec(spec) + if err != nil { + return err + } + if !added { + f.torrent = nil + return errors.New("torrent is already active") + } + f.storage = s + f.torrent.DisallowDataDownload() + // Extra trackers must be available while magnet metadata is still missing; + // they may be the only way to discover a peer that can provide it. + select { + case <-f.torrent.GotInfo(): + case <-f.torrentDropCtx.Done(): + f.safeDrop() + return f.torrentDropCtx.Err() + case <-f.torrent.Closed(): + return errors.New("torrent closed during resolution") + } f.torrentReady.Store(true) - go f.doUpload(fromUpload) return } @@ -848,7 +959,11 @@ func (fm *FetcherManager) DefaultConfig() any { func (fm *FetcherManager) Store(f fetcher.Fetcher) (data any, err error) { _f := f.(*Fetcher) - return _f.data, nil + _f.mu.Lock() + defer _f.mu.Unlock() + snapshot := *_f.data + snapshot.Progress = append(fetcher.Progress(nil), snapshot.Progress...) + return &snapshot, nil } func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher) { diff --git a/internal/protocol/bt/fetcher_test.go b/internal/protocol/bt/fetcher_test.go index b13e38ed..02e18713 100644 --- a/internal/protocol/bt/fetcher_test.go +++ b/internal/protocol/bt/fetcher_test.go @@ -18,11 +18,11 @@ import ( ) func TestFetcher_Resolve_Torrent(t *testing.T) { - doResolve(t, buildFetcher()) + doResolve(t, buildFetcher(t)) } func TestFetcher_Resolve_DataUri_Torrent(t *testing.T) { - fetcher := buildFetcher() + fetcher := buildFetcher(t) buf, err := os.ReadFile("./testdata/ubuntu-22.04-live-server-amd64.iso.torrent") if err != nil { t.Fatal(err) @@ -31,7 +31,7 @@ func TestFetcher_Resolve_DataUri_Torrent(t *testing.T) { dataUri := "data:application/x-bittorrent;base64," + base64.StdEncoding.EncodeToString(buf) err = fetcher.Resolve(&base.Request{ URL: dataUri, - }, nil) + }) if err != nil { panic(err) } @@ -53,7 +53,7 @@ func TestFetcher_Resolve_DataUri_Torrent(t *testing.T) { } func TestFetcher_Config(t *testing.T) { - doResolve(t, buildConfigFetcher(nil)) + doResolve(t, buildConfigFetcher(t, nil)) } func TestFetcher_ResolveWithProxy(t *testing.T) { @@ -61,7 +61,7 @@ func TestFetcher_ResolveWithProxy(t *testing.T) { proxyListener := test.StartSocks5Server(usr, pwd) defer proxyListener.Close() - doResolve(t, buildConfigFetcher(&base.DownloaderProxyConfig{ + doResolve(t, buildConfigFetcher(t, &base.DownloaderProxyConfig{ Enable: true, System: false, Scheme: "socks5", @@ -81,7 +81,7 @@ func doResolve(t *testing.T, fetcher fetcher.Fetcher) { "udp://tracker.bitsearch.to:1337/announce", }, }, - }, nil) + }) if err != nil { panic(err) } @@ -111,7 +111,7 @@ func doResolve(t *testing.T, fetcher fetcher.Fetcher) { "udp://tracker.bitsearch.to:1337/announce", }, }, - }, nil) + }) if err != nil { panic(err) } @@ -145,7 +145,7 @@ func doResolve(t *testing.T, fetcher fetcher.Fetcher) { t.Run("Resolve Unclean Torrent", func(t *testing.T) { err := fetcher.Resolve(&base.Request{ URL: "./testdata/test.unclean.torrent", - }, nil) + }) if err != nil { t.Errorf("Resolve Unclean Torrent Resolve() got = %v, want nil", err) } @@ -156,7 +156,7 @@ func doResolve(t *testing.T, fetcher fetcher.Fetcher) { uri := "file:///" + file err := fetcher.Resolve(&base.Request{ URL: uri, - }, nil) + }) if err != nil { t.Errorf("Resolve file scheme Torrent Resolve() got = %v, want nil", err) } @@ -211,7 +211,7 @@ func TestFetcherManager_ParseName(t *testing.T) { } } -func buildFetcher() fetcher.Fetcher { +func buildFetcher(t testing.TB) fetcher.Fetcher { fb := new(FetcherManager) fetcher := fb.Build() newController := controller.NewController() @@ -219,10 +219,11 @@ func buildFetcher() fetcher.Fetcher { json.Unmarshal([]byte(test.ToJson(fb.DefaultConfig())), v) } fetcher.Setup(newController) + t.Cleanup(func() { fetcher.Close(); closeClient() }) return fetcher } -func buildConfigFetcher(proxyConfig *base.DownloaderProxyConfig) fetcher.Fetcher { +func buildConfigFetcher(t testing.TB, proxyConfig *base.DownloaderProxyConfig) fetcher.Fetcher { fetcher := new(FetcherManager).Build() newController := controller.NewController() mockCfg := config{ @@ -237,29 +238,30 @@ func buildConfigFetcher(proxyConfig *base.DownloaderProxyConfig) fetcher.Fetcher return proxyConfig.ToHandler() } fetcher.Setup(newController) + t.Cleanup(func() { fetcher.Close(); closeClient() }) return fetcher } // TestFetcher_Patch tests the Patch functionality for BT fetcher. // It tests modifying selected files after Resolve (without downloading). func TestFetcher_Patch(t *testing.T) { - f := buildFetcher() + f := buildFetcher(t) // Resolve a multi-file torrent err := f.Resolve(&base.Request{ URL: "./testdata/test.torrent", - }, nil) + }) if err != nil { t.Fatalf("Resolve failed: %v", err) } - // Verify initial state: 3 files, all selected by default + // Resolution describes files without choosing download options. meta := f.Meta() if len(meta.Res.Files) != 3 { t.Fatalf("Expected 3 files, got %d", len(meta.Res.Files)) } - if len(meta.Opts.SelectFiles) != 3 { - t.Fatalf("Expected 3 selected files, got %d", len(meta.Opts.SelectFiles)) + if meta.Opts != nil { + t.Fatal("Resolve should not assign download options") } // Total size of all files: 107484864 @@ -395,7 +397,7 @@ func TestFetcher_PeerDiscoveryDisabledOnClient(t *testing.T) { t.Fatal(err) } f := buildFetcherWithConfig(map[string]any{"disableDht": true, "disablePex": true}) - if err := f.Resolve(&base.Request{URL: "./testdata/test.torrent"}, nil); err != nil { + if err := f.Resolve(&base.Request{URL: "./testdata/test.torrent"}); err != nil { t.Fatal(err) } if !cfg.NoDHT || !cfg.DisablePEX { diff --git a/internal/protocol/bt/fixture_test.go b/internal/protocol/bt/fixture_test.go new file mode 100644 index 00000000..a8b1fe18 --- /dev/null +++ b/internal/protocol/bt/fixture_test.go @@ -0,0 +1,212 @@ +package bt + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/binary" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/anacrolix/torrent" + "github.com/anacrolix/torrent/bencode" + "github.com/anacrolix/torrent/merkle" + "github.com/anacrolix/torrent/metainfo" + "golang.org/x/time/rate" +) + +// btFixture serves generated, trackerless content over a real loopback TCP peer. +// Its HTTP tracker exercises .torrent downloads; magnet links also carry the +// explicit x.pe address. +type btFixture struct { + MetaInfo *metainfo.MetaInfo + Magnet string + TorrentFile string + DataURI string + TrackerURL string + Files map[string][]byte // Paths relative to the selected download directory. + Seeder *torrent.Client + Torrent *torrent.Torrent + Limiter *rate.Limiter +} + +func newBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int) *btFixture { + return makeBTFixture(t, files, bytesPerSecond, false, true) +} + +func newBTHybridFixture(t testing.TB, payload []byte) *btFixture { + return makeBTFixture(t, map[string][]byte{"": payload}, 0, true, true) +} + +func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool) *btFixture { + t.Helper() + source := t.TempDir() + name := "payload.bin" + if _, single := files[""]; !single { + name = "bundle" + } + root := filepath.Join(source, name) + expected := make(map[string][]byte, len(files)) + for path, data := range files { + target := filepath.Join(root, filepath.FromSlash(path)) + if err := os.MkdirAll(filepath.Dir(target), 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(target, data, 0600); err != nil { + t.Fatal(err) + } + expected[filepath.Join(name, filepath.FromSlash(path))] = append([]byte(nil), data...) + } + info := metainfo.Info{PieceLength: 16 * 1024} + if err := info.BuildFromFilePath(root); err != nil { + t.Fatal(err) + } + var layers map[string]string + if v2 { + data := files[""] + h := merkle.NewHash() + h.Write(data) + root := h.Sum(nil) + var pieceHashes []byte + for off := 0; off < len(data); off += int(info.PieceLength) { + end := min(off+int(info.PieceLength), len(data)) + ph := merkle.NewHash() + ph.Write(data[off:end]) + pieceHashes = append(pieceHashes, ph.SumMinLength(nil, int(info.PieceLength))...) + } + info = metainfo.Info{Name: name, PieceLength: info.PieceLength, Length: int64(len(data)), Pieces: info.Pieces, MetaVersion: 2, FileTree: metainfo.FileTree{File: metainfo.FileTreeFile{Length: int64(len(data)), PiecesRoot: string(root)}}} + layers = map[string]string{string(root): string(pieceHashes)} + } + // Private torrents avoid public DHT discovery once metadata is available. + info.Private = &private + raw, err := bencode.Marshal(&info) + if err != nil { + t.Fatal(err) + } + mi := &metainfo.MetaInfo{InfoBytes: raw, PieceLayers: layers} + cfg := torrent.TestingConfig(t) + cfg.DataDir = source + cfg.Seed = true + cfg.DisableUTP = true + cfg.DisableIPv6 = true + cfg.DisableWebtorrent = true + cfg.KeepAliveTimeout = time.Minute + cfg.MaxAllocPeerRequestDataPerConn = 1 << 20 + limiter := rate.NewLimiter(rate.Inf, 16*1024) + if bytesPerSecond > 0 { + limiter.SetLimit(rate.Limit(bytesPerSecond)) + } + cfg.UploadRateLimiter = limiter + seed, err := torrent.NewClient(cfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { seed.Close() }) + seeded, _, err := seed.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi)) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := seeded.VerifyDataContext(ctx); err != nil { + t.Fatal(err) + } + select { + case <-seeded.Complete().On(): + case <-ctx.Done(): + t.Fatal("local seeder did not finish verification") + } + addr := seed.ListenAddrs()[0].String() + host, portString, err := net.SplitHostPort(addr) + if err != nil { + t.Fatal(err) + } + resolved, err := net.ResolveTCPAddr("tcp4", net.JoinHostPort(host, portString)) + if err != nil { + t.Fatal(err) + } + compact := make([]byte, 6) + copy(compact, resolved.IP.To4()) + binary.BigEndian.PutUint16(compact[4:], uint16(resolved.Port)) + tracker := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + response, err := bencode.Marshal(map[string]any{"interval": 3600, "peers": string(compact)}) + if err != nil { + http.Error(w, err.Error(), 500) + return + } + w.Write(response) + })) + t.Cleanup(tracker.Close) + // The pinned library has a hybrid tracker-dispatch assertion bug. + // Hybrid fixtures use their explicit loopback peer rather than a tracker. + if !v2 { + mi.Announce = tracker.URL + } + var encoded bytes.Buffer + if err := mi.Write(&encoded); err != nil { + t.Fatal(err) + } + torrentFile := filepath.Join(t.TempDir(), "fixture.torrent") + if err := os.WriteFile(torrentFile, encoded.Bytes(), 0600); err != nil { + t.Fatal(err) + } + m, err := mi.MagnetV2() + if err != nil { + t.Fatal(err) + } + magnet := m.String() + u, err := url.Parse(magnet) + if err != nil { + t.Fatal(err) + } + q := u.Query() + q.Add("x.pe", addr) + u.RawQuery = q.Encode() + return &btFixture{MetaInfo: mi, Magnet: u.String(), TorrentFile: torrentFile, DataURI: "data:application/x-bittorrent;base64," + base64.StdEncoding.EncodeToString(encoded.Bytes()), TrackerURL: tracker.URL, Files: expected, Seeder: seed, Torrent: seeded, Limiter: limiter} +} + +func btPayload(size int, label string) []byte { + data := make([]byte, size) + for i := range data { + data[i] = byte((i*31 + i/251) % 251) + } + copy(data, label) + return data +} + +func (f *btFixture) AssertFiles(t testing.TB, dir string, paths ...string) { + t.Helper() + if len(paths) == 0 { + for path := range f.Files { + paths = append(paths, path) + } + } + for _, path := range paths { + got, err := os.ReadFile(filepath.Join(dir, path)) + if err != nil { + t.Fatal(err) + } + want, ok := f.Files[path] + if !ok { + t.Fatalf("unknown fixture file %q", path) + } + if !bytes.Equal(got, want) { + t.Fatalf("BT file %s differs: received %d bytes, want %d", path, len(got), len(want)) + } + } +} + +func (f *btFixture) FileURI() string { + path := filepath.ToSlash(f.TorrentFile) + if !strings.HasPrefix(path, "/") { + path = "/" + path + } + return (&url.URL{Scheme: "file", Path: path}).String() +} diff --git a/internal/protocol/bt/selection_test.go b/internal/protocol/bt/selection_test.go index b5197f63..e17e8f78 100644 --- a/internal/protocol/bt/selection_test.go +++ b/internal/protocol/bt/selection_test.go @@ -17,25 +17,85 @@ import ( "github.com/anacrolix/torrent/storage" ) -func TestStartUsesSelectionSubmittedAfterResolve(t *testing.T) { - config := torrent.TestingConfig(t) - config.DataDir = t.TempDir() - config.DisableTCP = true - config.DisableUTP = true - c, err := torrent.NewClient(config) +func TestStartDoesNotPublishUnverifiedCompletion(t *testing.T) { + cfg := torrent.TestingConfig(t) + cfg.DisableTCP = true + cfg.DisableUTP = true + c, err := torrent.NewClient(cfg) if err != nil { t.Fatal(err) } defer c.Close() - tor, err := c.AddTorrentFromFile("testdata/test.torrent") + payload := btPayload(32*1024, "unverified") + info := metainfo.Info{Name: "payload.bin", Length: int64(len(payload)), PieceLength: 16 * 1024} + for offset := 0; offset < len(payload); offset += int(info.PieceLength) { + hash := sha1.Sum(payload[offset : offset+int(info.PieceLength)]) + info.Pieces = append(info.Pieces, hash[:]...) + } + raw, err := bencode.Marshal(info) if err != nil { t.Fatal(err) } - f := &Fetcher{torrent: tor, meta: &fetcher.FetcherMeta{Opts: &base.Options{}}, data: &fetcherData{}} + store := &deferredFileStorage{} + spec := torrent.TorrentSpecFromMetaInfo(&metainfo.MetaInfo{InfoBytes: raw}) + spec.Storage = store + tor, _, err := c.AddTorrentSpec(spec) + if err != nil { + t.Fatal(err) + } + tor.DisallowDataDownload() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f := &Fetcher{torrent: tor, storage: store, meta: &fetcher.FetcherMeta{}, data: &fetcherData{}, torrentDropCtx: ctx, torrentDropFunc: cancel, config: &config{SeedKeep: true}} f.torrentReady.Store(true) f.updateRes() - // Resolve initializes all files; creation replaces the selection before Start. - f.meta.Opts = &base.Options{SelectFiles: []int{2}} + dir := t.TempDir() + payload[20000] ^= 0xff + if err := os.WriteFile(filepath.Join(dir, info.Name), payload, 0600); err != nil { + t.Fatal(err) + } + f.meta.Opts = &base.Options{Path: dir} + // Retain the original event channel to detect even a transient completion. + // With no network transports, the damaged piece cannot have been repaired. + completed := tor.Complete().On() + if err := f.Start(); err != nil { + t.Fatal(err) + } + select { + case <-completed: + t.Fatal("Start published completion before verifying the existing file") + default: + } +} + +func TestStartUsesSelectionSubmittedAfterResolve(t *testing.T) { + cfg := torrent.TestingConfig(t) + cfg.DataDir = t.TempDir() + cfg.DisableTCP = true + cfg.DisableUTP = true + c, err := torrent.NewClient(cfg) + if err != nil { + t.Fatal(err) + } + defer c.Close() + store := &deferredFileStorage{} + mi, err := metainfo.LoadFromFile("testdata/test.torrent") + if err != nil { + t.Fatal(err) + } + spec := torrent.TorrentSpecFromMetaInfo(mi) + spec.Storage = store + tor, _, err := c.AddTorrentSpec(spec) + if err != nil { + t.Fatal(err) + } + tor.DisallowDataDownload() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f := &Fetcher{torrent: tor, storage: store, meta: &fetcher.FetcherMeta{}, data: &fetcherData{}, torrentDropCtx: ctx, torrentDropFunc: cancel, config: &config{SeedKeep: true}} + f.torrentReady.Store(true) + f.updateRes() + f.meta.Opts = &base.Options{Path: t.TempDir(), SelectFiles: []int{2}} if err := f.Start(); err != nil { t.Fatal(err) } diff --git a/internal/protocol/bt/storage.go b/internal/protocol/bt/storage.go new file mode 100644 index 00000000..35caebe6 --- /dev/null +++ b/internal/protocol/bt/storage.go @@ -0,0 +1,116 @@ +package bt + +import ( + "context" + "errors" + "os" + "sync" + + g "github.com/anacrolix/generics" + "github.com/anacrolix/torrent/metainfo" + "github.com/anacrolix/torrent/storage" +) + +// deferredFileStorage lets metadata resolve without opening the user's files. +// It binds once. After binding, pieces are supplied directly by file storage. +type deferredFileStorage struct { + mu sync.RWMutex + ctx context.Context + info *metainfo.Info + hash metainfo.Hash + knownHashes []bool + owner storage.ClientImplCloser + torrent storage.TorrentImpl + closed bool +} + +func (s *deferredFileStorage) OpenTorrent(ctx context.Context, info *metainfo.Info, hash metainfo.Hash) (storage.TorrentImpl, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed || s.info != nil { + return storage.TorrentImpl{}, errors.New("torrent storage already opened or closed") + } + s.ctx, s.info, s.hash = ctx, info, hash + s.knownHashes = make([]bool, info.NumPieces()) + return storage.TorrentImpl{ + PieceWithHash: func(p metainfo.Piece, hash g.Option[[]byte]) storage.PieceImpl { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return metadataPiece{err: os.ErrClosed} + } + if s.owner == nil { + s.knownHashes[p.Index()] = hash.Ok + // A missing v2 hash must remain unknown until its arrival changes + // completion and updates the library's piece request priorities. + return metadataPiece{known: hash.Ok} + } + if s.torrent.PieceWithHash != nil { + return s.torrent.PieceWithHash(p, hash) + } + return s.torrent.Piece(p) + }, + Close: s.Close, + }, nil +} + +func (s *deferredFileStorage) bind(opts storage.NewFileClientOpts) ([]bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return nil, os.ErrClosed + } + if s.info == nil { + return nil, errors.New("torrent metadata is not ready") + } + if s.owner != nil { + return nil, nil + } + if opts.FilePathMaker == nil { + opts.FilePathMaker = func(o storage.FilePathMakerOpts) string { _, name := torrentFileLayout(o.Info, *o.File); return name } + } + owner := storage.NewFileOpts(opts) + t, err := owner.OpenTorrent(s.ctx, s.info, s.hash) + if err != nil { + owner.Close() + return nil, err + } + s.owner, s.torrent = owner, t + return append([]bool(nil), s.knownHashes...), nil +} + +func (s *deferredFileStorage) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return nil + } + s.closed = true + if s.owner == nil { + return nil + } + var err error + if s.torrent.Close != nil { + err = s.torrent.Close() + } + return errors.Join(err, s.owner.Close()) +} + +type metadataPiece struct { + known bool + err error +} + +func (p metadataPiece) Completion() storage.Completion { + return storage.Completion{Ok: p.known, Err: p.err} +} +func (p metadataPiece) ReadAt([]byte, int64) (int, error) { + return 0, errors.New("download storage is not initialized") +} +func (p metadataPiece) WriteAt([]byte, int64) (int, error) { + return 0, errors.New("download storage is not initialized") +} +func (p metadataPiece) MarkComplete() error { return errors.New("download storage is not initialized") } +func (p metadataPiece) MarkNotComplete() error { + return errors.New("download storage is not initialized") +} diff --git a/internal/protocol/ed2k/fetcher.go b/internal/protocol/ed2k/fetcher.go index 81da7ef1..e7be40a5 100644 --- a/internal/protocol/ed2k/fetcher.go +++ b/internal/protocol/ed2k/fetcher.go @@ -67,17 +67,14 @@ func (f *Fetcher) Setup(ctl *controller.Controller) { f.ctl.GetConfig(&f.config) } -func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) error { +func (f *Fetcher) Resolve(req *base.Request) error { link, err := parseLink(req.URL) if err != nil { return err } f.meta.Req = req - f.meta.Opts = opts - if f.meta.Opts == nil { - f.meta.Opts = &base.Options{} - } + f.meta.Res = buildResource(link) return nil } diff --git a/internal/protocol/ed2k/fetcher_test.go b/internal/protocol/ed2k/fetcher_test.go index a613a21d..ddc11c5e 100644 --- a/internal/protocol/ed2k/fetcher_test.go +++ b/internal/protocol/ed2k/fetcher_test.go @@ -13,7 +13,8 @@ func TestFetcher_Resolve(t *testing.T) { f := (&FetcherManager{}).Build() f.Setup(controller.NewController()) - err := f.Resolve(&base.Request{URL: testLink}, &base.Options{Path: t.TempDir()}) + f.Meta().Opts = &base.Options{Path: t.TempDir()} + err := f.Resolve(&base.Request{URL: testLink}) if err != nil { t.Fatalf("Resolve() error = %v", err) } diff --git a/internal/protocol/hls/fetcher.go b/internal/protocol/hls/fetcher.go index 7259a9b0..5501502d 100644 --- a/internal/protocol/hls/fetcher.go +++ b/internal/protocol/hls/fetcher.go @@ -172,11 +172,11 @@ func (f *Fetcher) Meta() *fetcher.FetcherMeta { return f.meta } -func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) (err error) { +func (f *Fetcher) Resolve(req *base.Request) (err error) { f.mu.Lock() defer f.mu.Unlock() f.meta.Req = req - f.meta.Opts = opts + f.extra = &ReqExtra{} if err = base.ParseReqExtra[ReqExtra](req); err != nil { return @@ -194,10 +194,6 @@ func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) (err error) { } state.StagingID = randomStagingID() f.state = state - // Pin the staging folder now so the very first persisted state already - // carries it: a crash right after create still resumes into the same - // folder. - f.resolveStagingDirLocked() size := planTotalSize(state.Segments) f.meta.Res = &base.Resource{ Range: false, diff --git a/internal/protocol/hls/fetcher_test.go b/internal/protocol/hls/fetcher_test.go index 3dd785f0..de11afd3 100644 --- a/internal/protocol/hls/fetcher_test.go +++ b/internal/protocol/hls/fetcher_test.go @@ -80,7 +80,8 @@ func mustResolve(t *testing.T, f *Fetcher, playlistURL, referer string) { if referer != "" { req.Extra = &ReqExtra{Header: map[string]string{"Referer": referer}} } - if err := f.Resolve(req, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(req); err != nil { t.Fatalf("resolve: %v", err) } } @@ -207,7 +208,8 @@ func TestFetcherE2E_RefererEnforced(t *testing.T) { // Without Referer the resolve itself must fail (403 on the playlist). blocked := newTestFetcher(t, testConfig()) req := &base.Request{URL: server.URL + "/video.m3u8"} - if err := blocked.Resolve(req, &base.Options{Path: t.TempDir()}); err == nil { + blocked.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := blocked.Resolve(req); err == nil { t.Fatal("resolve without Referer should fail") } @@ -217,7 +219,8 @@ func TestFetcherE2E_RefererEnforced(t *testing.T) { URL: server.URL + "/video.m3u8", Extra: &ReqExtra{Header: map[string]string{"Referer": page}}, } - if err := allowed.Resolve(allowedReq, &base.Options{Path: t.TempDir()}); err != nil { + allowed.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := allowed.Resolve(allowedReq); err != nil { t.Fatal(err) } mustRun(t, allowed) @@ -560,7 +563,8 @@ func TestFetcherE2E_StoreRestoreAuthenticated(t *testing.T) { URL: server.URL + "/video.m3u8", Extra: &ReqExtra{Header: map[string]string{"Referer": page}}, } - if err := f.Resolve(req, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(req); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { @@ -630,7 +634,8 @@ func TestFetcherE2E_PatchPartialUpdate(t *testing.T) { Extra: &ReqExtra{Header: map[string]string{"Referer": page}}, } opts := &base.Options{Path: t.TempDir()} - if err := f.Resolve(req, opts); err != nil { + f.Meta().Opts = opts + if err := f.Resolve(req); err != nil { t.Fatal(err) } @@ -713,7 +718,8 @@ func TestFetcherE2E_PatchWhileRunningRejected(t *testing.T) { defer server.Close() f := newTestFetcher(t, testConfig()) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { @@ -767,7 +773,8 @@ func TestFetcherE2E_RetryAfterFailure(t *testing.T) { cfg.MaxRetries = 0 // fail the run on the first bad segment cfg.PrefetchContentLength = false f := newTestFetcher(t, cfg) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { @@ -828,7 +835,8 @@ func TestFetcherE2E_StateMissingSelfHeal(t *testing.T) { defer server.Close() f := newTestFetcher(t, testConfig()) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } // Simulate the crash window where the engine persisted the task with its diff --git a/internal/protocol/hls/media_validate_test.go b/internal/protocol/hls/media_validate_test.go index 276d2fd0..91e6741f 100644 --- a/internal/protocol/hls/media_validate_test.go +++ b/internal/protocol/hls/media_validate_test.go @@ -189,7 +189,8 @@ func TestHLSRealMedia_TS(t *testing.T) { defer server.Close() f := newTestFetcher(t, testConfig()) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { @@ -225,7 +226,8 @@ func TestHLSRealMedia_FMP4(t *testing.T) { defer server.Close() f := newTestFetcher(t, testConfig()) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { @@ -274,7 +276,8 @@ func TestHLSRealMedia_AES128(t *testing.T) { defer server.Close() f := newTestFetcher(t, testConfig()) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } if f.meta.Res.Size != 0 { diff --git a/internal/protocol/hls/staging_test.go b/internal/protocol/hls/staging_test.go index ec25acbf..ada03d0a 100644 --- a/internal/protocol/hls/staging_test.go +++ b/internal/protocol/hls/staging_test.go @@ -50,15 +50,20 @@ func TestFetcherE2E_SameURLTwoTasks(t *testing.T) { // user duplicating a task. The engine would rename the second output // (AutoRename); mirror that here. f1 := newTestFetcher(t, cfg) - if err := f1.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: dir}); err != nil { + f1.Meta().Opts = &base.Options{Path: dir} + if err := f1.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } f2 := newTestFetcher(t, cfg) - if err := f2.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: dir, Name: "video-2.ts"}); err != nil { + f2.Meta().Opts = &base.Options{Path: dir, Name: "video-2.ts"} + if err := f2.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } - if f1.state.StagingDir == "" || f1.state.StagingDir == f2.state.StagingDir { - t.Fatalf("staging dirs must be distinct per task: %q vs %q", f1.state.StagingDir, f2.state.StagingDir) + if f1.state.StagingID == "" || f1.state.StagingID == f2.state.StagingID { + t.Fatal("staging identities must be distinct per task") + } + if f1.state.StagingDir != "" || f2.state.StagingDir != "" { + t.Fatal("Resolve pinned an output directory") } // Start task 1, let it stage some segments, then pause it. @@ -135,7 +140,8 @@ func TestFetcherE2E_ForeignJournalDiscarded(t *testing.T) { cfg := testConfig() cfg.PrefetchContentLength = false f := newTestFetcher(t, cfg) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } // Plant a journal that does not belong to this plan (wrong fingerprint). @@ -176,7 +182,8 @@ func TestFetcherE2E_MergeFailurePreservesStaging(t *testing.T) { defer server.Close() f := newTestFetcher(t, testConfig()) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } run := &hlsRun{ @@ -250,7 +257,8 @@ func TestFetcherE2E_ReMergeOverExistingOutput(t *testing.T) { defer server.Close() f := newTestFetcher(t, testConfig()) - if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}, &base.Options{Path: t.TempDir()}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir()} + if err := f.Resolve(&base.Request{URL: server.URL + "/video.m3u8"}); err != nil { t.Fatal(err) } tempDir := f.resolveStagingDirLocked() diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 98f7e0c4..a2b850c0 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -372,19 +372,11 @@ func (f *Fetcher) updateMaxConnTime(d time.Duration) { } } -func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) error { +func (f *Fetcher) Resolve(req *base.Request) error { if err := base.ParseReqExtra[fhttp.ReqExtra](req); err != nil { return err } f.meta.Req = req - f.meta.Opts = opts - if f.meta.Opts == nil { - f.meta.Opts = &base.Options{} - } - - if err := f.initOptions(); err != nil { - return err - } f.setState(stateResolving) @@ -693,6 +685,9 @@ func (f *Fetcher) prepareFullRedownload() { // Options may be replaced between Resolve and Start when creating a resolved task. func (f *Fetcher) initOptions() error { + if f.meta.Opts == nil { + f.meta.Opts = &base.Options{} + } opts := f.meta.Opts // Parse options if err := base.ParseOptExtra[fhttp.OptsExtra](opts); err != nil { diff --git a/internal/protocol/http/fetcher_checksum_retry_test.go b/internal/protocol/http/fetcher_checksum_retry_test.go index 98439fed..b4298585 100644 --- a/internal/protocol/http/fetcher_checksum_retry_test.go +++ b/internal/protocol/http/fetcher_checksum_retry_test.go @@ -95,9 +95,10 @@ func TestFetcher_ChecksumMismatch_Retry(t *testing.T) { } // 1. Initial attempt -> will download corruptContent, failing checksum verification - if err := f.Resolve(&base.Request{URL: server.URL + "/retry_test.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/retry_test.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } @@ -215,9 +216,10 @@ func TestFetcher_ChecksumMismatch_Retry_PartialPrefetch(t *testing.T) { } // 1. Resolve phase - if err := f.Resolve(&base.Request{URL: server.URL + "/prefetch_retry.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/prefetch_retry.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + f.Meta().Opts = opts // Wait until prefetch has received exactly 7 bytes for i := 0; i < 200; i++ { @@ -338,9 +340,10 @@ func TestFetcher_ChecksumMismatch_Retry_NonRange(t *testing.T) { } // 1. Resolve and start initial attempt - if err := f.Resolve(&base.Request{URL: server.URL + "/nonrange_retry.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/nonrange_retry.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } diff --git a/internal/protocol/http/fetcher_checksum_test.go b/internal/protocol/http/fetcher_checksum_test.go index 8298f4eb..b51326c0 100644 --- a/internal/protocol/http/fetcher_checksum_test.go +++ b/internal/protocol/http/fetcher_checksum_test.go @@ -52,9 +52,11 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } @@ -79,9 +81,11 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } @@ -106,9 +110,11 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } @@ -137,9 +143,13 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}, opts) + if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}); err != nil { + t.Fatalf("Resolve error: %v", err) + } + f.Meta().Opts = opts + err := f.Start() if err == nil { - t.Fatal("Resolve expected error for unsupported algorithm, got nil") + t.Fatal("Start expected error for unsupported algorithm, got nil") } if !strings.Contains(err.Error(), "unsupported checksum algorithm") { t.Fatalf("expected 'unsupported checksum algorithm' in error, got %v", err) @@ -165,9 +175,11 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } @@ -192,9 +204,11 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } @@ -235,9 +249,11 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - if err := f.Resolve(&base.Request{URL: largeServer.URL + "/large.bin"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: largeServer.URL + "/large.bin"}); err != nil { t.Fatalf("Resolve error: %v", err) } + + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } @@ -258,9 +274,11 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}, opts); err != nil { + if err := f.Resolve(&base.Request{URL: server.URL + "/test.txt"}); err != nil { t.Fatalf("Resolve error: %v", err) } + + f.Meta().Opts = opts if err := f.Start(); err != nil { t.Fatalf("Start error: %v", err) } @@ -269,7 +287,7 @@ func TestFetcher_ChecksumVerification(t *testing.T) { } }) - t.Run("Invalid Checksum In Resolve Makes Zero Requests", func(t *testing.T) { + t.Run("Invalid Checksum In Start Makes No Additional Requests", func(t *testing.T) { var reqReceived atomic.Int32 customServer := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) { reqReceived.Add(1) @@ -293,12 +311,17 @@ func TestFetcher_ChecksumVerification(t *testing.T) { }, } - err := f.Resolve(&base.Request{URL: customServer.URL + "/test.txt"}, opts) - if err == nil { - t.Fatal("Resolve expected error for invalid checksum hash, got nil") + if err := f.Resolve(&base.Request{URL: customServer.URL + "/test.txt"}); err != nil { + t.Fatalf("Resolve error: %v", err) } - if reqReceived.Load() != 0 { - t.Fatalf("expected 0 requests to server on invalid checksum in Resolve, got %d", reqReceived.Load()) + requestsAfterResolve := reqReceived.Load() + f.Meta().Opts = opts + err := f.Start() + if err == nil { + t.Fatal("Start expected error for invalid checksum hash, got nil") + } + if got := reqReceived.Load(); got != requestsAfterResolve { + t.Fatalf("invalid checksum in Start issued additional requests: got %d, had %d after Resolve", got, requestsAfterResolve) } }) } diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index 0077f4a0..0e9445de 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -233,10 +233,11 @@ func TestFetcher_ResolveFilenameFromContentType(t *testing.T) { defer fetcher.Pause() const opaqueName = "OIP-C.u5nEXTXND3PJF0gNTgw5YgHaFP" - err := fetcher.Resolve(&base.Request{URL: server.URL + "/" + opaqueName}, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, - }) + } + err := fetcher.Resolve(&base.Request{URL: server.URL + "/" + opaqueName}) if err != nil { t.Fatal(err) } @@ -251,6 +252,10 @@ func TestFetcher_ResolveWithHostHeader(t *testing.T) { defer listener.Close() fetcher := buildFetcher() + fetcher.Meta().Opts = &base.Options{ + Name: test.DownloadName, + Path: test.Dir, + } err := fetcher.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/", Extra: &http.ReqExtra{ @@ -258,9 +263,6 @@ func TestFetcher_ResolveWithHostHeader(t *testing.T) { "Host": "test", }, }, - }, &base.Options{ - Name: test.DownloadName, - Path: test.Dir, }) // The server should return 400 for invalid Host header if err == nil || !strings.Contains(err.Error(), "400") { @@ -274,6 +276,10 @@ func TestFetcher_ResolveWithInvalidHeader(t *testing.T) { fetcher := buildFetcher() defer fetcher.Pause() // Close the resolve response to allow server shutdown + fetcher.Meta().Opts = &base.Options{ + Name: test.DownloadName, + Path: test.Dir, + } err := fetcher.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/", Extra: &http.ReqExtra{ @@ -281,9 +287,6 @@ func TestFetcher_ResolveWithInvalidHeader(t *testing.T) { "Referer": "\rtest", }, }, - }, &base.Options{ - Name: test.DownloadName, - Path: test.Dir, }) // Invalid header with \r should be sanitized by Go's http client, allowing the request to succeed if err != nil { @@ -303,12 +306,13 @@ func TestFetcher_ResolveAutomaticallyFallsBackToBrowserFingerprint(t *testing.T) fetcher := buildFetcher() defer fetcher.Pause() + fetcher.Meta().Opts = &base.Options{ + Name: test.DownloadName, + Path: test.Dir, + } err := fetcher.Resolve(&base.Request{ URL: server.URL + "/file.bin", SkipVerifyCert: true, - }, &base.Options{ - Name: test.DownloadName, - Path: test.Dir, }) if err != nil { t.Fatalf("Resolve() error = %v", err) @@ -369,11 +373,12 @@ func testResolve(startTestServer func() net.Listener, path string, t *testing.T, defer listener.Close() fetcher := buildFetcher() defer fetcher.Pause() // Close the resolve response to allow server shutdown - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + path, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + path, }) want, err := wantFn(err) if err != nil { @@ -425,13 +430,14 @@ func TestFetcher_PauseBeforeStartDiscardsPrefetchProgress(t *testing.T) { defer server.Close() fetcher := buildFetcher() - if err := fetcher.Resolve(&base.Request{URL: server.URL + "/" + test.BuildName}, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: test.DownloadName, Extra: &http.OptsExtra{ Connections: 2, }, - }); err != nil { + } + if err := fetcher.Resolve(&base.Request{URL: server.URL + "/" + test.BuildName}); err != nil { t.Fatal(err) } select { @@ -772,13 +778,14 @@ func TestFetcher_SequentialRetryReprobesRange(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/dynamic-range.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "dynamic-range.data", Extra: &http.OptsExtra{ Connections: 4, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/dynamic-range.data"}); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { @@ -884,13 +891,14 @@ func TestFetcher_InitialIgnoredRangeUsesResponseSize(t *testing.T) { defer releaseResolve() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/ignored-range.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "ignored-range.data", Extra: &http.OptsExtra{ Connections: 4, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/ignored-range.data"}); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { @@ -981,13 +989,14 @@ func TestFetcher_InterruptedChunkedReplacementRestartsFromZero(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/chunked-replacement.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "chunked-replacement.data", Extra: &http.OptsExtra{ Connections: 4, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/chunked-replacement.data"}); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { @@ -1167,13 +1176,14 @@ func TestFetcher_UsesResolveResponseWhenRangeTakeoverFails(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/resolve-fallback.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "resolve-fallback.data", Extra: &http.OptsExtra{ Connections: 1, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/resolve-fallback.data"}); err != nil { t.Fatal(err) } defer f.Close() @@ -1246,13 +1256,14 @@ func TestFetcher_ValidRangeTakesOverResolveResponse(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/resolve-takeover.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "resolve-takeover.data", Extra: &http.OptsExtra{ Connections: 4, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/resolve-takeover.data"}); err != nil { t.Fatal(err) } defer f.Close() @@ -1340,13 +1351,14 @@ func TestFetcher_UnknownSizeRangeUsesOpenEndedWorker(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size-range.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "unknown-size-range.data", Extra: &http.OptsExtra{ Connections: 4, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size-range.data"}); err != nil { t.Fatal(err) } defer f.Close() @@ -1434,13 +1446,14 @@ func TestFetcher_UnknownSizeRangeRetriesShortPartialBody(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/short-range.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "short-range.data", Extra: &http.OptsExtra{ Connections: 1, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/short-range.data"}); err != nil { t.Fatal(err) } defer f.Close() @@ -1493,13 +1506,14 @@ func TestFetcher_UnknownSizeRangeFallsBackToResolve(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size-fallback.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "unknown-size-fallback.data", Extra: &http.OptsExtra{ Connections: 1, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size-fallback.data"}); err != nil { t.Fatal(err) } defer f.Close() @@ -1580,13 +1594,14 @@ func TestFetcher_UnknownSizeRangeRemainsOpenEndedAfterPause(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/pause-open-range.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "pause-open-range.data", Extra: &http.OptsExtra{ Connections: 1, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/pause-open-range.data"}); err != nil { t.Fatal(err) } defer f.Close() @@ -1635,13 +1650,14 @@ func TestFetcher_UnknownSizeResolveReadErrorFailsTask(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/broken-chunked.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "broken-chunked.data", Extra: &http.OptsExtra{ Connections: 1, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/broken-chunked.data"}); err != nil { t.Fatal(err) } defer f.Close() @@ -1667,13 +1683,14 @@ func TestFetcher_UnknownSizeResolveCleanEOFUpdatesSize(t *testing.T) { defer server.Close() f := buildFetcher() - if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size.data"}, &base.Options{ + f.Meta().Opts = &base.Options{ Path: t.TempDir(), Name: "unknown-size.data", Extra: &http.OptsExtra{ Connections: 1, }, - }); err != nil { + } + if err := f.Resolve(&base.Request{URL: server.URL + "/unknown-size.data"}); err != nil { t.Fatal(err) } defer f.Close() @@ -2269,14 +2286,15 @@ func TestFetcher_DownloadOneTimeURL(t *testing.T) { defer listener.Close() fetcher := buildFetcher() - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, Extra: &http.OptsExtra{ Connections: 4, // Try to use multiple connections, but only first should work }, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, }) if err != nil { t.Fatal(err) @@ -2335,14 +2353,15 @@ func TestFetcher_SlowStartExpansion(t *testing.T) { Connections: tc.maxConns, }) - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, Extra: &http.OptsExtra{ Connections: tc.maxConns, }, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, }) if err != nil { cleanup() @@ -2398,14 +2417,15 @@ func TestFetcher_AsyncPrefetch(t *testing.T) { defer listener.Close() fetcher := buildFetcher() - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, Extra: &http.OptsExtra{ Connections: 4, }, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, }) if err != nil { t.Fatal(err) @@ -2469,14 +2489,15 @@ func TestFetcher_AsyncPrefetch(t *testing.T) { defer listener.Close() fetcher := buildFetcher() - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, Extra: &http.OptsExtra{ Connections: 4, }, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, }) if err != nil { t.Fatal(err) @@ -2551,14 +2572,15 @@ func TestFetcher_DownloadExpiringRedirectURL(t *testing.T) { defer listener.Close() fetcher := buildFetcher() - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, Extra: &http.OptsExtra{ Connections: 4, // Use multiple connections to trigger redirect expiration }, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, }) if err != nil { t.Fatal(err) @@ -2592,14 +2614,15 @@ func TestFetcher_DownloadExpiringRedirectURL(t *testing.T) { defer listener.Close() fetcher := buildFetcher() - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, Extra: &http.OptsExtra{ Connections: 8, }, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, }) if err != nil { t.Fatal(err) @@ -2639,14 +2662,15 @@ func TestFetcher_RetryAfterError(t *testing.T) { defer listener.Close() fetcher := buildFetcher() - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, Extra: &http.OptsExtra{ Connections: 1, // Use single connection to simplify test }, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, }) if err != nil { t.Fatal(err) @@ -3067,9 +3091,10 @@ func doDownloadReady(f fetcher.Fetcher, listener net.Listener, connections int, Path: test.Dir, Extra: extra, } + f.Meta().Opts = opts err := f.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, opts) + }) if err != nil { t.Fatal(err) } @@ -3118,7 +3143,8 @@ func downloadPost(listener net.Listener, connections int, t *testing.T) { Body: fmt.Sprintf(`{"name":"%s"}`, test.BuildName), }, } - err := f.Resolve(req, opts) + f.Meta().Opts = opts + err := f.Resolve(req) if err != nil { t.Fatal(err) } @@ -3164,11 +3190,12 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { func downloadError(listener net.Listener, connections int, t *testing.T) { fetcher := buildFetcher() - err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, &base.Options{ + fetcher.Meta().Opts = &base.Options{ Name: test.DownloadName, Path: test.Dir, + } + err := fetcher.Resolve(&base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, }) // With the new async design, Resolve may succeed (HTTP response received) // but errors occur during async download or Start/Wait @@ -3292,7 +3319,8 @@ func TestFetcher_Patch_URLChange(t *testing.T) { } // Step 1: Try to resolve with bad URL - should fail with error - err := f.Resolve(&base.Request{URL: badURL}, opts) + f.Meta().Opts = opts + err := f.Resolve(&base.Request{URL: badURL}) if err == nil { t.Fatal("Expected error for bad URL, got nil") } @@ -3302,7 +3330,8 @@ func TestFetcher_Patch_URLChange(t *testing.T) { f2 := buildFetcher() // First resolve with good URL to create a valid fetcher state - err = f2.Resolve(&base.Request{URL: goodURL}, opts) + f2.Meta().Opts = opts + err = f2.Resolve(&base.Request{URL: goodURL}) if err != nil { t.Fatal(err) } @@ -3354,13 +3383,14 @@ func TestFetcher_Patch_Labels(t *testing.T) { Extra: &http.OptsExtra{Connections: 1}, } + f.Meta().Opts = opts err := f.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, Labels: map[string]string{ "key1": "value1", "key3": "value3", }, - }, opts) + }) if err != nil { t.Fatal(err) } @@ -3411,6 +3441,7 @@ func TestFetcher_Patch_Extra(t *testing.T) { } // Resolve with initial Extra + f.Meta().Opts = opts err := f.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, Extra: &http.ReqExtra{ @@ -3421,7 +3452,7 @@ func TestFetcher_Patch_Extra(t *testing.T) { "X-Custom": "original", }, }, - }, opts) + }) if err != nil { t.Fatal(err) } @@ -3499,9 +3530,10 @@ func TestFetcher_Patch_NilData(t *testing.T) { Extra: &http.OptsExtra{Connections: 1}, } + f.Meta().Opts = opts err := f.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, - }, opts) + }) if err != nil { t.Fatal(err) } @@ -3551,6 +3583,7 @@ func TestFetcher_Patch_CookieExpired(t *testing.T) { // Step 1: Resolve with old_token - should succeed (first request accepts old_token) f := buildFetcher() + f.Meta().Opts = opts err := f.Resolve(&base.Request{ URL: downloadURL, Extra: &http.ReqExtra{ @@ -3558,7 +3591,7 @@ func TestFetcher_Patch_CookieExpired(t *testing.T) { "Cookie": "session=old_token", }, }, - }, opts) + }) if err != nil { t.Fatalf("Resolve should succeed with old_token: %v", err) } diff --git a/internal/protocol/http/resolve_range_test.go b/internal/protocol/http/resolve_range_test.go index d7e0b335..29bc3158 100644 --- a/internal/protocol/http/resolve_range_test.go +++ b/internal/protocol/http/resolve_range_test.go @@ -49,7 +49,8 @@ func TestResolveResponseSize(t *testing.T) { defer server.Close() f := buildFetcher() defer f.Close() - err := f.Resolve(&base.Request{URL: server.URL + "/file"}, &base.Options{Path: t.TempDir()}) + f.Meta().Opts = &base.Options{Path: t.TempDir()} + err := f.Resolve(&base.Request{URL: server.URL + "/file"}) if tc.invalid { if !errors.Is(err, errInvalidRangeResponse) { t.Fatalf("Resolve error = %v, want invalid range", err) @@ -97,7 +98,8 @@ func TestDownloadPartialResolve(t *testing.T) { defer server.Close() f := buildFetcher() defer f.Close() - if err := f.Resolve(&base.Request{URL: server.URL + "/file"}, &base.Options{Path: t.TempDir(), Name: "file", Extra: &fhttp.OptsExtra{Connections: connections}}); err != nil { + f.Meta().Opts = &base.Options{Path: t.TempDir(), Name: "file", Extra: &fhttp.OptsExtra{Connections: connections}} + if err := f.Resolve(&base.Request{URL: server.URL + "/file"}); err != nil { t.Fatal(err) } if err := f.Start(); err != nil { diff --git a/pkg/api/service.go b/pkg/api/service.go index db888ca9..86a91353 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -139,7 +139,7 @@ func (s *Service) resolve(ctx *Context) *Response { if response := decodeRequest(ctx.Body, &request); response != nil { return response } - result, err := s.Downloader.Resolve(request.Req, request.Opts) + result, err := s.Downloader.Resolve(request.Req) if err != nil { return newErrorResponse(err.Error()) } @@ -188,7 +188,7 @@ func (s *Service) patchTask(ctx *Context) *Response { if taskID == "" { return invalidIDResponse() } - var request restmodel.ResolveTask + var request restmodel.PatchTask if response := decodeRequest(ctx.Body, &request); response != nil { return response } diff --git a/pkg/api/service_test.go b/pkg/api/service_test.go new file mode 100644 index 00000000..2562756c --- /dev/null +++ b/pkg/api/service_test.go @@ -0,0 +1,82 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "testing" + + "github.com/GopeedLab/gopeed/pkg/base" + "github.com/GopeedLab/gopeed/pkg/download" + restmodel "github.com/GopeedLab/gopeed/pkg/rest/model" +) + +func TestResolveThenCreateWithDownloadOptions(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Disposition", `attachment; filename="payload.txt"`) + w.Header().Set("Content-Length", "4") + _, _ = w.Write([]byte("test")) + })) + t.Cleanup(upstream.Close) + d := download.NewDownloader(&download.DownloaderConfig{StorageDir: t.TempDir()}) + if err := d.Setup(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := d.Close(); err != nil { + t.Error(err) + } + }) + // Resolving must work even when the default download directory is unusable. + blocked := filepath.Join(t.TempDir(), "not-a-directory") + if err := os.WriteFile(blocked, nil, 0600); err != nil { + t.Fatal(err) + } + current, err := d.GetConfig() + if err != nil { + t.Fatal(err) + } + config := *current + config.DownloadDir = blocked + config.MaxRunning = 0 // Inspect the created task before it starts downloading. + if err := d.PutConfig(&config); err != nil { + t.Fatal(err) + } + service, err := NewService(d) + if err != nil { + t.Fatal(err) + } + resolved := serviceTestRequest[*download.ResolveResult](t, service, http.MethodPost, "/api/v1/resolve", &restmodel.ResolveTask{ + Req: &base.Request{URL: upstream.URL}, + }) + if resolved.ID == "" || resolved.Res.Size != 4 || resolved.Res.Files[0].Name != "payload.txt" { + t.Fatalf("unexpected resolved resource: %+v", resolved) + } + dir := filepath.Join(t.TempDir(), "chosen-after-resolve") + opts := &base.Options{Path: dir, Name: "renamed.txt", SelectFiles: []int{0}} + id := serviceTestRequest[string](t, service, http.MethodPost, "/api/v1/tasks", &restmodel.CreateTask{Rid: resolved.ID, Opts: opts}) + task := serviceTestRequest[*download.Task](t, service, http.MethodGet, "/api/v1/tasks/"+id, nil) + if task.Meta.Opts.Path != dir || task.Meta.Opts.Name != opts.Name || !reflect.DeepEqual(task.Meta.Opts.SelectFiles, opts.SelectFiles) { + t.Fatalf("creation lost submitted download options: %+v", task.Meta.Opts) + } +} + +func serviceTestRequest[T any](t *testing.T, service *Service, method, path string, body any) T { + t.Helper() + encoded, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + status, response := service.JSON(method, path, nil, encoded, nil) + var result restmodel.Result[T] + if err := json.Unmarshal(response, &result); err != nil { + t.Fatal(err) + } + if status != http.StatusOK || result.Code != restmodel.CodeOk { + t.Fatalf("%s %s failed: status=%d response=%s", method, path, status, response) + } + return result.Data +} diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 89323144..f9bc2192 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -244,16 +244,25 @@ func (d *Downloader) Setup() error { tasks := append([]*Task(nil), d.tasks...) d.lock.Unlock() for _, task := range tasks { - if task.Status == base.DownloadStatusDone && task.Uploading { + func() { + task.lock.Lock() + defer task.lock.Unlock() + if d.closed.Load() { + return + } + if d.taskStatus(task) != base.DownloadStatusDone || !task.Uploading { + return + } if err := d.restoreTask(task); err != nil { - d.Logger.Error().Stack().Err(err).Msgf("task upload restore fetcher failed, task id: %s", task.ID) + d.Logger.Error().Err(err).Msgf("task upload restore fetcher failed, task id: %s", task.ID) + return } if uploader, ok := task.fetcher.(fetcher.Uploader); ok { if err := uploader.Upload(); err != nil { - d.Logger.Error().Stack().Err(err).Msgf("task upload failed, task id: %s", task.ID) + d.Logger.Error().Err(err).Msgf("task upload failed, task id: %s", task.ID) } } - } + }() } }() @@ -467,7 +476,10 @@ func (d *Downloader) saveTask(task *Task) error { return err } } - if err := d.storage.Put(bucketTask, task.ID, task); err != nil { + task.statusLock.Lock() + snapshot := task.clone() + task.statusLock.Unlock() + if err := d.storage.Put(bucketTask, task.ID, snapshot); err != nil { d.Logger.Error().Stack().Err(err).Msgf("persist task failed: %s", task.ID) return err } @@ -498,7 +510,7 @@ func ensureResourceRequestRawURLs(parentReq *base.Request, res *base.Resource) { } } -func (d *Downloader) Resolve(req *base.Request, opts *base.Options) (rr *ResolveResult, err error) { +func (d *Downloader) Resolve(req *base.Request) (rr *ResolveResult, err error) { rrId, err := gonanoid.New() if err != nil { return @@ -520,11 +532,7 @@ func (d *Downloader) Resolve(req *base.Request, opts *base.Options) (rr *Resolve if err != nil { return } - initOpt, err := d.initOptions(opts) - if err != nil { - return - } - err = fetcher.Resolve(req, initOpt) + err = fetcher.Resolve(req) if err != nil { return } @@ -558,7 +566,7 @@ func (d *Downloader) notifyRunning() { func (d *Downloader) remainRunningCount() int { runningCount := 0 for _, t := range d.tasks { - if t.Status == base.DownloadStatusRunning { + if d.taskStatus(t) == base.DownloadStatusRunning { runningCount++ } } @@ -621,7 +629,7 @@ func (d *Downloader) Create(rrId string) (taskId string, err error) { } // CreateWithOptions creates a resolved task using the submitted options, or the -// resolve-time options when opts is nil. +// default download options when opts is nil. func (d *Downloader) CreateWithOptions(rrId string, opts *base.Options) (taskId string, err error) { d.fetcherMapLock.RLock() fetcher, ok := d.fetcherCache[rrId] @@ -629,21 +637,22 @@ func (d *Downloader) CreateWithOptions(rrId string, opts *base.Options) (taskId if !ok { return "", errors.New("invalid resource id") } - if opts != nil { - opts, err = d.initOptions(opts.Clone()) - if err != nil { - return "", err - } - if res := fetcher.Meta().Res; res != nil { - for _, index := range opts.SelectFiles { - if index < 0 || index >= len(res.Files) { - return "", fmt.Errorf("invalid selected file index: %d", index) - } - } - opts.InitSelectFiles(len(res.Files)) - } - fetcher.Meta().Opts = opts + if opts == nil { + opts = &base.Options{} } + opts, err = d.initOptions(opts.Clone()) + if err != nil { + return "", err + } + if res := fetcher.Meta().Res; res != nil { + for _, index := range opts.SelectFiles { + if index < 0 || index >= len(res.Files) { + return "", fmt.Errorf("invalid selected file index: %d", index) + } + } + opts.InitSelectFiles(len(res.Files)) + } + fetcher.Meta().Opts = opts if res := fetcher.Meta().Res; res != nil { res.CalcSize(fetcher.Meta().Opts.SelectFiles) } @@ -1469,7 +1478,10 @@ func (d *Downloader) watch(task *Task) { if err := d.captureTaskStats(task, false, true); err != nil { d.Logger.Error().Stack().Err(err).Msgf("persist completed task stats failed, task id: %s", task.ID) } - d.storage.Put(bucketTask, task.ID, task.clone()) + task.statusLock.Lock() + snapshot := task.clone() + task.statusLock.Unlock() + d.storage.Put(bucketTask, task.ID, snapshot) d.emit(EventKeyDone, task) d.emit(EventKeyFinally, task, err) d.notifyRunning() @@ -1826,7 +1838,7 @@ func (d *Downloader) doStart(task *Task) (err error) { return err } if task.Meta.Res == nil { - err := task.fetcher.Resolve(task.Meta.Req, task.Meta.Opts) + err := task.fetcher.Resolve(task.Meta.Req) if err != nil { return err } diff --git a/pkg/download/downloader_test.go b/pkg/download/downloader_test.go index 545effb3..db0bd592 100644 --- a/pkg/download/downloader_test.go +++ b/pkg/download/downloader_test.go @@ -132,7 +132,7 @@ func (f *generationTestFetcher) Setup(*controller.Controller) { f.done = make(chan error, 2) } } -func (f *generationTestFetcher) Resolve(req *base.Request, opts *base.Options) error { +func (f *generationTestFetcher) Resolve(req *base.Request) error { if f.manager.resolveStarted != nil { f.manager.resolveOnce.Do(func() { close(f.manager.resolveStarted) }) } @@ -143,7 +143,7 @@ func (f *generationTestFetcher) Resolve(req *base.Request, opts *base.Options) e return f.manager.resolveErr } f.meta.Req = req - f.meta.Opts = opts + f.meta.Res = &base.Resource{Files: []*base.FileInfo{{Name: "generation.bin", Size: 1}}} return nil } @@ -245,7 +245,7 @@ func TestDownloader_Resolve(t *testing.T) { req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } - rr, err := downloader.Resolve(req, nil) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } @@ -772,7 +772,7 @@ func TestDownloader_Create(t *testing.T) { URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } opts := newTestDownloadOpt(t) - rr, err := downloader.Resolve(req, opts) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } @@ -784,7 +784,7 @@ func TestDownloader_Create(t *testing.T) { wg.Done() } }) - _, err = downloader.Create(rr.ID) + _, err = downloader.CreateWithOptions(rr.ID, opts) if err != nil { t.Fatal(err) } @@ -878,15 +878,15 @@ func TestDownloader_CreateNotInWhite(t *testing.T) { req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } - // With new fetcher design, white list check happens during Resolve (not Create) - // because Resolve now requires Options which includes the download path - _, err := downloader.Resolve(req, newTestDownloadOpt(t)) + rr, err := downloader.Resolve(req) + if err != nil { + t.Fatal(err) + } + _, err = downloader.CreateWithOptions(rr.ID, newTestDownloadOpt(t)) if err == nil { - t.Error("TestDownloader_CreateNotInWhite() expected error but got nil") - } - if !strings.Contains(err.Error(), "white") { - t.Errorf("TestDownloader_CreateNotInWhite() got = %v, want error containing 'white'", err.Error()) + t.Fatal("Create should reject a directory outside the allowlist") } + } func TestDownloader_CreateDirectBatch(t *testing.T) { @@ -1072,7 +1072,7 @@ func doTestDownloaderCreateWithProxy(t *testing.T, auth bool, buildReqProxy func Pwd: pwd, }) } - rr, err := downloader.Resolve(req, nil) + rr, err := downloader.Resolve(req) if err != nil { if errHandler == nil { t.Fatal(err) @@ -1158,7 +1158,7 @@ func TestDownloader_StoreAndRestore(t *testing.T) { URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } opts := newTestDownloadOpt(t) - rr, err := downloader.Resolve(req, opts) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } @@ -1172,7 +1172,7 @@ func TestDownloader_StoreAndRestore(t *testing.T) { }) } }) - id, err := downloader.Create(rr.ID) + id, err := downloader.CreateWithOptions(rr.ID, opts) if err != nil { t.Fatal(err) } @@ -1434,7 +1434,7 @@ func TestDownloader_Stats(t *testing.T) { req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } - rr, err := downloader.Resolve(req, newTestDownloadOpt(t)) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } @@ -1447,7 +1447,7 @@ func TestDownloader_Stats(t *testing.T) { } }) - taskId, err := downloader.Create(rr.ID) + taskId, err := downloader.CreateWithOptions(rr.ID, newTestDownloadOpt(t)) if err != nil { t.Fatal(err) } @@ -1677,7 +1677,7 @@ func TestDownloader_CompletedHTTPStatsSurviveRestore(t *testing.T) { t.Fatal(err) } req := &base.Request{URL: "http://" + listener.Addr().String() + "/" + test.BuildName} - rr, err := downloader.Resolve(req, newTestDownloadOpt(t)) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } @@ -1688,7 +1688,7 @@ func TestDownloader_CompletedHTTPStatsSurviveRestore(t *testing.T) { done <- struct{}{} } }) - taskID, err := downloader.Create(rr.ID) + taskID, err := downloader.CreateWithOptions(rr.ID, newTestDownloadOpt(t)) if err != nil { t.Fatal(err) } @@ -1861,11 +1861,11 @@ func TestDownloader_PauseAndContinue(t *testing.T) { req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } - rr, err := downloader.Resolve(req, newTestDownloadOpt(t)) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } - taskId, err := downloader.Create(rr.ID) + taskId, err := downloader.CreateWithOptions(rr.ID, newTestDownloadOpt(t)) if err != nil { t.Fatal(err) } @@ -2045,11 +2045,11 @@ func TestDownloader_PauseAllAndContinueAll(t *testing.T) { req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } - rr, err := downloader.Resolve(req, newTestDownloadOpt(t)) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } - taskId, err := downloader.Create(rr.ID) + taskId, err := downloader.CreateWithOptions(rr.ID, newTestDownloadOpt(t)) if err != nil { t.Fatal(err) } diff --git a/pkg/download/extension.go b/pkg/download/extension.go index 29993cb5..01585ac8 100644 --- a/pkg/download/extension.go +++ b/pkg/download/extension.go @@ -753,6 +753,10 @@ type OnErrorExtensionTask struct { } func cloneExtensionTask(task *Task) *Task { + if task.statusLock != nil { + task.statusLock.Lock() + defer task.statusLock.Unlock() + } newTask := task.clone() newTask.Meta.Req = task.Meta.Req return newTask diff --git a/pkg/download/extension_extra_test.go b/pkg/download/extension_extra_test.go index d298946a..f6770fff 100644 --- a/pkg/download/extension_extra_test.go +++ b/pkg/download/extension_extra_test.go @@ -65,11 +65,11 @@ func TestExtensionExtraMutationHTTP(t *testing.T) { var id string var err error if resolved { - rr, resolveErr := d.Resolve(req, opts) + rr, resolveErr := d.Resolve(req) if resolveErr != nil { t.Fatal(resolveErr) } - id, err = d.Create(rr.ID) + id, err = d.CreateWithOptions(rr.ID, opts) } else { id, err = d.CreateDirect(req, opts) } diff --git a/pkg/download/extension_test.go b/pkg/download/extension_test.go index e1f81ab3..8644664b 100644 --- a/pkg/download/extension_test.go +++ b/pkg/download/extension_test.go @@ -32,7 +32,7 @@ func TestDownloader_InstallExtensionByFolder(t *testing.T) { URL: "https://github.com/test", Labels: map[string]string{"original": "true"}, } - rr, err := downloader.Resolve(req, nil) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } @@ -58,7 +58,7 @@ func TestDownloader_InstallExtensionByFolderDevMode(t *testing.T) { } rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/test", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -76,7 +76,7 @@ func TestDownloader_Extension_BlobBlob(t *testing.T) { rr, err := downloader.Resolve(&base.Request{ URL: "https://example.com/blob", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -130,7 +130,7 @@ func TestDownloader_Extension_BlobFunctionOpener(t *testing.T) { rr, err := downloader.Resolve(&base.Request{ URL: "https://example.com/opener", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -166,7 +166,7 @@ func TestDownloader_Extension_BlobFunctionOpenerUnknownSize(t *testing.T) { rr, err := downloader.Resolve(&base.Request{ URL: "https://example.com/opener-unknown", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -222,7 +222,7 @@ func TestDownloader_Extension_BlobSourceSizePropagatesToCreatedTask(t *testing.T rr, err := downloader.Resolve(&base.Request{ URL: "https://example.com/opener", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -303,7 +303,7 @@ func TestDownloader_Extension_BlobFetchDrainedStreamReportsEarlyProgress(t *test rr, err := downloader.Resolve(&base.Request{ URL: fmt.Sprintf("%s/single-fetch-stream?size=%d", server.URL, len(payload)), - }, nil) + }) if err != nil { t.Fatal(err) } @@ -419,7 +419,7 @@ func TestDownloader_Extension_BlobRangeFetchDrainedStreamReportsEarlyProgress(t rr, err := downloader.Resolve(&base.Request{ URL: fmt.Sprintf("%s/range-fetch-stream?size=%d", server.URL, len(payload)), - }, nil) + }) if err != nil { t.Fatal(err) } @@ -494,7 +494,7 @@ func TestDownloader_Extension_BlobMultiplexFetchStreamsReportEarlyProgress(t *te rr, err := downloader.Resolve(&base.Request{ URL: server.URL + "/multiplex-fetch-stream", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -550,7 +550,7 @@ func TestDownloader_Extension_BlobFunctionOpenerRangeResume(t *testing.T) { rr, err := downloader.Resolve(&base.Request{ URL: "https://example.com/opener-range", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -683,7 +683,7 @@ func TestDownloader_Extension_BlobResumeAfterRestartViaOnError(t *testing.T) { rr, err := downloader.Resolve(&base.Request{ URL: server.URL + "/restart-range", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -855,7 +855,7 @@ func TestDownloader_Extension_BlobPauseContinueKeepsSource(t *testing.T) { rr, err := downloader.Resolve(&base.Request{ URL: tc.url, - }, nil) + }) if err != nil { t.Fatal(err) } @@ -932,7 +932,7 @@ func TestDownloader_Extension_BlobUnknownSizePauseContinueKeepsSource(t *testing rr, err := downloader.Resolve(&base.Request{ URL: "https://example.com/opener-unknown", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1009,7 +1009,7 @@ func TestDownloader_Extension_BlobRecoverOnError(t *testing.T) { path := path t.Run(path, func(t *testing.T) { rawURL := "https://example.com/" + path - rr, err := downloader.Resolve(&base.Request{URL: rawURL}, nil) + rr, err := downloader.Resolve(&base.Request{URL: rawURL}) if err != nil { t.Fatal(err) } @@ -1076,7 +1076,7 @@ func TestDownloader_InstallExtensionByGit(t *testing.T) { } rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/GopeedLab/gopeed/releases", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1093,7 +1093,7 @@ func TestDownloader_InstallExtensionByGitSimple(t *testing.T) { } rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/GopeedLab/gopeed/releases", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1110,7 +1110,7 @@ func TestDownloader_InstallExtensionByGitFull(t *testing.T) { } rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/GopeedLab/gopeed/releases", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1180,7 +1180,7 @@ func TestDownloader_UpgradeExtension(t *testing.T) { rr, err := downloader.Resolve(&base.Request{ URL: "https://test.com", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1355,7 +1355,7 @@ func TestDownloader_Extension_Errors(t *testing.T) { } rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/test", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1370,7 +1370,7 @@ func TestDownloader_Extension_Errors(t *testing.T) { } rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/test", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1385,7 +1385,7 @@ func TestDownloader_Extension_Errors(t *testing.T) { } _, err := downloader.Resolve(&base.Request{ URL: "https://github.com/test", - }, nil) + }) if err == nil { t.Fatalf("except error, but got nil") } @@ -1407,7 +1407,7 @@ func TestDownloader_Extension_Settings(t *testing.T) { } rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/test", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1428,7 +1428,7 @@ func TestDownloader_Extension_Settings(t *testing.T) { }) rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/test", - }, nil) + }) if err != nil { t.Fatal(err) } @@ -1445,7 +1445,7 @@ func TestDownloader_ExtensionStorage(t *testing.T) { } rr, err := downloader.Resolve(&base.Request{ URL: "https://github.com/test", - }, nil) + }) if err != nil { t.Fatal(err) } diff --git a/pkg/download/extension_unnamed_resource_test.go b/pkg/download/extension_unnamed_resource_test.go index 7ef765eb..6b70198f 100644 --- a/pkg/download/extension_unnamed_resource_test.go +++ b/pkg/download/extension_unnamed_resource_test.go @@ -24,7 +24,7 @@ func TestExtensionUnnamedResource(t *testing.T) { if _, err = d.InstallExtensionByFolder(dir, true); err != nil { t.Fatal(err) } - res, err := d.Resolve(&base.Request{URL: "https://example.com/unnamed"}, nil) + res, err := d.Resolve(&base.Request{URL: "https://example.com/unnamed"}) if err != nil { t.Fatal(err) } diff --git a/pkg/download/hls_live_test.go b/pkg/download/hls_live_test.go index 7fb41054..03a6681e 100644 --- a/pkg/download/hls_live_test.go +++ b/pkg/download/hls_live_test.go @@ -40,7 +40,7 @@ func TestDownloader_HlsLiveSite(t *testing.T) { if referer != "" { req.Extra = map[string]any{"header": map[string]string{"Referer": referer}} } - rr, err := downloader.Resolve(req, nil) + rr, err := downloader.Resolve(req) if err != nil { t.Fatalf("resolve: %v", err) } diff --git a/pkg/download/hls_protocol_test.go b/pkg/download/hls_protocol_test.go index 8ef48df8..88a2aefe 100644 --- a/pkg/download/hls_protocol_test.go +++ b/pkg/download/hls_protocol_test.go @@ -89,7 +89,7 @@ func TestDownloader_HlsProtocolEndToEnd(t *testing.T) { req := hlsResolveReferer() req.URL = server.URL + "/master.m3u8" - rr, err := downloader.Resolve(req, nil) + rr, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } diff --git a/pkg/download/selection_test.go b/pkg/download/selection_test.go index e97a65b1..cc31bd9c 100644 --- a/pkg/download/selection_test.go +++ b/pkg/download/selection_test.go @@ -18,7 +18,7 @@ func TestCreateResolvedFileSelection(t *testing.T) { }{ {"partial", []int{2, 0}, true, []int{2, 0}, 40}, {"default all", nil, true, []int{0, 1, 2}, 60}, - {"legacy retains resolve options", nil, false, []int{1}, 20}, + {"default creation options", nil, false, []int{0, 1, 2}, 60}, } { t.Run(tc.name, func(t *testing.T) { manager := &generationTestManager{} @@ -28,8 +28,8 @@ func TestCreateResolvedFileSelection(t *testing.T) { } defer d.Clear() d.cfg.MaxRunning = 0 - original := &base.Options{Path: t.TempDir(), SelectFiles: []int{1}} - rr, err := d.Resolve(&base.Request{URL: "generation://selection"}, original) + original := &base.Options{Path: t.TempDir()} + rr, err := d.Resolve(&base.Request{URL: "generation://selection"}) if err != nil { t.Fatal(err) } @@ -40,7 +40,7 @@ func TestCreateResolvedFileSelection(t *testing.T) { if tc.submit { id, err = d.CreateWithOptions(rr.ID, &base.Options{Path: original.Path, SelectFiles: tc.selection}) } else { - id, err = d.Create(rr.ID) + id, err = d.CreateWithOptions(rr.ID, original) } if err != nil { t.Fatal(err) diff --git a/pkg/mcpserver/server.go b/pkg/mcpserver/server.go index 10c6ec53..367d29d2 100644 --- a/pkg/mcpserver/server.go +++ b/pkg/mcpserver/server.go @@ -222,8 +222,7 @@ func (o *downloadOptions) baseOptions() *base.Options { } type resolveTaskInput struct { - Req *downloadRequest `json:"req" jsonschema:"Download request to resolve"` - Opts *downloadOptions `json:"opts,omitempty" jsonschema:"Optional download settings used while resolving"` + Req *downloadRequest `json:"req" jsonschema:"Download request to resolve"` } type resolveTaskOutput struct { @@ -328,7 +327,7 @@ func registerTools(server *mcp.Server, downloader *download.Downloader) { if err != nil { return nil, nil, err } - result, err := downloader.Resolve(req, input.Opts.baseOptions()) + result, err := downloader.Resolve(req) if err != nil { return nil, nil, err } diff --git a/pkg/mcpserver/server_test.go b/pkg/mcpserver/server_test.go index 225d6ed8..3da3bbce 100644 --- a/pkg/mcpserver/server_test.go +++ b/pkg/mcpserver/server_test.go @@ -101,6 +101,13 @@ func TestStreamableHTTPTools(t *testing.T) { if reqExtra["type"] != "object" || !ok || len(branches) != 2 { t.Fatalf("%s req.extra schema = %#v, want an object with two oneOf branches", tool.Name, reqExtra) } + if tool.Name == "resolve_task" { + properties := schema["properties"].(map[string]any) + if _, exists := properties["opts"]; exists || len(properties) != 1 { + t.Fatalf("resolve_task must expose only req: %s", inputSchema) + } + continue + } optsExtra := nestedSchemaProperty(t, schema, "opts", "extra") if !schemaIncludesType(optsExtra, "object") || optsExtra["additionalProperties"] != false { t.Fatalf("%s opts.extra schema = %#v, want concrete HTTP task settings", tool.Name, optsExtra) diff --git a/pkg/rest/api.go b/pkg/rest/api.go index 7e6d2e3c..0b0457af 100644 --- a/pkg/rest/api.go +++ b/pkg/rest/api.go @@ -26,7 +26,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, req.Opts) + rr, err := Downloader.Resolve(req.Req) if err != nil { WriteJson(w, model.NewErrorResult(err.Error())) return @@ -82,7 +82,7 @@ func PatchTask(w http.ResponseWriter, r *http.Request) { return } - var req model.ResolveTask + var req model.PatchTask if ReadJson(r, w, &req) { if err := Downloader.Patch(taskId, req.Req, req.Opts); err != nil { if err == download.ErrTaskNotFound { diff --git a/pkg/rest/model/task.go b/pkg/rest/model/task.go index 8c0bd1b2..4091e977 100644 --- a/pkg/rest/model/task.go +++ b/pkg/rest/model/task.go @@ -3,6 +3,10 @@ package model import "github.com/GopeedLab/gopeed/pkg/base" type ResolveTask struct { + Req *base.Request `json:"req"` +} + +type PatchTask struct { Req *base.Request `json:"req"` Opts *base.Options `json:"opts"` } diff --git a/pkg/rest/server_test.go b/pkg/rest/server_test.go index 553dd381..e37d32c4 100644 --- a/pkg/rest/server_test.go +++ b/pkg/rest/server_test.go @@ -55,8 +55,7 @@ var ( }, } resolveReq = &model.ResolveTask{ - Req: taskReq, - Opts: createOpts, + Req: taskReq, } createReq = &model.CreateTask{ Req: taskReq, @@ -105,7 +104,8 @@ func TestCreateTask(t *testing.T) { }) taskId := httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", &model.CreateTask{ - Rid: resp.ID, + Rid: resp.ID, + Opts: createOpts, }) if taskId == "" { t.Fatal("create task failed") @@ -295,7 +295,7 @@ func TestPatchTask(t *testing.T) { time.Sleep(time.Millisecond * 100) // Patch the task with new labels - patchReq := &model.ResolveTask{ + patchReq := &model.PatchTask{ Req: &base.Request{ Labels: map[string]string{ "patched": "true", @@ -322,7 +322,7 @@ func TestPatchTask(t *testing.T) { func TestPatchTaskNotFound(t *testing.T) { doTest(func() { // Try to patch a non-existent task - patchReq := &model.ResolveTask{ + patchReq := &model.PatchTask{ Req: &base.Request{ Labels: map[string]string{ "test": "value", diff --git a/ui/flutter/lib/api/api.dart b/ui/flutter/lib/api/api.dart index 3ee841ce..2845007a 100644 --- a/ui/flutter/lib/api/api.dart +++ b/ui/flutter/lib/api/api.dart @@ -10,6 +10,7 @@ import 'model/install_extension.dart'; import 'model/login.dart'; import 'model/resolve_result.dart'; import 'model/resolve_task.dart'; +import 'model/patch_task.dart'; import 'model/result.dart'; import 'model/switch_extension.dart'; import 'model/task.dart'; @@ -91,7 +92,7 @@ Future> createTaskBatch(CreateTaskBatch createTaskBatch) { ); } -Future patchTask(String id, ResolveTask patchTask) { +Future patchTask(String id, PatchTask patchTask) { return _parse(() => _request('api/v1/tasks/$id', method: 'PATCH', data: patchTask), null); } diff --git a/ui/flutter/lib/api/model/patch_task.dart b/ui/flutter/lib/api/model/patch_task.dart new file mode 100644 index 00000000..66f0b4d0 --- /dev/null +++ b/ui/flutter/lib/api/model/patch_task.dart @@ -0,0 +1,16 @@ +import 'options.dart'; +import 'request.dart'; + +class PatchTask { + final Request? req; + final Options? opts; + + PatchTask({this.req, this.opts}); + + factory PatchTask.fromJson(Map json) => PatchTask( + req: json['req'] == null ? null : Request.fromJson(json['req'] as Map), + opts: json['opts'] == null ? null : Options.fromJson(json['opts'] as Map), + ); + + Map toJson() => {'req': req?.toJson(), 'opts': opts?.toJson()}; +} diff --git a/ui/flutter/lib/api/model/resolve_task.dart b/ui/flutter/lib/api/model/resolve_task.dart index c1458e75..5fc8cc1b 100644 --- a/ui/flutter/lib/api/model/resolve_task.dart +++ b/ui/flutter/lib/api/model/resolve_task.dart @@ -1,6 +1,5 @@ import 'package:json_annotation/json_annotation.dart'; -import 'options.dart'; import 'request.dart'; part 'resolve_task.g.dart'; @@ -8,9 +7,8 @@ part 'resolve_task.g.dart'; @JsonSerializable() class ResolveTask { Request? req; - Options? opts; - ResolveTask({this.req, this.opts}); + ResolveTask({this.req}); 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 90680bda..0305902b 100644 --- a/ui/flutter/lib/api/model/resolve_task.g.dart +++ b/ui/flutter/lib/api/model/resolve_task.g.dart @@ -7,11 +7,10 @@ part of 'resolve_task.dart'; // ************************************************************************** ResolveTask _$ResolveTaskFromJson(Map json) => ResolveTask( - req: json['req'] == null ? null : Request.fromJson(json['req'] as Map), - opts: json['opts'] == null ? null : Options.fromJson(json['opts'] as Map), + req: json['req'] == null + ? null + : Request.fromJson(json['req'] as Map), ); -Map _$ResolveTaskToJson(ResolveTask instance) => { - 'req': ?instance.req, - 'opts': ?instance.opts, -}; +Map _$ResolveTaskToJson(ResolveTask instance) => + {'req': ?instance.req}; diff --git a/ui/flutter/lib/core/capabilities/app_capabilities.dart b/ui/flutter/lib/core/capabilities/app_capabilities.dart index 106a42c7..be71ca0e 100644 --- a/ui/flutter/lib/core/capabilities/app_capabilities.dart +++ b/ui/flutter/lib/core/capabilities/app_capabilities.dart @@ -8,6 +8,7 @@ import '../../api/model/extension.dart'; import '../../api/model/install_extension.dart'; import '../../api/model/resolve_result.dart'; import '../../api/model/resolve_task.dart'; +import '../../api/model/patch_task.dart'; import '../../api/model/switch_extension.dart'; import '../../api/model/task.dart'; import '../../api/model/update_check_extension_resp.dart'; @@ -85,7 +86,7 @@ void _bindGopeed(CapabilityRegistry registry) { ..bind(GopeedMethods.createTask, api.createTask) ..bind(GopeedMethods.createTaskBatch, api.createTaskBatch) ..bind(GopeedMethods.patchTask, (params) async { - await api.patchTask(params['id'] as String, ResolveTask.fromJson(params['request'] as Map)); + await api.patchTask(params['id'] as String, PatchTask.fromJson(params['request'] as Map)); return const RpcUnit(); }) ..bind( diff --git a/ui/flutter/lib/core/capabilities/gopeed_capability.dart b/ui/flutter/lib/core/capabilities/gopeed_capability.dart index ea03dd68..d7e497d4 100644 --- a/ui/flutter/lib/core/capabilities/gopeed_capability.dart +++ b/ui/flutter/lib/core/capabilities/gopeed_capability.dart @@ -5,6 +5,7 @@ import '../../api/model/extension.dart'; import '../../api/model/install_extension.dart'; import '../../api/model/resolve_result.dart'; import '../../api/model/resolve_task.dart'; +import '../../api/model/patch_task.dart'; import '../../api/model/switch_extension.dart'; import '../../api/model/task.dart'; import '../../api/model/update_check_extension_resp.dart'; @@ -49,7 +50,7 @@ class GopeedService { Future> createTaskBatch(CreateTaskBatch request) => _invoker.invoke(GopeedMethods.createTaskBatch, request); - Future patchTask(String id, ResolveTask request) async { + Future patchTask(String id, PatchTask request) async { await _invoker.invoke(GopeedMethods.patchTask, {'id': id, 'request': request.toJson()}); } diff --git a/ui/flutter/lib/features/tasks/application/tasks_controller.dart b/ui/flutter/lib/features/tasks/application/tasks_controller.dart index 947a0de4..43071f32 100644 --- a/ui/flutter/lib/features/tasks/application/tasks_controller.dart +++ b/ui/flutter/lib/features/tasks/application/tasks_controller.dart @@ -3,7 +3,7 @@ import 'dart:async'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../../../api/model/request.dart'; -import '../../../api/model/resolve_task.dart'; +import '../../../api/model/patch_task.dart'; import '../../../api/model/task.dart' as api_model; import '../../../core/capabilities/app_capabilities.dart'; import '../domain/task_record.dart'; @@ -99,7 +99,7 @@ class TasksController extends AsyncNotifier { } Future updateRequest(String id, Request request) async { - await ref.read(gopeedServiceProvider).patchTask(id, ResolveTask(req: request)); + await ref.read(gopeedServiceProvider).patchTask(id, PatchTask(req: request)); await ref.read(gopeedServiceProvider).continueTask(id); await refresh(silent: true); } diff --git a/ui/flutter/lib/features/tasks/presentation/pages/create_task_window_page.dart b/ui/flutter/lib/features/tasks/presentation/pages/create_task_window_page.dart index 3c36086e..d7b9e653 100644 --- a/ui/flutter/lib/features/tasks/presentation/pages/create_task_window_page.dart +++ b/ui/flutter/lib/features/tasks/presentation/pages/create_task_window_page.dart @@ -927,8 +927,7 @@ class _CreateTaskWindowPageState extends ConsumerState { } final request = _buildRequest(urls.first, protocolUrl: _protocolUrlFor(urls.first)); - final options = _buildOptions(); - final result = await ref.read(gopeedServiceProvider).resolve(ResolveTask(req: request, opts: options)); + final result = await ref.read(gopeedServiceProvider).resolve(ResolveTask(req: request)); if (!mounted) return; final created = await _showResolveDialog(request, result); if (!created) { diff --git a/ui/flutter/test/api/resolve_test.dart b/ui/flutter/test/api/resolve_test.dart index 14c40002..323a38c6 100644 --- a/ui/flutter/test/api/resolve_test.dart +++ b/ui/flutter/test/api/resolve_test.dart @@ -3,30 +3,49 @@ import 'dart:convert'; import 'package:dio/dio.dart' as dio; import 'package:flutter_test/flutter_test.dart'; import 'package:gopeed/api/api.dart' as api; +import 'package:gopeed/api/model/create_task.dart'; import 'package:gopeed/api/model/options.dart'; +import 'package:gopeed/api/model/patch_task.dart'; import 'package:gopeed/api/model/request.dart'; import 'package:gopeed/api/model/resolve_task.dart'; import 'package:gopeed/core/network/gopeed/gopeed_transport.dart'; void main() { - test('resolve sends the complete resolve task for magnet links', () async { + test('resolve omits download options and creation submits them separately', () async { final transport = _RecordingTransport(); api.setTransportForTesting(transport); final result = await api.resolve( - ResolveTask( - req: Request(url: 'magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567'), - opts: Options(path: '/downloads', selectFiles: const []), - ), + ResolveTask(req: Request(url: 'magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567')), ); expect(result.id, 'resolved-id'); final requestBody = transport.data as Map; - expect(requestBody.keys, containsAll(['req', 'opts'])); + expect(requestBody.keys, equals(['req'])); expect( (requestBody['req'] as Map)['url'], 'magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567', ); - expect((requestBody['opts'] as Map)['path'], '/downloads'); + expect(requestBody, isNot(contains('opts'))); + + final id = await api.createTask( + CreateTask( + rid: result.id, + opts: Options(path: '/chosen-after-resolve', selectFiles: [1]), + ), + ); + expect(id, 'created-id'); + final createBody = transport.data as Map; + expect(createBody['rid'], 'resolved-id'); + expect((createBody['opts'] as Map)['path'], '/chosen-after-resolve'); + expect((createBody['opts'] as Map)['selectFiles'], [1]); + }); + + test('patch preserves options in its independent request model', () async { + final transport = _RecordingTransport(); + api.setTransportForTesting(transport); + await api.patchTask('task-id', PatchTask(opts: Options(selectFiles: [1]))); + final requestBody = transport.data as Map; + expect((requestBody['opts'] as Map)['selectFiles'], [1]); }); } @@ -41,6 +60,12 @@ class _RecordingTransport implements GopeedTransport { Map? queryParameters, }) async { this.data = jsonDecode(jsonEncode(data)); + if (path == 'api/v1/tasks') { + return {'code': 0, 'data': 'created-id'}; + } + if (method == 'PATCH') { + return {'code': 0, 'data': null}; + } return { 'code': 0, 'data': { diff --git a/ui/flutter/test/core/capabilities/capability_rpc_test.dart b/ui/flutter/test/core/capabilities/capability_rpc_test.dart index a6b1dd78..c3c5a324 100644 --- a/ui/flutter/test/core/capabilities/capability_rpc_test.dart +++ b/ui/flutter/test/core/capabilities/capability_rpc_test.dart @@ -1,5 +1,13 @@ import 'package:flutter_test/flutter_test.dart'; +import 'package:gopeed/api/model/options.dart'; +import 'package:gopeed/api/model/patch_task.dart'; +import 'package:gopeed/api/model/request.dart'; +import 'package:gopeed/api/model/resolve_result.dart'; +import 'package:gopeed/api/model/resolve_task.dart'; +import 'package:gopeed/api/model/resource.dart'; +import 'package:gopeed/core/capabilities/app_capabilities.dart'; import 'package:gopeed/core/capabilities/capability_rpc.dart'; +import 'package:gopeed/core/capabilities/gopeed_capability.dart'; import 'package:gopeed/core/window/app_window_appearance.dart'; class _Payload { @@ -13,6 +21,33 @@ class _Payload { } void main() { + test('serialized resolve and patch capabilities keep separate request contracts', () async { + final codecs = createAppCapabilityCodecs(); + String? resolvedUrl; + Map? patch; + final registry = CapabilityRegistry(codecs) + ..bind(GopeedMethods.resolve, (request) { + resolvedUrl = request.req?.url; + return ResolveResult(id: 'resolved-id', res: Resource(files: [])); + }) + ..bind(GopeedMethods.patchTask, (request) { + patch = request; + return const RpcUnit(); + }); + final request = ResolveTask(req: Request(url: 'magnet:?xt=urn:btih:hash')); + final wireRequest = codecs.encode(request) as Map; + expect(wireRequest.keys, ['req']); + final result = codecs.decode(await registry.invoke(GopeedMethods.resolve.name, wireRequest)); + expect(result.id, 'resolved-id'); + expect(resolvedUrl, request.req?.url); + + final service = GopeedService(LocalCapabilityInvoker(registry)); + await service.patchTask('task-id', PatchTask(opts: Options(selectFiles: [1]))); + expect(patch?['id'], 'task-id'); + final patchBody = patch?['request'] as Map; + expect((patchBody['opts'] as Map)['selectFiles'], [1]); + }); + test('one method descriptor supports direct and serialized dispatch', () async { const method = RpcMethod<_Payload, _Payload>('test.increment'); final codecs = RpcCodecRegistry()..register<_Payload>((json) => _Payload.fromJson(json! as Map));