diff --git a/backend/download/apiclient.go b/backend/download/apiclient.go index cb86e9b..d98641c 100644 --- a/backend/download/apiclient.go +++ b/backend/download/apiclient.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -144,17 +145,36 @@ func (c *apiClient) checkStatus(resp *http.Response) error { case resp.StatusCode >= 400: snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) - return fmt.Errorf( - "%w: HTTP %d: %s", - c.errUnreachable, - resp.StatusCode, - strings.TrimSpace(string(snippet)), - ) + return fmt.Errorf("%w: %w", c.errUnreachable, &httpStatusError{ + code: resp.StatusCode, + body: strings.TrimSpace(string(snippet)), + }) default: return nil } } +// httpStatusError is a non-2xx answer, kept typed so a caller can tell +// a missing endpoint from a daemon that is down. +type httpStatusError struct { + code int + body string +} + +func (e *httpStatusError) Error() string { + return fmt.Sprintf("HTTP %d: %s", e.code, e.body) +} + +// statusCode returns the HTTP status an error carries, or 0. +func statusCode(err error) int { + var se *httpStatusError + if errors.As(err, &se) { + return se.code + } + + return 0 +} + // decodeJSON decodes a JSON string into out. Providers whose auth or // response handling does not fit apiClient still parse bodies the same // way, so the helper lives here rather than being repeated. diff --git a/backend/download/candidate_shape_test.go b/backend/download/candidate_shape_test.go index 6b979a0..5fcc4ec 100644 --- a/backend/download/candidate_shape_test.go +++ b/backend/download/candidate_shape_test.go @@ -159,7 +159,7 @@ func TestSlskdCollectKeepsDiscFolders(t *testing.T) { got, err := s.collect(Candidate{Files: []CandidateFile{ {Path: `\m\Album\CD1\01 Intro.flac`, IsAudio: true}, {Path: `\m\Album\CD2\01 Intro.flac`, IsAudio: true}, - }}, dst) + }}, dst, "", nil) if err != nil { t.Fatalf("collect: %v", err) } diff --git a/backend/download/provider_slskd.go b/backend/download/provider_slskd.go index 51a6064..66f9ca2 100644 --- a/backend/download/provider_slskd.go +++ b/backend/download/provider_slskd.go @@ -5,13 +5,16 @@ import ( "errors" "fmt" "log/slog" + "net/http" "net/url" "os" "path" "path/filepath" "regexp" "slices" + "strconv" "strings" + "sync/atomic" "time" "github.com/google/uuid" @@ -68,6 +71,10 @@ const ( // do not get cut off by the context deadline. slskdSearchWait = 20 * time.Second + // slskdMinSearchTimeout is the smallest searchTimeout slskd accepts, + // in seconds. + slskdMinSearchTimeout = 5 + // slskdTransferPoll is how often transfer state is polled. slskdTransferPoll = 3 * time.Second @@ -161,6 +168,10 @@ type slskd struct { transferPoll time.Duration stallAfter time.Duration absentGrace time.Duration + + // batches is whether the daemon takes batch downloads, which is + // how a grab gets a folder of its own (see enqueue). + batches atomic.Int32 } // newSlskd builds the provider from config. @@ -431,14 +442,18 @@ func (s *slskd) searchRequest(id, text string, minFiles int) map[string]any { maximumPeerQueueLength = 100 ) - // A tenth of the wait is left for the last poll and the responses - // fetch. - timeout := s.searchWait - s.searchWait/10 + // slskd reads this in whole seconds, counted from the last response + // rather than from the start, with a floor of 5 (#274). A tenth of + // our own wait is left for the last poll and the responses fetch. + timeout := max( + int((s.searchWait-s.searchWait/10)/time.Second), + slskdMinSearchTimeout, + ) return map[string]any{ "id": id, "searchText": text, - "searchTimeout": timeout.Milliseconds(), + "searchTimeout": timeout, "responseLimit": responseLimit, "fileLimit": fileLimit, "filterResponses": true, @@ -736,20 +751,99 @@ func (s *slskd) Grab( ) } + // Only the per-user enqueue writes into folders other grabs share; + // a batch has a folder of its own. Until the daemon has answered a + // batch either way, take the locks anyway. + if s.batches.Load() != batchesSupported { + release, err := lockSlskdFolders(ctx, s.localFolders(c)) + if err != nil { + return Result{}, err + } + defer release() + } + // slskd keeps finished transfers listed until someone removes them, // and a transfer is matched to the request by filename. A record // left by an earlier attempt at the same file from the same peer // would otherwise be read as this attempt's answer the moment the // first poll came back — an old failure failing a transfer that has // not started. So what is already terminal is noted before enqueueing - // and ignored after. - release, err := lockSlskdFolders(ctx, s.localFolders(c)) + // and ignored after. The files already on disk are noted for the + // same reason (see arrivedFile). + stale := s.terminalTransferIDs(ctx, username) + existing := s.snapshotFolders(c) + + dest, err := s.enqueue(ctx, username, c) if err != nil { return Result{}, err } - defer release() - stale := s.terminalTransferIDs(ctx, username) + if err := s.awaitTransfers( + ctx, username, stale, c, onProgress, + ); err != nil { + s.discardDestination(dest) + + return Result{}, err + } + + result, err := s.collect(c, dst, dest, existing) + + s.discardDestination(dest) + + return result, err +} + +// Whether the daemon has the batch endpoint, learned from the first +// grab that asks. +const ( + batchesUnknown int32 = iota + batchesSupported + batchesUnsupported +) + +// slskdDestRoot is the folder under slskd's downloads directory that +// batch destinations are made in, so everything this app asked slskd to +// write is in one place and nothing else is. +const slskdDestRoot = "yellowjacket" + +// enqueue asks slskd for a candidate's files and returns the folder, +// relative to the downloads directory, they will be written to — or "" +// when slskd will choose, which is its per-user enqueue. +// +// slskd 0.26 takes a batch with an explicit destination, which is the +// only way to know for certain where a file lands. Without one it is +// `//`, shared with every other download +// of a same-named folder and with the user's own, renamed with a +// `_` suffix when a name is taken, and moved by the user's +// `Destination.Subdirectory` setting (#274). +func (s *slskd) enqueue( + ctx context.Context, + username string, + c Candidate, +) (string, error) { + if s.batches.Load() != batchesUnsupported { + dest := slskdDestRoot + "/" + uuid.NewString() + + err := s.enqueueBatches(ctx, username, c, dest) + if err == nil { + s.batches.Store(batchesSupported) + + return dest, nil + } + + if !batchEndpointMissing(err) { + return "", err + } + + // An older daemon routes this path to the per-user enqueue with + // "batches" as the username and rejects the body, which is the + // 400; 404 and 405 are a daemon that routes it nowhere. + s.batches.Store(batchesUnsupported) + s.logger.Info( + "slskd has no batch downloads; files will be found by name", + "error", err, + ) + } wanted := make([]map[string]any, 0, len(c.Files)) for _, f := range c.Files { @@ -762,16 +856,88 @@ func (s *slskd) Grab( if err := s.client.post( ctx, slskdDownloadsPath(username), wanted, nil, ); err != nil { - return Result{}, err + return "", err } - if err := s.awaitTransfers( - ctx, username, stale, c, onProgress, - ); err != nil { - return Result{}, err + return "", nil +} + +func batchEndpointMissing(err error) bool { + switch statusCode(err) { + case http.StatusBadRequest, http.StatusNotFound, http.StatusMethodNotAllowed: + return true + default: + return false + } +} + +// enqueueBatches enqueues one batch per destination folder. A batch's +// files all land directly in its destination, so a multi-disc rip needs +// one per disc or disc 2's "01" is renamed out of the way of disc 1's. +func (s *slskd) enqueueBatches( + ctx context.Context, + username string, + c Candidate, + dest string, +) error { + groups := map[string][]map[string]any{} + + for _, f := range c.Files { + sub := batchSubfolder(f.Path) + groups[sub] = append(groups[sub], map[string]any{ + "filename": f.Path, + "size": f.Size, + }) } - return s.collect(c, dst) + subs := make([]string, 0, len(groups)) + for sub := range groups { + subs = append(subs, sub) + } + + slices.Sort(subs) + + for _, sub := range subs { + body := map[string]any{ + "username": username, + "files": groups[sub], + "options": map[string]any{"destination": path.Join(dest, sub)}, + } + + if err := s.client.post( + ctx, "/api/v0/transfers/downloads/batches", body, nil, + ); err != nil { + return err + } + } + + return nil +} + +// batchSubfolder is where under a batch's destination a file goes: its +// disc folder, renamed to a form slskd's path sanitising leaves alone +// and ParsePath still reads a disc number from, or nothing. +func batchSubfolder(remote string) string { + norm := strings.ReplaceAll(remote, `\`, "/") + + if n, ok := discFolder(path.Base(path.Dir(norm))); ok { + return "Disc " + strconv.Itoa(n) + } + + return "" +} + +// discardDestination removes a batch's folder once its files have been +// collected or the grab abandoned. It is this grab's own folder under +// slskdDestRoot, so nothing in it belongs to anyone else. +func (s *slskd) discardDestination(dest string) { + if dest == "" || !strings.HasPrefix(dest, slskdDestRoot+"/") { + return + } + + if err := os.RemoveAll(filepath.Join(s.downloadsPath, filepath.FromSlash(dest))); err != nil { + s.logger.Debug("could not remove slskd batch folder", "dest", dest, "error", err) + } } // slskdFolders serialises grabs that land in the same local folder. @@ -1102,10 +1268,17 @@ func (s *slskd) transfersFor( } // collect moves finished files out of slskd's download directory into -// staging. slskd lays them out as //, so each -// wanted file is looked up by its base name under the folder slskd -// derived from the remote path. -func (s *slskd) collect(c Candidate, dst string) (Result, error) { +// staging. +// +// With a batch destination each file is exactly where it was asked to +// go. Without one slskd lays files out as //, +// and arrivedFile has to tell this grab's file from whatever else has +// that name there. +func (s *slskd) collect( + c Candidate, + dst, dest string, + existing map[string]fileStamp, +) (Result, error) { result := Result{Dir: dst, Files: make([]string, 0, len(c.Files))} for _, f := range c.Files { @@ -1113,10 +1286,28 @@ func (s *slskd) collect(c Candidate, dst string) (Result, error) { folder := path.Base(path.Dir(norm)) base := path.Base(norm) - src := filepath.Join(s.downloadsPath, folder, base) + var ( + src string + info os.FileInfo + ok bool + ) - info, err := os.Stat(src) - if err != nil || info.Size() == 0 { + if dest != "" { + src = filepath.Join( + s.downloadsPath, filepath.FromSlash(dest), batchSubfolder(f.Path), base, + ) + + var err error + + info, err = os.Stat(src) + ok = err == nil && info.Size() > 0 + } else { + src, info, ok = arrivedFile( + filepath.Join(s.downloadsPath, folder), base, existing, + ) + } + + if !ok { // Not every requested file arrives; that is expected and // handled by completeness scoring downstream. continue @@ -1126,7 +1317,7 @@ func (s *slskd) collect(c Candidate, dst string) (Result, error) { // Flattened, disc 2's "01 Intro.flac" overwrites disc 1's, and // the importer loses the folder it reads the disc number from. target := filepath.Join(dst, base) - if _, ok := discFolder(folder); ok { + if _, disc := discFolder(folder); disc { target = filepath.Join(dst, folder, base) } @@ -1147,3 +1338,116 @@ func (s *slskd) collect(c Candidate, dst string) (Result, error) { return result, nil } + +// fileStamp is enough of a file to tell whether it has been replaced. +type fileStamp struct { + size int64 + modTime time.Time +} + +// snapshotFolders records the files already in the folders a per-user +// enqueue will write to, so collect does not take one of them for the +// file this grab asked for. A batch writes to a new folder and needs +// none. +func (s *slskd) snapshotFolders(c Candidate) map[string]fileStamp { + if s.batches.Load() == batchesSupported { + return nil + } + + out := map[string]fileStamp{} + + for _, dir := range s.localFolders(c) { + entries, err := os.ReadDir(dir) + if err != nil { + continue + } + + for _, e := range entries { + info, err := e.Info() + if err != nil || !info.Mode().IsRegular() { + continue + } + + out[filepath.Join(dir, e.Name())] = fileStamp{ + size: info.Size(), + modTime: info.ModTime(), + } + } + } + + return out +} + +// arrivedFile finds the file slskd wrote for base in dir. +// +// slskd's default when a name is taken is to write `name_.ext` +// beside it, so a file with that name left by an earlier failed attempt +// — or by the user's own download of the same folder — would otherwise +// be collected while this grab's copy sat beside it under another name. +// A candidate is the name itself or a renamed form of it that was not +// already there, unchanged, before the grab enqueued; the newest wins. +// Comparing against the snapshot rather than a clock matters because +// slskd may run on another machine whose clock is not ours. +func arrivedFile( + dir, base string, + existing map[string]fileStamp, +) (string, os.FileInfo, bool) { + entries, err := os.ReadDir(dir) + if err != nil { + return "", nil, false + } + + ext := filepath.Ext(base) + stem := strings.TrimSuffix(base, ext) + + var ( + best string + bestInfo os.FileInfo + ) + + for _, e := range entries { + name := e.Name() + if name != base && !isRenamedCopy(name, stem, ext) { + continue + } + + info, err := e.Info() + if err != nil || !info.Mode().IsRegular() || info.Size() == 0 { + continue + } + + full := filepath.Join(dir, name) + if was, ok := existing[full]; ok && + was.size == info.Size() && was.modTime.Equal(info.ModTime()) { + continue + } + + if bestInfo == nil || info.ModTime().After(bestInfo.ModTime()) { + best, bestInfo = full, info + } + } + + return best, bestInfo, bestInfo != nil +} + +// isRenamedCopy reports whether name is stem_ext, which is how +// slskd names a download whose name was taken. +func isRenamedCopy(name, stem, ext string) bool { + rest, ok := strings.CutPrefix(name, stem+"_") + if !ok { + return false + } + + digits, ok := strings.CutSuffix(rest, ext) + if !ok || digits == "" { + return false + } + + for _, r := range digits { + if r < '0' || r > '9' { + return false + } + } + + return true +} diff --git a/backend/download/provider_slskd_live_test.go b/backend/download/provider_slskd_live_test.go new file mode 100644 index 0000000..6e521e1 --- /dev/null +++ b/backend/download/provider_slskd_live_test.go @@ -0,0 +1,130 @@ +package download + +import ( + "context" + "os" + "path/filepath" + "slices" + "testing" + "time" +) + +// TestSlskdLive runs the provider against a real slskd daemon. It is +// skipped unless YJ_SLSKD_URL, YJ_SLSKD_API_KEY and YJ_SLSKD_DOWNLOADS +// are set, and it downloads something only when YJ_SLSKD_GRAB=1 — then +// the smallest candidate the search returns, from whichever stranger +// is sharing it. +// +// Everything else here tests the provider against a stub written from +// reading slskd's source. This is where those readings are checked: +// the search options, the responses endpoint, the batch destination, +// the cancel. +// +// YJ_SLSKD_URL=http://localhost:5030 YJ_SLSKD_API_KEY=… \ +// YJ_SLSKD_DOWNLOADS=/path/to/slskd/downloads YJ_SLSKD_GRAB=1 \ +// go test -run TestSlskdLive -v ./backend/download/ +func TestSlskdLive(t *testing.T) { + base, key, downloads := os.Getenv("YJ_SLSKD_URL"), + os.Getenv("YJ_SLSKD_API_KEY"), os.Getenv("YJ_SLSKD_DOWNLOADS") + if base == "" || key == "" || downloads == "" { + t.Skip( + "set YJ_SLSKD_URL, YJ_SLSKD_API_KEY and YJ_SLSKD_DOWNLOADS to run against a real slskd", + ) + } + + query := os.Getenv("YJ_SLSKD_QUERY") + if query == "" { + query = "Radiohead OK Computer" + } + + p, err := newSlskd( + Config{ + ID: 1, Kind: KindSlskd, Name: "live", Enabled: true, + Settings: map[string]string{"url": base, "downloadsPath": downloads}, + }, + func(string) (string, error) { return key, nil }, + slogDiscard(), + ) + if err != nil { + t.Fatalf("newSlskd: %v", err) + } + + s, ok := p.(*slskd) + if !ok { + t.Fatalf("provider is %T", p) + } + + ctx := context.Background() + + if err := s.Check(ctx); err != nil { + t.Fatalf("Check: %v", err) + } + + started := time.Now() + + got, err := s.Search(ctx, Download{Query: query}) + if err != nil { + t.Fatalf("Search: %v", err) + } + + t.Logf( + "search %q: %d candidates in %s", + query, + len(got), + time.Since(started).Round(time.Millisecond), + ) + + if len(got) == 0 { + t.Fatal("no candidates; try a more common YJ_SLSKD_QUERY") + } + + timed := 0 + + for _, c := range got { + for _, f := range c.Files { + if f.LengthMillis > 0 { + timed++ + } + } + } + + t.Logf("%d files carry a length", timed) + + if os.Getenv("YJ_SLSKD_GRAB") != "1" { + return + } + + smallest := slices.MinFunc(got, func(a, b Candidate) int { + return int(a.TotalSize - b.TotalSize) + }) + + t.Logf("grabbing %q from %s (%d files, %d bytes)", + smallest.Title, smallest.Origin, len(smallest.Files), smallest.TotalSize) + + s.stallAfter = 3 * time.Minute + + gctx, cancel := context.WithTimeout(ctx, 15*time.Minute) + defer cancel() + + res, err := s.Grab(gctx, smallest, t.TempDir(), func(p Progress) { + t.Logf("%s: %d/%d bytes", p.Phase, p.Current, p.Total) + }) + if err != nil { + // A stranger going offline is not a defect; what matters is + // that the transfers were cancelled, which slskd's UI shows. + t.Fatalf("Grab: %v", err) + } + + t.Logf("batches: %v", s.batches.Load() == batchesSupported) + + for _, f := range res.Files { + info, err := os.Stat(f) + if err != nil || info.Size() == 0 { + t.Errorf("collected %s is missing or empty: %v", f, err) + } + } + + if entries, _ := os.ReadDir(filepath.Join(downloads, slskdDestRoot)); len(entries) != 0 { + t.Errorf("%d batch folders left in slskd's downloads", len(entries)) + } +} diff --git a/backend/download/provider_slskd_test.go b/backend/download/provider_slskd_test.go index d06cec5..5dd293d 100644 --- a/backend/download/provider_slskd_test.go +++ b/backend/download/provider_slskd_test.go @@ -7,7 +7,9 @@ import ( "net/http" "net/http/httptest" "os" + "path" "path/filepath" + "strconv" "strings" "sync" "testing" @@ -55,6 +57,16 @@ type slskdStub struct { // noResponsesEndpoint makes /searches/{id}/responses 404, as an // older daemon would. noResponsesEndpoint bool + + // batches makes the daemon take batch downloads, as 0.26 does. + // Without it the batch endpoint answers 400, which is what an older + // daemon's per-user route does with a batch body. batchBodies + // records each batch, and delivered is written into its destination + // under downloads when it is enqueued, keyed by file base name. + batches bool + batchBodies []map[string]any + delivered map[string]string + downloads string } func newSlskdStub(t *testing.T) *slskdStub { @@ -138,6 +150,12 @@ func newSlskdStub(t *testing.T) *slskdStub { s.paths = append(s.paths, r.URL.EscapedPath()) s.mu.Unlock() + if r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/batches") { + s.enqueueBatch(t, w, r) + + return + } + switch r.Method { case http.MethodPost: var body []map[string]any @@ -149,6 +167,13 @@ func newSlskdStub(t *testing.T) *slskdStub { s.mu.Lock() s.enqueued = body s.posted = true + + for _, file := range body { + name, _ := file["filename"].(string) + norm := strings.ReplaceAll(name, `\`, "/") + s.write(t, path.Base(path.Dir(norm)), path.Base(norm)) + } + s.mu.Unlock() w.WriteHeader(http.StatusCreated) @@ -202,6 +227,89 @@ func newSlskdStub(t *testing.T) *slskdStub { return s } +// enqueueBatch answers the batch endpoint. +func (s *slskdStub) enqueueBatch(t *testing.T, w http.ResponseWriter, r *http.Request) { + t.Helper() + + s.mu.Lock() + defer s.mu.Unlock() + + if !s.batches { + w.WriteHeader(http.StatusBadRequest) + + return + } + + var body map[string]any + + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Errorf("decode batch body: %v", err) + } + + s.batchBodies = append(s.batchBodies, body) + s.posted = true + + files, _ := body["files"].([]any) + options, _ := body["options"].(map[string]any) + dest, _ := options["destination"].(string) + + for _, f := range files { + file, _ := f.(map[string]any) + s.enqueued = append(s.enqueued, file) + + name, _ := file["filename"].(string) + base := path.Base(strings.ReplaceAll(name, `\`, "/")) + + s.write(t, filepath.FromSlash(dest), base) + } + + w.WriteHeader(http.StatusCreated) +} + +// deliver names the files that arrive once enqueued. +func (s *slskdStub) deliver(names ...string) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.delivered == nil { + s.delivered = map[string]string{} + } + + for _, n := range names { + s.delivered[n] = "audio" + } +} + +// write puts a delivered file where slskd would: under dir in the +// downloads folder, renamed name_.ext when the name is taken, as +// slskd's default Destination.Exists does. Callers hold s.mu. +func (s *slskdStub) write(t *testing.T, dir, base string) { + t.Helper() + + content, ok := s.delivered[base] + if !ok { + return + } + + full := filepath.Join(s.downloads, dir) + if err := os.MkdirAll(full, 0o750); err != nil { + t.Errorf("mkdir: %v", err) + } + + target := filepath.Join(full, base) + if _, err := os.Stat(target); err == nil { + ext := filepath.Ext(base) + target = filepath.Join( + full, + strings.TrimSuffix(base, ext)+"_"+strconv.FormatInt(time.Now().UnixNano(), 10)+ext, + ) + } + + if err := os.WriteFile(target, []byte(content), 0o600); err != nil { + t.Errorf("write: %v", err) + } +} + // reject enforces API-key auth like the real daemon. func (s *slskdStub) reject(w http.ResponseWriter, r *http.Request) bool { s.mu.Lock() @@ -234,6 +342,10 @@ func newStubSlskd(t *testing.T, stub *slskdStub) (*slskd, string) { downloads := t.TempDir() + stub.mu.Lock() + stub.downloads = downloads + stub.mu.Unlock() + p, err := newSlskd( Config{ ID: 1, @@ -471,24 +583,9 @@ func TestSlskdGrabCollectsFromDownloadsFolder(t *testing.T) { }, } - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) - // slskd writes into //. - folder := filepath.Join(downloads, "OK Computer") - if err := os.MkdirAll(folder, 0o750); err != nil { - t.Fatalf("mkdir: %v", err) - } - - for _, name := range []string{ - "01 Airbag.flac", - "02 Paranoid Android.flac", - } { - if err := os.WriteFile( - filepath.Join(folder, name), []byte("audio"), 0o600, - ); err != nil { - t.Fatalf("write: %v", err) - } - } + stub.deliver("01 Airbag.flac", "02 Paranoid Android.flac") c := Candidate{ ID: "slskd:peer:OK Computer", @@ -547,18 +644,9 @@ func TestSlskdGrabToleratesPartialFailure(t *testing.T) { {Filename: `\s\Album\02 B.flac`, State: "Completed, Errored"}, }} - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) - folder := filepath.Join(downloads, "Album") - if err := os.MkdirAll(folder, 0o750); err != nil { - t.Fatalf("mkdir: %v", err) - } - - if err := os.WriteFile( - filepath.Join(folder, "01 A.flac"), []byte("audio"), 0o600, - ); err != nil { - t.Fatalf("write: %v", err) - } + stub.deliver("01 A.flac") c := Candidate{ Files: []CandidateFile{ @@ -643,23 +731,12 @@ func TestSlskdRequiresConfiguration(t *testing.T) { } } -// slskdAlbum is a two-file candidate from peer, with the files slskd -// would have written already in place under downloads. -func slskdAlbum(t *testing.T, downloads, peer string, arrived ...string) Candidate { +// slskdAlbum is a two-file candidate from peer, whose arrived files +// the stub writes where slskd would once they are enqueued. +func slskdAlbum(t *testing.T, stub *slskdStub, peer string, arrived ...string) Candidate { t.Helper() - folder := filepath.Join(downloads, "Album") - if err := os.MkdirAll(folder, 0o750); err != nil { - t.Fatalf("mkdir: %v", err) - } - - for _, name := range arrived { - if err := os.WriteFile( - filepath.Join(folder, name), []byte("audio"), 0o600, - ); err != nil { - t.Fatalf("write: %v", err) - } - } + stub.deliver(arrived...) return Candidate{ Files: []CandidateFile{ @@ -691,11 +768,11 @@ func TestSlskdGrabGivesUpOnAStalledPeer(t *testing.T) { {ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"}, }} - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) s.stallAfter = 30 * time.Millisecond _, err := s.Grab( - context.Background(), slskdAlbum(t, downloads, "peer"), t.TempDir(), nil, + context.Background(), slskdAlbum(t, stub, "peer"), t.TempDir(), nil, ) if !errors.Is(err, ErrSlskdTimeout) { t.Fatalf("error = %v, want ErrSlskdTimeout", err) @@ -728,12 +805,12 @@ func TestSlskdGrabKeepsWhatArrivedBeforeAStall(t *testing.T) { {ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"}, }} - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) s.stallAfter = 30 * time.Millisecond got, err := s.Grab( context.Background(), - slskdAlbum(t, downloads, "peer", "01 A.flac"), + slskdAlbum(t, stub, "peer", "01 A.flac"), t.TempDir(), nil, ) if err != nil { @@ -779,7 +856,7 @@ func TestSlskdGrabWaitsOnATransferThatIsMoving(t *testing.T) { }, }) - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) // A hundred polls take several times the stall window; each one // moves a byte. The window is kept well above one poll so a // descheduled test runner does not read as a stall. @@ -788,7 +865,7 @@ func TestSlskdGrabWaitsOnATransferThatIsMoving(t *testing.T) { got, err := s.Grab( context.Background(), - slskdAlbum(t, downloads, "peer", "01 A.flac", "02 B.flac"), + slskdAlbum(t, stub, "peer", "01 A.flac", "02 B.flac"), t.TempDir(), nil, ) if err != nil { @@ -814,12 +891,12 @@ func TestSlskdGrabCountsAnUnlistedFileAsFailed(t *testing.T) { }, }} - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) s.absentGrace = 20 * time.Millisecond got, err := s.Grab( context.Background(), - slskdAlbum(t, downloads, "peer", "01 A.flac"), + slskdAlbum(t, stub, "peer", "01 A.flac"), t.TempDir(), nil, ) if err != nil { @@ -854,9 +931,9 @@ func TestSlskdGrabIgnoresAnEarlierAttemptsRecord(t *testing.T) { }, } - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) - c := slskdAlbum(t, downloads, "peer", "01 A.flac") + c := slskdAlbum(t, stub, "peer", "01 A.flac") c.Files = c.Files[:1] got, err := s.Grab(context.Background(), c, t.TempDir(), nil) @@ -880,12 +957,12 @@ func TestSlskdGrabCancelsTransfersWhenTheCallerGivesUp(t *testing.T) { {ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"}, }} - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) defer cancel() - _, err := s.Grab(ctx, slskdAlbum(t, downloads, "peer"), t.TempDir(), nil) + _, err := s.Grab(ctx, slskdAlbum(t, stub, "peer"), t.TempDir(), nil) if !errors.Is(err, ErrSlskdTimeout) { t.Fatalf("error = %v, want ErrSlskdTimeout", err) } @@ -912,11 +989,11 @@ func TestSlskdEscapesTheUsername(t *testing.T) { }, }} - s, downloads := newStubSlskd(t, stub) + s, _ := newStubSlskd(t, stub) if _, err := s.Grab( context.Background(), - slskdAlbum(t, downloads, "dj a/b", "01 A.flac", "02 B.flac"), + slskdAlbum(t, stub, "dj a/b", "01 A.flac", "02 B.flac"), t.TempDir(), nil, ); err != nil { t.Fatalf("Grab: %v", err) @@ -927,7 +1004,9 @@ func TestSlskdEscapesTheUsername(t *testing.T) { stub.mu.Unlock() for _, p := range paths { - if p != "/api/v0/transfers/downloads/dj%20a%2Fb" { + // The batch endpoint carries the name in its body. + if p != "/api/v0/transfers/downloads/dj%20a%2Fb" && + p != "/api/v0/transfers/downloads/batches" { t.Errorf("transfers call went to %s", p) } } diff --git a/backend/download/slskd_destination_test.go b/backend/download/slskd_destination_test.go new file mode 100644 index 0000000..921024b --- /dev/null +++ b/backend/download/slskd_destination_test.go @@ -0,0 +1,328 @@ +package download + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// Where slskd writes a grab's files, and how collect finds them (#274). + +// slskd reads searchTimeout in whole seconds, from the last response. +func TestSlskdSearchTimeoutIsInSeconds(t *testing.T) { + t.Parallel() + + cases := []struct { + wait time.Duration + want int + }{ + {20 * time.Second, 18}, + {200 * time.Millisecond, slskdMinSearchTimeout}, + } + + for _, tc := range cases { + s := &slskd{searchWait: tc.wait} + + got, ok := s.searchRequest("id", "text", 2)["searchTimeout"].(int) + if !ok || got != tc.want { + t.Errorf("wait %s: searchTimeout = %v, want %d seconds", tc.wait, got, tc.want) + } + } +} + +func TestIsRenamedCopy(t *testing.T) { + t.Parallel() + + cases := map[string]bool{ + "01 A_638912345678901234.flac": true, + "01 A.flac": false, + "01 A_.flac": false, + "01 A_v2.flac": false, + "01 A_123.mp3": false, + "01 AB_123.flac": false, + } + + for name, want := range cases { + if got := isRenamedCopy(name, "01 A", ".flac"); got != want { + t.Errorf("isRenamedCopy(%q) = %v, want %v", name, got, want) + } + } +} + +func succeeded(names ...string) [][]slskdTransfer { + out := make([]slskdTransfer, 0, len(names)) + for i, n := range names { + out = append(out, slskdTransfer{ + ID: "t" + itoa(i), + Filename: n, + State: "Completed, Succeeded", + BytesTransferred: 500, + }) + } + + return [][]slskdTransfer{out} +} + +// On a daemon with batches, each grab writes into a folder of its own, +// collect reads from exactly there, and the folder is gone afterwards. +func TestSlskdBatchGrabUsesItsOwnFolder(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.batches = true + stub.transfers = succeeded(`\s\Album\01 A.flac`, `\s\Album\02 B.flac`) + + s, downloads := newStubSlskd(t, stub) + + // A same-named file in the folder a per-user enqueue would use is + // someone else's, and must not be touched. + other := filepath.Join(downloads, "Album", "01 A.flac") + if err := os.MkdirAll(filepath.Dir(other), 0o750); err != nil { + t.Fatal(err) + } + + if err := os.WriteFile(other, []byte("the user's"), 0o600); err != nil { + t.Fatal(err) + } + + dst := t.TempDir() + + got, err := s.Grab( + context.Background(), slskdAlbum(t, stub, "peer", "01 A.flac", "02 B.flac"), dst, nil, + ) + if err != nil { + t.Fatalf("Grab: %v", err) + } + + if len(got.Files) != 2 { + t.Fatalf("collected %d files, want 2", len(got.Files)) + } + + stub.mu.Lock() + bodies := append([]map[string]any(nil), stub.batchBodies...) + stub.mu.Unlock() + + if len(bodies) != 1 { + t.Fatalf("%d batches, want 1", len(bodies)) + } + + if bodies[0]["username"] != "peer" { + t.Errorf("batch username = %v", bodies[0]["username"]) + } + + dest, _ := bodies[0]["options"].(map[string]any)["destination"].(string) + if !strings.HasPrefix(dest, slskdDestRoot+"/") { + t.Errorf("destination %q is not under %s", dest, slskdDestRoot) + } + + if _, err := os.Stat(filepath.Join(downloads, filepath.FromSlash(dest))); !os.IsNotExist(err) { + t.Errorf("batch folder left behind: %v", err) + } + + if data, _ := os.ReadFile(other); string(data) != "the user's" { + t.Errorf("the user's own file was taken or changed: %q", data) + } +} + +// A batch's files land flat in its destination, so a two-disc rip is +// two batches, one per disc, or disc 2's "01" is renamed out of the way +// of disc 1's. +func TestSlskdBatchSplitsDiscs(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.batches = true + stub.transfers = succeeded(`\s\Wall\CD1\01 In.flac`, `\s\Wall\CD2\01 Hey You.flac`) + stub.deliver("01 In.flac", "01 Hey You.flac") + + s, _ := newStubSlskd(t, stub) + + dst := t.TempDir() + + got, err := s.Grab(context.Background(), Candidate{ + Files: []CandidateFile{ + {Path: `\s\Wall\CD1\01 In.flac`, Size: 500, IsAudio: true}, + {Path: `\s\Wall\CD2\01 Hey You.flac`, Size: 500, IsAudio: true}, + }, + Payload: map[string]string{"username": "peer"}, + }, dst, nil) + if err != nil { + t.Fatalf("Grab: %v", err) + } + + stub.mu.Lock() + n := len(stub.batchBodies) + stub.mu.Unlock() + + if n != 2 { + t.Errorf("%d batches, want one per disc", n) + } + + for _, want := range []string{ + filepath.Join(dst, "CD1", "01 In.flac"), + filepath.Join(dst, "CD2", "01 Hey You.flac"), + } { + found := false + + for _, f := range got.Files { + found = found || f == want + } + + if !found { + t.Errorf("%s not collected; got %q", want, got.Files) + } + } +} + +// An abandoned batch grab leaves nothing in slskd's folder either. +func TestSlskdBatchFailureDiscardsItsFolder(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.batches = true + stub.transfers = [][]slskdTransfer{{ + {ID: "t0", Filename: `\s\Album\01 A.flac`, State: "Completed, Errored"}, + {ID: "t1", Filename: `\s\Album\02 B.flac`, State: "Completed, Errored"}, + }} + + s, downloads := newStubSlskd(t, stub) + + // The stub delivers the file, as a partial slskd left behind would. + if _, err := s.Grab( + context.Background(), slskdAlbum(t, stub, "peer", "01 A.flac"), t.TempDir(), nil, + ); err == nil { + t.Fatal("Grab succeeded with every transfer failed") + } + + entries, _ := os.ReadDir(filepath.Join(downloads, slskdDestRoot)) + if len(entries) != 0 { + t.Errorf("%d batch folders left behind", len(entries)) + } +} + +// An older daemon answers the batch endpoint with 400; the grab falls +// back to the per-user enqueue, and later grabs do not ask again. +func TestSlskdFallsBackWithoutBatches(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.transfers = succeeded(`\s\Album\01 A.flac`, `\s\Album\02 B.flac`) + + s, _ := newStubSlskd(t, stub) + + for range 2 { + stub.mu.Lock() + stub.posted = false + stub.pollCount = 0 + stub.mu.Unlock() + + if _, err := s.Grab( + context.Background(), slskdAlbum(t, stub, "peer", "01 A.flac"), t.TempDir(), nil, + ); err != nil { + t.Fatalf("Grab: %v", err) + } + } + + stub.mu.Lock() + paths := append([]string(nil), stub.paths...) + stub.mu.Unlock() + + batchCalls := 0 + + for _, p := range paths { + if strings.HasSuffix(p, "/batches") { + batchCalls++ + } + } + + if batchCalls != 1 { + t.Errorf("asked for a batch %d times, want once", batchCalls) + } +} + +// Without batches, a file of the same name already in slskd's folder is +// not this grab's: slskd wrote ours beside it as name_.ext, and +// that is the one collected. +func TestSlskdCollectsTheRenamedCopyNotTheOldFile(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.transfers = succeeded(`\s\Album\01 A.flac`, `\s\Album\02 B.flac`) + + s, downloads := newStubSlskd(t, stub) + + old := filepath.Join(downloads, "Album", "01 A.flac") + if err := os.MkdirAll(filepath.Dir(old), 0o750); err != nil { + t.Fatal(err) + } + + if err := os.WriteFile(old, []byte("left by an earlier attempt"), 0o600); err != nil { + t.Fatal(err) + } + + dst := t.TempDir() + + got, err := s.Grab( + context.Background(), slskdAlbum(t, stub, "peer", "01 A.flac", "02 B.flac"), dst, nil, + ) + if err != nil { + t.Fatalf("Grab: %v", err) + } + + if len(got.Files) != 2 { + t.Fatalf("collected %d files, want 2", len(got.Files)) + } + + data, err := os.ReadFile(filepath.Join(dst, "01 A.flac")) + if err != nil || string(data) != "audio" { + t.Errorf("collected %q, want this grab's file", data) + } + + if data, _ := os.ReadFile(old); string(data) != "left by an earlier attempt" { + t.Error("the file that was already there was moved") + } +} + +// And when this grab's copy never arrived, the old one is not taken in +// its place. +func TestSlskdDoesNotCollectAFileThatWasAlreadyThere(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.transfers = [][]slskdTransfer{ + { + {ID: "t0", Filename: `\s\Album\01 A.flac`, State: "Completed, Errored"}, + { + ID: "t1", + Filename: `\s\Album\02 B.flac`, + State: "Completed, Succeeded", + BytesTransferred: 500, + }, + }, + } + + s, downloads := newStubSlskd(t, stub) + + old := filepath.Join(downloads, "Album", "01 A.flac") + if err := os.MkdirAll(filepath.Dir(old), 0o750); err != nil { + t.Fatal(err) + } + + if err := os.WriteFile(old, []byte("stale"), 0o600); err != nil { + t.Fatal(err) + } + + got, err := s.Grab( + context.Background(), slskdAlbum(t, stub, "peer", "02 B.flac"), t.TempDir(), nil, + ) + if err != nil { + t.Fatalf("Grab: %v", err) + } + + if len(got.Files) != 1 || filepath.Base(got.Files[0]) != "02 B.flac" { + t.Errorf("collected %q, want only 02 B.flac", got.Files) + } +}