Compare commits

...
Author SHA1 Message Date
yonluandClaude Opus 5.5 399dcc05b3 docs(notes): record slskd's API as read from its source
CI / check (push) Skipped
CI / e2e (push) Skipped
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017HJiuc3ZZhxsPXz3ozTirT
2026-09-26 22:11:30 -04:00
yonluandClaude Opus 5.5 ef89707bb1 feat(download): fill in the tracks a nearly complete album is missing
A grab that delivers nine of twelve tracks clears the completeness floor
and is imported, and the other three were never looked for. On Soulseek
that is the commonest way an album ends up almost right: one peer's
folder lacks a track, or one file fails.

After a successful import the manager now compares the tracks the files
were aligned to (ImportResult.Matched, new) with the expected tracklist.
When one to three are missing, and fewer than half, it searches for each
one on its own, as the track's artist and title with the album kept for
ranking. It grabs the first auto-acceptable copy that is not from the
source that already failed to supply it, and is not a delegate, which
would place it in its own library. The candidate is trimmed to the one
file aligned to the track. The import uses the album's own request with
ImportOptions.Only, which skips the completeness check and imports only
a file aligned to the missing track, so it is tagged and placed as part
of the album, and anything else the folder brought is left out.

One attempt per track, and nothing here fails the download: the album
is already imported, so a track that cannot be found is logged on the
job and left. slskd accepts one-file folders for a request that expects
one track, which the per-track search needs.

Closes #276

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017HJiuc3ZZhxsPXz3ozTirT
2026-09-26 22:11:03 -04:00
yonluandClaude Opus 5.5 c88fc1c7c7 feat(download): judge a queued slskd peer by its queue position
No bytes for ten minutes usually means a peer has queued us, and the
stall timer could not tell position 2 from position 400: the first was
abandoned while it was about to start, the second was waited on for ten
minutes for nothing.

While a requested file is "Queued, Remotely", the grab asks slskd for
its place (GET .../downloads/{user}/{id}/position, which asks the peer)
once a minute. A place that improved counts as progress and restarts
the stall clock. A place beyond 50 twice running gives the peer up at
once, and the manager moves to the next copy; two readings because
slskd documents the figure as possibly inaccurate. Waiting in a queue
has an overall ceiling of an hour without a byte, since a queue moving
one place an hour would otherwise hold the grab all day. As with a
stall, files that already arrived still go forward.

Closes #275

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017HJiuc3ZZhxsPXz3ozTirT
2026-09-26 22:07:18 -04:00
yonluandClaude Opus 5.5 7a9dd69d30 fix(download): own folder per slskd grab, search timeout in seconds
Checked against slskd 0.26.0's source rather than a live daemon, which
#267 never had.

slskd reads a search's searchTimeout in seconds, counted from the last
response. We sent milliseconds, telling it a search may idle for five
hours, so a search never completed on its own. It now sends seconds,
with slskd's floor of 5.

slskd writes a finished file to <downloads>/<remote leaf folder>/, and
when a name is taken it writes name_<ticks>.ext beside it. collect found
files by name there, so a file left by an earlier failed attempt, or by
the user's own download, was collected in place of this grab's. A user
who changed slskd's destination setting got nothing collected at all.

