diff --git a/backend/download/provider_slskd.go b/backend/download/provider_slskd.go index 02d1adb..ffe0542 100644 --- a/backend/download/provider_slskd.go +++ b/backend/download/provider_slskd.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log/slog" + "net/url" "os" "path" "path/filepath" @@ -75,6 +76,24 @@ const ( // slskdHTTPTimeout bounds one API call. slskdHTTPTimeout = 20 * time.Second + + // slskdStallAfter is how long a grab may go without a byte arriving + // before the peer is given up on. It is measured from enqueue, so + // 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. + slskdStallAfter = 10 * time.Minute + + // slskdAbsentGrace is how long a requested file may be missing from + // slskd's transfer list before it is counted as failed. slskd lists + // a transfer as soon as it accepts it, so a file still absent after + // a few polls was refused. + slskdAbsentGrace = 30 * time.Second + + // slskdCancelTimeout bounds the cleanup that cancels abandoned + // transfers. + slskdCancelTimeout = 15 * time.Second ) func init() { @@ -134,6 +153,8 @@ type slskd struct { searchPoll time.Duration searchWait time.Duration transferPoll time.Duration + stallAfter time.Duration + absentGrace time.Duration } // newSlskd builds the provider from config. @@ -188,6 +209,8 @@ func newSlskd( searchPoll: slskdSearchPoll, searchWait: slskdSearchWait, transferPoll: slskdTransferPoll, + stallAfter: slskdStallAfter, + absentGrace: slskdAbsentGrace, }, nil } @@ -467,6 +490,15 @@ func (s *slskd) Grab( ) } + // 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. + stale := s.terminalTransferIDs(ctx, username) + wanted := make([]map[string]any, 0, len(c.Files)) for _, f := range c.Files { wanted = append(wanted, map[string]any{ @@ -476,24 +508,69 @@ func (s *slskd) Grab( } if err := s.client.post( - ctx, "/api/v0/transfers/downloads/"+username, wanted, nil, + ctx, slskdDownloadsPath(username), wanted, nil, ); err != nil { return Result{}, err } - if err := s.awaitTransfers(ctx, username, c, onProgress); err != nil { + if err := s.awaitTransfers( + ctx, username, stale, c, onProgress, + ); err != nil { return Result{}, err } return s.collect(c, dst) } +// slskdDownloadsPath is the transfers endpoint for one peer. Soulseek +// usernames may contain spaces and punctuation, so the name is escaped +// rather than spliced into the path. +func slskdDownloadsPath(username string) string { + return "/api/v0/transfers/downloads/" + url.PathEscape(username) +} + +// terminalTransferIDs returns the ids of this peer's transfers that are +// already finished. Best effort: slskd answers 404 for a peer it has no +// transfers with, and any failure here means only that there is nothing +// to ignore. +func (s *slskd) terminalTransferIDs( + ctx context.Context, + username string, +) map[string]bool { + transfers, err := s.transfersFor(ctx, username) + if err != nil { + return nil + } + + out := make(map[string]bool, len(transfers)) + + for _, t := range transfers { + if finished, _ := t.done(); finished && t.ID != "" { + out[t.ID] = true + } + } + + return out +} + // awaitTransfers polls until every requested file reaches a terminal -// state. Soulseek queues are measured in hours, so the only deadline -// is the caller's context. +// 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. +// +// Whatever way this ends short of every file finishing, the transfers +// still live in slskd are cancelled there. Returning without doing so +// leaves the daemon downloading into its own folder for a request +// nobody is waiting on any more. func (s *slskd) awaitTransfers( ctx context.Context, username string, + stale map[string]bool, c Candidate, onProgress ProgressFunc, ) error { @@ -502,9 +579,18 @@ func (s *slskd) awaitTransfers( wanted[f.Path] = true } + var ( + started = time.Now() + lastProgress = started + lastBytes int64 + live []slskdTransfer + ) + for { select { case <-ctx.Done(): + s.cancelTransfers(username, live) + return fmt.Errorf("%w: transfer cancelled", ErrSlskdTimeout) case <-time.After(s.transferPoll): } @@ -512,60 +598,177 @@ func (s *slskd) awaitTransfers( transfers, err := s.transfersFor(ctx, username) if err != nil { // A blip talking to the daemon should not abandon a - // transfer that may be hours in. + // transfer that may be hours in — but a daemon that stays + // away is a stall like any other. s.logger.Debug("slskd transfer poll failed", "error", err) + if time.Since(lastProgress) >= s.stallAfter { + s.cancelTransfers(username, live) + + return fmt.Errorf( + "%w: slskd has not answered for %s: %w", + ErrSlskdTimeout, s.stallAfter, err, + ) + } + continue } - var ( - done, failed int - current int64 + tally := tallyTransfers( + transfers, wanted, stale, + time.Since(started) >= s.absentGrace, ) + live = tally.live - for _, t := range transfers { - if !wanted[t.Filename] { - continue - } - - current += t.BytesTransferred - - finished, ok := t.done() - if !finished { - continue - } - - if ok { - done++ - } else { - failed++ - } + if tally.bytes > lastBytes { + lastBytes = tally.bytes + lastProgress = time.Now() } if onProgress != nil { onProgress(Progress{ - Current: current, + Current: tally.bytes, Total: c.TotalSize, Phase: fmt.Sprintf( - "Transferring from %s (%d/%d)", username, done, len(wanted), + "Transferring from %s (%d/%d)", + username, tally.done, len(wanted), ), }) } - if done+failed < len(wanted) { + if tally.done+tally.failed >= len(wanted) { + // Some files failing is normal — a peer goes offline + // mid-folder. Let the importer's completeness check decide + // whether what arrived is enough, rather than discarding it + // here. + if tally.done == 0 { + return fmt.Errorf( + "%w: all %d files failed", + ErrSlskdTransferFailed, tally.failed, + ) + } + + return nil + } + + if time.Since(lastProgress) < s.stallAfter { continue } - // Some files failing is normal — a peer goes offline mid-folder. - // Let the importer's completeness check decide whether what - // arrived is enough, rather than discarding it here. - if done == 0 { - return fmt.Errorf( - "%w: all %d files failed", ErrSlskdTransferFailed, failed, + s.cancelTransfers(username, live) + + // A folder that stalls on its last track is the same shape as + // one whose last track failed, and goes forward the same way. + if tally.done > 0 { + s.logger.Info( + "slskd transfer stalled; keeping what arrived", + "peer", username, + "done", tally.done, + "wanted", len(wanted), ) + + return nil } - return nil + return fmt.Errorf( + "%w: %s sent nothing in %s", + ErrSlskdTimeout, username, s.stallAfter, + ) + } +} + +// transferTally is one poll's reading of the files a grab asked for. +type transferTally struct { + done, failed int + bytes int64 + + // live are the requested transfers slskd is still working on, + // which are what has to be cancelled if the grab is abandoned. + live []slskdTransfer +} + +// tallyTransfers reads a peer's transfer list against the files a grab +// asked for. +// +// A requested file slskd does not list at all is one it never accepted +// — refused at enqueue, or dropped — and it will never reach a terminal +// state to be counted by. Once absentExpired, such a file counts as +// failed, or the grab would wait on it until the six-hour ceiling. +func tallyTransfers( + transfers []slskdTransfer, + wanted map[string]bool, + stale map[string]bool, + absentExpired bool, +) transferTally { + seen := make(map[string]slskdTransfer, len(wanted)) + + for _, t := range transfers { + if !wanted[t.Filename] || stale[t.ID] { + continue + } + + seen[t.Filename] = t + } + + var out transferTally + + for name := range wanted { + t, ok := seen[name] + if !ok { + if absentExpired { + out.failed++ + } + + continue + } + + out.bytes += t.BytesTransferred + + finished, succeeded := t.done() + + switch { + case !finished: + out.live = append(out.live, t) + case succeeded: + out.done++ + default: + out.failed++ + } + } + + return out +} + +// cancelTransfers asks slskd to cancel and forget transfers this grab +// is abandoning. It runs on a context of its own: the usual reason to +// be here is that the caller's context has just been cancelled, and a +// cleanup that inherited it would never be sent. +func (s *slskd) cancelTransfers(username string, live []slskdTransfer) { + if len(live) == 0 { + return + } + + ctx, cancel := context.WithTimeout( + context.Background(), slskdCancelTimeout, + ) + defer cancel() + + for _, t := range live { + if t.ID == "" { + continue + } + + endpoint := slskdDownloadsPath(username) + "/" + + url.PathEscape(t.ID) + "?remove=true" + + if err := s.client.delete(ctx, endpoint); err != nil { + s.logger.Warn( + "could not cancel slskd transfer", + "peer", username, + "file", t.Filename, + "error", err, + ) + } } } @@ -581,9 +784,7 @@ func (s *slskd) transfersFor( } `json:"directories"` } - if err := s.client.get( - ctx, "/api/v0/transfers/downloads/"+username, &raw, - ); err != nil { + if err := s.client.get(ctx, slskdDownloadsPath(username), &raw); err != nil { return nil, err } diff --git a/backend/download/provider_slskd_test.go b/backend/download/provider_slskd_test.go index 992e6d5..de855a4 100644 --- a/backend/download/provider_slskd_test.go +++ b/backend/download/provider_slskd_test.go @@ -31,8 +31,18 @@ type slskdStub struct { transfers [][]slskdTransfer pollCount int + // before is what the downloads endpoint reports until something is + // enqueued: records slskd already held from earlier attempts. + before []slskdTransfer + // enqueued records what was requested for download. enqueued []map[string]any + posted bool + + // paths records the escaped path of every transfers call, and + // cancelled the escaped request URI of every DELETE. + paths []string + cancelled []string // unauthorized makes every call return 401. unauthorized bool @@ -87,7 +97,12 @@ func newSlskdStub(t *testing.T) *slskdStub { return } - if r.Method == http.MethodPost { + s.mu.Lock() + s.paths = append(s.paths, r.URL.EscapedPath()) + s.mu.Unlock() + + switch r.Method { + case http.MethodPost: var body []map[string]any if err := json.NewDecoder(r.Body).Decode(&body); err != nil { @@ -96,15 +111,35 @@ func newSlskdStub(t *testing.T) *slskdStub { s.mu.Lock() s.enqueued = body + s.posted = true s.mu.Unlock() w.WriteHeader(http.StatusCreated) + return + case http.MethodDelete: + s.mu.Lock() + s.cancelled = append(s.cancelled, r.URL.RequestURI()) + s.mu.Unlock() + + w.WriteHeader(http.StatusNoContent) + return } s.mu.Lock() + if !s.posted { + before := s.before + s.mu.Unlock() + + writeJSON(t, w, map[string]any{ + "directories": []map[string]any{{"files": before}}, + }) + + return + } + idx := s.pollCount if idx >= len(s.transfers) { idx = len(s.transfers) - 1 @@ -191,6 +226,11 @@ func newStubSlskd(t *testing.T, stub *slskdStub) (*slskd, string) { s.searchWait = 200 * time.Millisecond s.transferPoll = time.Millisecond + // Long enough that no existing test trips them by accident; the + // tests about stalls and absences set their own. + s.stallAfter = time.Minute + s.absentGrace = time.Minute + return s, downloads } @@ -565,3 +605,293 @@ 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 { + 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) + } + } + + return Candidate{ + Files: []CandidateFile{ + {Path: `\s\Album\01 A.flac`, Size: 500, IsAudio: true}, + {Path: `\s\Album\02 B.flac`, Size: 500, IsAudio: true}, + }, + TotalSize: 1000, + Payload: map[string]string{"username": peer}, + } +} + +// cancelledURIs returns what the stub was asked to cancel. +func (s *slskdStub) cancelledURIs() []string { + s.mu.Lock() + defer s.mu.Unlock() + + return append([]string(nil), s.cancelled...) +} + +// A peer that queues us and never sends a byte is given up on, and the +// queued transfers are cancelled in slskd rather than left to start +// hours later for a request nobody is waiting on. +func TestSlskdGrabGivesUpOnAStalledPeer(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.transfers = [][]slskdTransfer{{ + {ID: "t1", Filename: `\s\Album\01 A.flac`, State: "Queued, Remotely"}, + {ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"}, + }} + + s, downloads := newStubSlskd(t, stub) + s.stallAfter = 30 * time.Millisecond + + _, err := s.Grab( + context.Background(), slskdAlbum(t, downloads, "peer"), t.TempDir(), nil, + ) + if !errors.Is(err, ErrSlskdTimeout) { + t.Fatalf("error = %v, want ErrSlskdTimeout", err) + } + + got := stub.cancelledURIs() + if len(got) != 2 { + t.Fatalf("cancelled %v, want both queued transfers", got) + } + + for _, uri := range got { + if !strings.HasSuffix(uri, "?remove=true") { + t.Errorf("cancel %s does not remove the record", uri) + } + } +} + +// A folder that stalls on its last track goes forward with what +// arrived, the same as one whose last track failed; the importer's +// completeness check decides whether that is enough. +func TestSlskdGrabKeepsWhatArrivedBeforeAStall(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.transfers = [][]slskdTransfer{{ + { + ID: "t1", Filename: `\s\Album\01 A.flac`, + State: "Completed, Succeeded", BytesTransferred: 500, + }, + {ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"}, + }} + + s, downloads := newStubSlskd(t, stub) + s.stallAfter = 30 * time.Millisecond + + got, err := s.Grab( + context.Background(), + slskdAlbum(t, downloads, "peer", "01 A.flac"), + t.TempDir(), nil, + ) + if err != nil { + t.Fatalf("Grab: %v", err) + } + + if len(got.Files) != 1 { + t.Errorf("collected %d files, want the 1 that arrived", len(got.Files)) + } + + if cancelled := stub.cancelledURIs(); len(cancelled) != 1 || + !strings.Contains(cancelled[0], "/t2") { + t.Errorf("cancelled %v, want only the stalled t2", cancelled) + } +} + +// Progress is what holds the stall timer off. A transfer that keeps +// moving bytes is never abandoned, however long it takes. +func TestSlskdGrabWaitsOnATransferThatIsMoving(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + + for b := int64(1); b <= 100; b++ { + stub.transfers = append(stub.transfers, []slskdTransfer{ + {ID: "t1", Filename: `\s\Album\01 A.flac`, State: "InProgress", BytesTransferred: b}, + {ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"}, + }) + } + + stub.transfers = append(stub.transfers, []slskdTransfer{ + { + ID: "t1", + Filename: `\s\Album\01 A.flac`, + State: "Completed, Succeeded", + BytesTransferred: 500, + }, + { + ID: "t2", + Filename: `\s\Album\02 B.flac`, + State: "Completed, Succeeded", + BytesTransferred: 500, + }, + }) + + s, downloads := 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. + s.transferPoll = 5 * time.Millisecond + s.stallAfter = 150 * time.Millisecond + + got, err := s.Grab( + context.Background(), + slskdAlbum(t, downloads, "peer", "01 A.flac", "02 B.flac"), + t.TempDir(), nil, + ) + if err != nil { + t.Fatalf("Grab: %v", err) + } + + if len(got.Files) != 2 { + t.Errorf("collected %d files, want 2", len(got.Files)) + } +} + +// A file slskd never lists was refused at enqueue and will never reach +// a terminal state. It counts as failed once the grace period is up, +// rather than being waited on until the six-hour ceiling. +func TestSlskdGrabCountsAnUnlistedFileAsFailed(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.transfers = [][]slskdTransfer{{ + { + ID: "t1", Filename: `\s\Album\01 A.flac`, + State: "Completed, Succeeded", BytesTransferred: 500, + }, + }} + + s, downloads := newStubSlskd(t, stub) + s.absentGrace = 20 * time.Millisecond + + got, err := s.Grab( + context.Background(), + slskdAlbum(t, downloads, "peer", "01 A.flac"), + t.TempDir(), nil, + ) + if err != nil { + t.Fatalf("Grab: %v", err) + } + + if len(got.Files) != 1 { + t.Errorf("collected %d files, want 1", len(got.Files)) + } +} + +// A finished record left by an earlier attempt at the same file is not +// this attempt's answer. Without the snapshot it would fail the grab on +// the first poll, before the new transfer had started. +func TestSlskdGrabIgnoresAnEarlierAttemptsRecord(t *testing.T) { + t.Parallel() + + stale := slskdTransfer{ + ID: "old", Filename: `\s\Album\01 A.flac`, State: "Completed, Errored", + } + + stub := newSlskdStub(t) + stub.before = []slskdTransfer{stale} + stub.transfers = [][]slskdTransfer{ + {stale}, + { + stale, + { + ID: "new", Filename: `\s\Album\01 A.flac`, + State: "Completed, Succeeded", BytesTransferred: 500, + }, + }, + } + + s, downloads := newStubSlskd(t, stub) + + c := slskdAlbum(t, downloads, "peer", "01 A.flac") + c.Files = c.Files[:1] + + got, err := s.Grab(context.Background(), c, t.TempDir(), nil) + if err != nil { + t.Fatalf("Grab: %v", err) + } + + if len(got.Files) != 1 { + t.Errorf("collected %d files, want 1", len(got.Files)) + } +} + +// Cancelling the download cancels the transfer in slskd too. The +// cleanup must not inherit the cancelled context, or it is never sent. +func TestSlskdGrabCancelsTransfersWhenTheCallerGivesUp(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.transfers = [][]slskdTransfer{{ + {ID: "t1", Filename: `\s\Album\01 A.flac`, State: "InProgress", BytesTransferred: 10}, + {ID: "t2", Filename: `\s\Album\02 B.flac`, State: "Queued, Remotely"}, + }} + + s, downloads := 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) + if !errors.Is(err, ErrSlskdTimeout) { + t.Fatalf("error = %v, want ErrSlskdTimeout", err) + } + + if got := stub.cancelledURIs(); len(got) != 2 { + t.Errorf("cancelled %v, want both live transfers", got) + } +} + +// Soulseek usernames carry spaces and punctuation; spliced raw into the +// path, a name with a slash addresses a different endpoint entirely. +func TestSlskdEscapesTheUsername(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + stub.transfers = [][]slskdTransfer{{ + { + ID: "t1", Filename: `\s\Album\01 A.flac`, + State: "Completed, Succeeded", BytesTransferred: 500, + }, + { + ID: "t2", Filename: `\s\Album\02 B.flac`, + State: "Completed, Succeeded", BytesTransferred: 500, + }, + }} + + s, downloads := newStubSlskd(t, stub) + + if _, err := s.Grab( + context.Background(), + slskdAlbum(t, downloads, "dj a/b", "01 A.flac", "02 B.flac"), + t.TempDir(), nil, + ); err != nil { + t.Fatalf("Grab: %v", err) + } + + stub.mu.Lock() + paths := append([]string(nil), stub.paths...) + stub.mu.Unlock() + + for _, p := range paths { + if p != "/api/v0/transfers/downloads/dj%20a%2Fb" { + t.Errorf("transfers call went to %s", p) + } + } +}