slskd 0.26 takes a batch download with an explicit destination. Each
grab now enqueues batches into yellowjacket/<uuid>/ (one per disc, since
a batch's files land flat), collects from exactly there, and removes the
folder afterwards, including after a failure. An older daemon answers
the batch route with 400, which is remembered, and the per-user enqueue
is used. There, collect skips files that were already present,
unchanged, before the enqueue, and takes the renamed copy slskd wrote
instead. The comparison is against a snapshot, not a clock, because
slskd may run on another machine. The per-folder lock from #272 is kept
only while batches are not known to work.

The test stub now writes files when they are enqueued, as slskd does,
including the rename, so tests no longer stage files before a grab.
An opt-in TestSlskdLive runs against a real daemon when YJ_SLSKD_URL,
YJ_SLSKD_API_KEY and YJ_SLSKD_DOWNLOADS are set.

Closes #274

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017HJiuc3ZZhxsPXz3ozTirT
2026-09-26 22:04:50 -04:00
yonlu 3f23bb4396 Merge pull request 'Batch: explore catalog + UI, Go 1.26, Soulseek downloads, CI fixes' (#273) from batch/258-272 into main
CI / check (push) Successful in 4m8s
CI / e2e (push) Successful in 14m37s
Reviewed-on: #273
2026-09-27 01:45:18 +00:00
12 changed files with 1692 additions and 97 deletions
+33
View File
@@ -5097,3 +5097,36 @@ beside what they explain. These three did not:
The drawer-style gutter would buy the affordance by taking width off a
full-screen surface on a 424px viewport; back and a 44px close button
answer it instead.
## slskd's API, read from its source rather than a live daemon (2026-09-26)
`backend/download/provider_slskd.go` had never run against a real slskd
when #263–#272 shipped, so its assumptions were checked against slskd
0.26.0's source. One was wrong, and one design was only safe by luck.
These are properties of someone else's server; re-check on an upgrade.
`TestSlskdLive` (env-gated, see its comment) is the way to confirm them
against a running one.
- **`searchTimeout` is seconds, from the last response**, minimum 5
(`SearchRequest.cs`). We sent milliseconds (#274). The other search
options — `responseLimit`, `fileLimit`, `filterResponses`,
`minimumResponseFileCount`, `maximumPeerQueueLength` — are named as we
send them; slskd's defaults are 100 responses, 10 000 files, queue
1 000 000.
- **`GET /searches/{id}/responses` exists**, and `DELETE
/transfers/downloads/{user}/{id}?remove=true` cancels and removes.
- **A finished download is moved to `<downloads>/<Subdirectory>/`**,
where `Destination.Subdirectory` defaults to `${SOURCE_DIRECTORY}`
(the remote leaf folder) and is user-configurable. A taken name is
written as `name_<ticks>.ext` (`Destination.Exists = rename`, the
default). No transfer record says where the file went.
- **Batch enqueue (`POST /transfers/downloads/batches`) is new in 0.26.0**
and is the only way to choose where a file lands: `options.destination`
overrides the subdirectory pattern. A batch's files land flat in it,
so one batch per disc. On an older daemon that path is routed to the
per-user enqueue as username "batches" and the object body is
rejected with 400 — which is why 400 means "no batches" here.
- **`GET .../downloads/{user}/{id}/position` asks the peer** and returns
a bare integer. slskd's own comment on `PlaceInQueue` is "may be
wildly innacurate to the point of uselessness", which is why #275 acts
only on two readings in a row.
+26 -6
View File
@@ -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.
+1 -1
View File
@@ -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)
}
+190
View File
@@ -0,0 +1,190 @@
package download
import (
"cmp"
"context"
"fmt"
"strings"
"yellowjacket/backend/jobs"
)
// Filling in an almost-complete album (#276).
//
// A grab that delivers nine of twelve tracks clears the completeness
// floor and is imported, and before this the other three were never
// looked for. On Soulseek that is the commonest way an album ends up
// almost right: one peer's folder is missing a track, or one file
// failed. So after an import, each missing track is searched for on
// its own and fetched from somewhere else, into the same album.
// maxFillInTracks bounds how many tracks are fetched one by one. An
// album missing more than a few is a different candidate's job, not a
// dozen single-track grabs.
const maxFillInTracks = 3
// fillIn fetches the tracks a successful import did not deliver. It
// never fails the download: the album is already imported, and a track
// it cannot find is logged and left.
func (m *Manager) fillIn(
ctx context.Context,
dl Download,
main Candidate,
imported ImportResult,
job *jobs.Handle,
) {
missing := missingTracks(dl, imported.Matched)
if len(missing) == 0 {
return
}
for _, t := range missing {
if ctx.Err() != nil {
return
}
paths, err := m.fillInTrack(ctx, dl, main, t, job)
if err != nil {
m.logger.Info(
"could not fill in a missing track",
"download", dl.ID,
"track", t.Title,
"error", err,
)
if job != nil {
job.Logf(jobs.LevelWarn, fmt.Sprintf(
"Could not find %q elsewhere: %v", t.Title, err,
))
}
continue
}
if job != nil {
job.Logf(jobs.LevelInfo, fmt.Sprintf(
"Filled in %q from another source (%d file)", t.Title, len(paths),
))
}
}
}
// missingTracks is what an import left out, when filling it in is
// worth trying: an album with a tracklist, a few tracks short.
func missingTracks(dl Download, matched []ExpectedTrack) []ExpectedTrack {
// A recording request is one track; there is no album to complete.
if dl.RecordingMBID != "" || len(dl.Expected) < 2 {
return nil
}
have := make(map[trackKey]bool, len(matched))
for _, t := range matched {
have[keyOf(t)] = true
}
var missing []ExpectedTrack
for _, t := range dl.Expected {
if !have[keyOf(t)] {
missing = append(missing, t)
}
}
// A half-empty album was a poor copy, not a nearly complete one.
if len(missing) > maxFillInTracks || 2*len(missing) >= len(dl.Expected) {
return nil
}
return missing
}
// fillInTrack searches for one track and grabs the first acceptable
// copy that is not from the source that already failed to supply it.
// One attempt: a fill-in that walks a candidate list per track would
// multiply a download's grabs by the number of gaps.
func (m *Manager) fillInTrack(
ctx context.Context,
dl Download,
main Candidate,
t ExpectedTrack,
job *jobs.Handle,
) ([]string, error) {
want := trackRequest(dl, t)
ranked, err := m.Search(ctx, want)
if err != nil {
return nil, err
}
prefs := m.preferences()
for _, c := range ranked {
if ruledOutBy(c, []Candidate{main}) || !autoAcceptable(want, c, prefs) {
continue
}
// The track is being put into an album this app placed; a
// delegate would put it in its own library instead.
if plan, err := m.planTransfer(want, c); err != nil || plan.delegated() {
continue
}
narrowed, ok := narrowTo(c, t)
if !ok {
continue
}
out := m.attemptGrab(ctx, dl, narrowed, job, []ExpectedTrack{t})
if out.item.StagingDir != "" {
if err := m.staging.Release(out.item.StagingDir); err != nil {
m.logger.Warn("could not release staging dir", "error", err)
}
}
if out.err != nil {
return nil, out.err
}
if err := m.store.SetItemImported(
ctx, out.item.ID, out.imported.Paths,
); err != nil {
m.logger.Warn("could not record imported paths", "error", err)
}
return out.imported.Paths, nil
}
return nil, ErrNoCandidates
}
// trackRequest is the search for one track of an album: the track's
// artist and title as the query, the album kept so a copy from that
// album outranks the same song off a compilation, and one expected
// track so a single file is a complete answer.
func trackRequest(dl Download, t ExpectedTrack) Download {
artist := cmp.Or(t.Artist, dl.Artist)
want := dl
want.Query = strings.TrimSpace(artist + " " + t.Title)
want.Expected = []ExpectedTrack{t}
return want
}
// narrowTo trims a candidate to the one file that aligns to t, so the
// grab fetches a track rather than whatever else the folder offered.
func narrowTo(c Candidate, t ExpectedTrack) (Candidate, bool) {
aligned, _ := matchFiles(c.Files, []ExpectedTrack{t})
for _, f := range aligned {
if f.IsAudio && f.MatchedTo == t.Position {
c.Files = []CandidateFile{f}
c.TotalSize = f.Size
return c, true
}
}
return Candidate{}, false
}
+168
View File
@@ -0,0 +1,168 @@
package download
import (
"context"
"errors"
"os"
"path/filepath"
"testing"
)
// Filling in the tracks an almost-complete album is missing (#276).
func fiveTrackTitles() []string {
return append(allTitles(), "Let Down")
}
func fiveTrackDownload() Download {
dl := fourTrackDownload()
dl.Expected = append(dl.Expected, ExpectedTrack{Position: 5, Title: "Let Down"})
return dl
}
func TestMissingTracks(t *testing.T) {
t.Parallel()
dl := fiveTrackDownload()
got := func(positions ...int) []ExpectedTrack {
out := make([]ExpectedTrack, 0, len(positions))
for _, p := range positions {
out = append(out, dl.Expected[p-1])
}
return out
}
cases := []struct {
name string
dl Download
matched []ExpectedTrack
want int
}{
{"complete", dl, got(1, 2, 3, 4, 5), 0},
{"one short", dl, got(1, 2, 3, 4), 1},
{"two short", dl, got(1, 2, 3), 2},
{"half gone is a poor copy", dl, got(1, 2), 0},
{"a recording is not an album", func() Download {
d := dl
d.RecordingMBID = "rec"
return d
}(), got(1, 2, 3, 4), 0},
}
for _, tc := range cases {
if n := len(missingTracks(tc.dl, tc.matched)); n != tc.want {
t.Errorf("%s: %d missing, want %d", tc.name, n, tc.want)
}
}
big := fiveTrackDownload()
for i := 6; i <= 20; i++ {
big.Expected = append(big.Expected, ExpectedTrack{Position: i, Title: "T" + itoa(i)})
}
if n := len(missingTracks(big, big.Expected[:16])); n != 0 {
t.Errorf("four of twenty missing: %d filled in, want none past %d", n, maxFillInTracks)
}
}
// A fill-in import takes only the file that is the missing track, and
// places it in the album with the album's tags; anything else the grab
// brought is left out.
func TestImportOnlyTakesTheMissingTrack(t *testing.T) {
t.Parallel()
f := newImportFixture(t,
"05 - Let Down.flac",
"02 - Paranoid Android.flac",
)
dl := fiveTrackDownload()
got, err := f.importer.Import(
context.Background(), dl,
Result{Dir: f.dir, Files: f.files},
ImportOptions{LibraryRoot: f.root, WriteTags: true, Only: dl.Expected[4:]},
)
if err != nil {
t.Fatalf("Import: %v", err)
}
want := filepath.Join(f.root, "Radiohead", "OK Computer", "05 Let Down.flac")
if len(got.Paths) != 1 || got.Paths[0] != want {
t.Errorf("imported %q, want only %s", got.Paths, want)
}
// A file that is not the missing track is not imported at all.
g := newImportFixture(t, "02 - Paranoid Android.flac")
if _, err := g.importer.Import(
context.Background(), dl,
Result{Dir: g.dir, Files: g.files},
ImportOptions{LibraryRoot: g.root, WriteTags: true, Only: dl.Expected[4:]},
); !errors.Is(err, ErrTooIncomplete) {
t.Errorf("Import = %v, want nothing matched", err)
}
}
// The album comes from one source missing its fifth track; the fifth is
// then found on its own at another and lands in the same album.
func TestManagerFillsInAMissingTrack(t *testing.T) {
t.Parallel()
f := newManagerFixture(t)
titles := fiveTrackTitles()
album := NewFakeProvider(1, "album", Caps{CanSearch: true, CanTransport: true})
ac := candidateFor("album-cand", titles, ".flac", 30_000_000)
ac.ProviderID = 1
album.Candidates = []Candidate{ac}
for i, tt := range titles[:4] {
album.Written[trackToken(i+1)+" - "+tt+".flac"] = []byte("audio-data")
}
single := NewFakeProvider(2, "single", Caps{CanSearch: true, CanTransport: true})
sc := Candidate{
ID: "single-cand",
Protocol: ProtocolDirect,
Title: "Radiohead - OK Computer",
Artist: "Radiohead",
Files: []CandidateFile{{
Path: "Radiohead - OK Computer/05 - Let Down.flac",
Size: 30_000_000,
}},
Health: 0.5,
ProviderID: 2,
}
single.Candidates = []Candidate{sc}
single.Written["05 - Let Down.flac"] = []byte("audio-data")
f.manager.installProvider(Config{ID: 1, Priority: 90}, album)
f.manager.installProvider(Config{ID: 2, Priority: 10}, single)
dl := fiveTrackDownload()
if _, err := f.manager.Start(context.Background(), dl); err != nil {
t.Fatalf("Start: %v", err)
}
waitForDownloadState(t, f.store, dl.ID, StateComplete)
if album.GrabCallCount() != 1 || single.GrabCallCount() != 1 {
t.Errorf(
"grabs: album=%d single=%d, want 1 and 1",
album.GrabCallCount(), single.GrabCallCount(),
)
}
for i, tt := range titles {
p := filepath.Join(f.root, "Radiohead", "OK Computer", trackToken(i+1)+" "+tt+".flac")
if _, err := os.Stat(p); err != nil {
t.Errorf("track %d not in the library: %v", i+1, err)
}
}
}
+61 -2
View File
@@ -79,6 +79,12 @@ type ImportOptions struct {
// them. Off for delegate providers, which have already imported
// and tagged the files themselves.
WriteTags bool
// Only, when set, imports just the files that align to these
// tracks of the download and skips the completeness check: it is a
// fill-in for tracks an earlier grab of the same album did not
// deliver (#276), tagged and placed as part of that album.
Only []ExpectedTrack
}
// DefaultPathTemplate is the layout used when none is configured.
@@ -118,6 +124,10 @@ type ImportResult struct {
// Skipped counts non-audio files left in staging (logs, cue sheets,
// scene .nfo files) — deliberately not imported.
Skipped int
// Matched are the expected tracks an imported file was aligned to,
// which is how a caller learns what the grab did not deliver.
Matched []ExpectedTrack
}
// Import verifies, tags and moves a completed grab into the library.
@@ -140,14 +150,25 @@ func (i *Importer) Import(
return ImportResult{}, ErrNoAudio
}
if err := checkCompleteness(len(audio), dl); err != nil {
return ImportResult{}, err
if len(opts.Only) == 0 {
if err := checkCompleteness(len(audio), dl); err != nil {
return ImportResult{}, err
}
}
// Align staged files to the expected tracklist so tags and
// filenames reflect the release, not the uploader's naming.
plan := i.planFiles(audio, dl)
if len(opts.Only) > 0 {
plan = onlyTracks(plan, opts.Only)
if len(plan) == 0 {
return ImportResult{}, fmt.Errorf(
"%w: no file matched the missing track", ErrTooIncomplete,
)
}
}
out := ImportResult{
Paths: make([]string, 0, len(plan)),
Skipped: skipped,
@@ -184,11 +205,49 @@ func (i *Importer) Import(
}
out.Paths = append(out.Paths, dest)
if p.Matched {
out.Matched = append(out.Matched, p.Track)
}
}
return out, nil
}
// trackKey identifies an expected track within a release.
type trackKey struct{ disc, position int }
func keyOf(t ExpectedTrack) trackKey {
return trackKey{disc: t.DiscNumber, position: t.Position}
}
// onlyTracks keeps the planned files aligned to one of want. A fill-in
// grab can bring more than the one file it was after — a folder where
// the title also matched a live take — and anything else would land in
// the album as a duplicate or a stranger.
func onlyTracks(plan []plannedFile, want []ExpectedTrack) []plannedFile {
keys := make(map[trackKey]bool, len(want))
for _, t := range want {
keys[keyOf(t)] = true
}
out := make([]plannedFile, 0, len(want))
seen := map[trackKey]bool{}
for _, p := range plan {
k := keyOf(p.Track)
if !p.Matched || !keys[k] || seen[k] {
continue
}
seen[k] = true
out = append(out, p)
}
return out
}
// plannedFile pairs a staged file with the expected track it matched.
type plannedFile struct {
Source string
+4 -1
View File
@@ -747,8 +747,9 @@ func (m *Manager) grab(
var failed []Candidate
for {
out := m.attemptGrab(ctx, dl, c, job)
out := m.attemptGrab(ctx, dl, c, job, nil)
if out.err == nil {
m.fillIn(ctx, dl, c, out.imported, job)
m.finishGrab(ctx, dl, out.item, out.imported, job)
return
@@ -836,6 +837,7 @@ func (m *Manager) attemptGrab(
dl Download,
c Candidate,
job *jobs.Handle,
only []ExpectedTrack,
) grabOutcome {
// Who will move the bytes is decided before any slot is taken, so
// the transfer waits in its own provider's queue rather than in a
@@ -942,6 +944,7 @@ func (m *Manager) attemptGrab(
opts := m.importOptions()
opts.WriteTags = true
opts.Only = only
opts.LibraryRoot, err = m.library.LibraryPath(dl.LibraryID)
if err != nil {
+458 -29
View File
@@ -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
@@ -88,7 +95,8 @@ const (
// it covers a peer that queues us and never starts as well as one
// that starts and stops. Ten minutes is long enough for a short
// queue ahead of us to clear and short enough that one unresponsive
// peer does not hold slskd's single transfer slot for an evening.
// peer does not hold a transfer slot for an evening. A queue whose
// position improves restarts it (see queueWatch).
slskdStallAfter = 10 * time.Minute
// slskdAbsentGrace is how long a requested file may be missing from
@@ -97,6 +105,23 @@ const (
// a few polls was refused.
slskdAbsentGrace = 30 * time.Second
// slskdPositionPoll is how often a peer that has queued us is asked
// where we are in its queue. Each ask is a message to the peer, so
// it is far slower than the transfer poll.
slskdPositionPoll = time.Minute
// slskdMaxQueuePosition is the queue position past which a peer is
// not worth waiting for: at a few minutes a track, fifty albums
// ahead of us is days. slskd warns the figure can be inaccurate, so
// it takes two readings in a row to act on (see awaitTransfers).
slskdMaxQueuePosition = 50
// slskdQueueCeiling is the longest a grab waits in a peer's queue
// without a byte arriving, however steadily the queue moves. An
// improving position restarts the stall clock, so without this a
// queue moving one place an hour would hold the grab all day.
slskdQueueCeiling = time.Hour
// slskdCancelTimeout bounds the cleanup that cancels abandoned
// transfers.
slskdCancelTimeout = 15 * time.Second
@@ -161,6 +186,12 @@ type slskd struct {
transferPoll time.Duration
stallAfter time.Duration
absentGrace time.Duration
positionPoll time.Duration
queueCeiling 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.
@@ -217,6 +248,8 @@ func newSlskd(
transferPoll: slskdTransferPoll,
stallAfter: slskdStallAfter,
absentGrace: slskdAbsentGrace,
positionPoll: slskdPositionPoll,
queueCeiling: slskdQueueCeiling,
}, nil
}
@@ -289,6 +322,12 @@ type slskdTransfer struct {
BytesTransferred int64 `json:"bytesTransferred"`
}
// remotelyQueued reports whether the peer has accepted the request and
// put it in its upload queue, where it waits for a slot.
func (t slskdTransfer) remotelyQueued() bool {
return strings.Contains(t.State, "Queued") && strings.Contains(t.State, "Remotely")
}
// done reports whether the transfer reached a terminal state, and
// whether it succeeded. slskd reports compound states such as
// "Completed, Succeeded" and "Completed, Errored".
@@ -431,14 +470,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,
@@ -530,8 +573,10 @@ func isVariousArtists(artist string) bool {
// usually matches one file per folder. The two-file floor that filters
// out one-file noise for an album therefore filtered out every result
// for a track, and a single-track request could never be served here.
// A request expecting one track — a recording, or the fill-in for one
// missing from an album (#276) — takes a one-file folder.
func minFilesFor(dl Download) int {
if dl.RecordingMBID != "" {
if dl.RecordingMBID != "" || len(dl.Expected) == 1 {
return 1
}
@@ -736,20 +781,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
// `<downloads>/<remote leaf folder>/`, shared with every other download
// of a same-named folder and with the user's own, renamed with a
// `_<ticks>` 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 +886,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.
@@ -862,11 +1058,12 @@ func (s *slskd) terminalTransferIDs(
// state, the transfer stalls, or the caller gives up.
//
// Soulseek queues are measured in hours, so there is no deadline on the
// transfer as a whole — but there is one on *progress*. slskd's
// transfer limit is one, so a peer that holds us in its queue without
// sending a byte is not only failing this download, it is holding every
// other Soulseek download behind it. After stallAfter with nothing
// moving the peer is given up on, and the manager tries another.
// transfer as a whole — but there is one on *progress*. A peer that
// holds us without sending a byte is failing this download and holding
// one of the daemon's few transfer slots. After stallAfter with nothing
// moving the peer is given up on, and the manager tries another; a peer
// that has queued us is judged by its queue position as well (see
// queueWatch).
//
// Whatever way this ends short of every file finishing, the transfers
// still live in slskd are cancelled there. Returning without doing so
@@ -889,6 +1086,7 @@ func (s *slskd) awaitTransfers(
lastProgress = started
lastBytes int64
live []slskdTransfer
queue queueWatch
)
for {
@@ -930,6 +1128,28 @@ func (s *slskd) awaitTransfers(
lastProgress = time.Now()
}
if err := queue.observe(ctx, s, username, tally.live, &lastProgress); err != nil {
s.cancelTransfers(username, live)
// As with a stall: what already arrived goes forward.
if tally.done > 0 {
s.logger.Info("slskd queue too long; keeping what arrived", "error", err)
return nil
}
return err
}
if tally.bytes == 0 && time.Since(started) >= s.queueCeiling {
s.cancelTransfers(username, live)
return fmt.Errorf(
"%w: %s sent nothing in %s",
ErrSlskdTimeout, username, s.queueCeiling,
)
}
if onProgress != nil {
onProgress(Progress{
Current: tally.bytes,
@@ -982,6 +1202,77 @@ func (s *slskd) awaitTransfers(
}
}
// queueWatch follows our place in a peer's upload queue while nothing is
// arriving (#275).
//
// No bytes for stallAfter usually means the peer has queued us, and the
// timer alone cannot tell position 2 from position 400. So a queued
// grab asks where it stands every positionPoll: a place that improved is
// progress and restarts the stall clock, and a place past
// slskdMaxQueuePosition twice running gives the peer up at once, so the
// manager moves to the next copy without waiting out the timer.
type queueWatch struct {
lastAsk time.Time
lastPlace int
far int
}
func (q *queueWatch) observe(
ctx context.Context,
s *slskd,
username string,
live []slskdTransfer,
lastProgress *time.Time,
) error {
var queued *slskdTransfer
for i := range live {
if live[i].remotelyQueued() && live[i].ID != "" {
queued = &live[i]
break
}
}
if queued == nil || time.Since(q.lastAsk) < s.positionPoll {
return nil
}
q.lastAsk = time.Now()
var place int
if err := s.client.get(
ctx, slskdDownloadsPath(username)+"/"+url.PathEscape(queued.ID)+"/position", &place,
); err != nil {
// The peer may simply not answer; the stall timer still applies.
s.logger.Debug("slskd queue position unavailable", "peer", username, "error", err)
return nil
}
if q.lastPlace > 0 && place > 0 && place < q.lastPlace {
*lastProgress = time.Now()
}
q.lastPlace = place
if place > slskdMaxQueuePosition {
q.far++
} else {
q.far = 0
}
if q.far >= 2 {
return fmt.Errorf(
"%w: %s has us at position %d in its queue",
ErrSlskdTimeout, username, place,
)
}
return nil
}
// transferTally is one poll's reading of the files a grab asked for.
type transferTally struct {
done, failed int
@@ -1102,10 +1393,17 @@ func (s *slskd) transfersFor(
}
// collect moves finished files out of slskd's download directory into
// staging. slskd lays them out as <downloads>/<folder>/<file>, 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 <downloads>/<folder>/<file>,
// 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 +1411,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 +1442,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 +1463,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_<ticks>.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_<digits>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
}
@@ -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))
}
}
+161 -58
View File
@@ -7,7 +7,9 @@ import (
"net/http"
"net/http/httptest"
"os"
"path"
"path/filepath"
"strconv"
"strings"
"sync"
"testing"
@@ -55,6 +57,21 @@ 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
// positions is what the queue-position endpoint answers, in order;
// the last repeats. positionAsks counts the calls.
positions []int
positionAsks int
delivered map[string]string
downloads string
}
func newSlskdStub(t *testing.T) *slskdStub {
@@ -138,6 +155,29 @@ func newSlskdStub(t *testing.T) *slskdStub {
s.paths = append(s.paths, r.URL.EscapedPath())
s.mu.Unlock()
if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/position") {
s.mu.Lock()
idx := min(s.positionAsks, len(s.positions)-1)
s.positionAsks++
place := 0
if idx >= 0 {
place = s.positions[idx]
}
s.mu.Unlock()
writeJSON(t, w, place)
return
}
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 +189,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 +249,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_<ticks>.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 +364,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,
@@ -267,6 +401,8 @@ func newStubSlskd(t *testing.T, stub *slskdStub) (*slskd, string) {
// tests about stalls and absences set their own.
s.stallAfter = time.Minute
s.absentGrace = time.Minute
s.positionPoll = time.Millisecond
s.queueCeiling = time.Hour
return s, downloads
}
@@ -471,24 +607,9 @@ func TestSlskdGrabCollectsFromDownloadsFolder(t *testing.T) {
},
}
s, downloads := newStubSlskd(t, stub)
s, _ := newStubSlskd(t, stub)
// slskd writes into <downloads>/<folder>/<file>.
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 +668,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 +755,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 +792,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 +829,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 +880,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 +889,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 +915,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 +955,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 +981,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 +1013,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 +1028,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)
}
}
+328
View File
@@ -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_<ticks>.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)
}
}
+132
View File
@@ -0,0 +1,132 @@
package download
import (
"context"
"errors"
"strings"
"testing"
"time"
)
// A peer that has queued us is judged by where we are in its queue, not
// only by a timer (#275).
func queuedThen(polls int, final string) [][]slskdTransfer {
queued := []slskdTransfer{
{ID: "t0", Filename: `\s\Album\01 A.flac`, State: "Queued, Remotely"},
{ID: "t1", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"},
}
out := make([][]slskdTransfer, 0, polls+1)
for range polls {
out = append(out, queued)
}
return append(out, []slskdTransfer{
{ID: "t0", Filename: `\s\Album\01 A.flac`, State: final, BytesTransferred: 500},
{ID: "t1", Filename: `\s\Album\02 B.flac`, State: final, BytesTransferred: 500},
})
}
func descending(from int) []int {
out := make([]int, 0, from)
for p := from; p >= 1; p-- {
out = append(out, p)
}
return out
}
// Two readings far back in the queue give the peer up at once, rather
// than after the stall timer.
func TestSlskdGivesUpOnALongQueue(t *testing.T) {
t.Parallel()
stub := newSlskdStub(t)
stub.transfers = queuedThen(100_000, "Completed, Succeeded")
stub.positions = []int{400}
s, _ := newStubSlskd(t, stub)
s.stallAfter = time.Hour
started := time.Now()
_, err := s.Grab(context.Background(), slskdAlbum(t, stub, "peer"), t.TempDir(), nil)
if !errors.Is(err, ErrSlskdTimeout) || !strings.Contains(err.Error(), "position 400") {
t.Fatalf("Grab = %v, want a queue-position give-up", err)
}
if time.Since(started) > 5*time.Second {
t.Error("the give-up waited on something other than the position")
}
if len(stub.cancelledURIs()) == 0 {
t.Error("the queued transfers were not cancelled")
}
}
// One far reading is not enough: slskd says the figure can be wildly
// wrong.
func TestSlskdOneBadPositionIsNotEnough(t *testing.T) {
t.Parallel()
stub := newSlskdStub(t)
stub.transfers = queuedThen(20, "Completed, Succeeded")
stub.positions = []int{400, 3}
s, _ := newStubSlskd(t, stub)
s.positionPoll = 0
if _, err := s.Grab(
context.Background(),
slskdAlbum(t, stub, "peer", "01 A.flac", "02 B.flac"),
t.TempDir(), nil,
); err != nil {
t.Fatalf("Grab: %v", err)
}
}
// A queue that is moving is progress: the grab outlives the stall timer
// while its position improves.
func TestSlskdAMovingQueueIsProgress(t *testing.T) {
t.Parallel()
stub := newSlskdStub(t)
// Near enough to wait for, with more improving readings (45) than
// there are queued polls (30), so the queue outlives the stall timer
// (30 polls of at least 2 ms against 40 ms) while still improving,
// however slowly the machine runs the loop.
stub.transfers = queuedThen(30, "Completed, Succeeded")
stub.positions = descending(slskdMaxQueuePosition - 5)
s, _ := newStubSlskd(t, stub)
s.transferPoll = 2 * time.Millisecond
s.positionPoll = 0
s.stallAfter = 40 * time.Millisecond
if _, err := s.Grab(
context.Background(),
slskdAlbum(t, stub, "peer", "01 A.flac", "02 B.flac"),
t.TempDir(), nil,
); err != nil {
t.Fatalf("Grab: %v; a moving queue was treated as a stall", err)
}
}
// However steadily the queue moves, waiting in it has a ceiling.
func TestSlskdQueueHasACeiling(t *testing.T) {
t.Parallel()
stub := newSlskdStub(t)
stub.transfers = queuedThen(100_000, "Completed, Succeeded")
stub.positions = descending(100_000)
s, _ := newStubSlskd(t, stub)
s.stallAfter = time.Hour
s.queueCeiling = 50 * time.Millisecond
_, err := s.Grab(context.Background(), slskdAlbum(t, stub, "peer"), t.TempDir(), nil)
if !errors.Is(err, ErrSlskdTimeout) {
t.Fatalf("Grab = %v, want the queue ceiling", err)
}
}