Files
yellowjacket/backend/download/provider_slskd.go
T
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

1577 lines
42 KiB
Go

package download
import (
"context"
"errors"
"fmt"
"log/slog"
"net/http"
"net/url"
"os"
"path"
"path/filepath"
"regexp"
"slices"
"strconv"
"strings"
"sync/atomic"
"time"
"github.com/google/uuid"
)
// Soulseek is reached through a user-run slskd daemon rather than the
// wire protocol. That trades a setup step for not having to implement
// peer connections, distributed search, and upload obligations — and
// keeps the user's Soulseek credentials in their daemon instead of in
// this process.
//
// One wrinkle shapes this adapter: slskd downloads into its own
// configured directory, not one we hand it. There is no API to stream
// a finished file back. So the user tells us where that directory is,
// and Grab waits for the transfer, then moves the files into staging.
// When slskd runs on another machine, that path has to be a mount —
// which is why Check verifies it exists rather than discovering the
// problem after a two-hour transfer.
// slskd provider errors.
var (
// ErrSlskdUnreachable means the daemon did not answer.
ErrSlskdUnreachable = errors.New("slskd is unreachable")
// ErrSlskdAuth means the API key was rejected.
ErrSlskdAuth = errors.New("slskd rejected the API key")
// ErrSlskdDownloadsPath means the configured downloads directory is
// missing or unreadable from this machine.
ErrSlskdDownloadsPath = errors.New(
"slskd downloads directory is not readable from here",
)
// ErrSlskdTransferFailed means a peer transfer ended badly.
ErrSlskdTransferFailed = errors.New("slskd transfer failed")
// ErrSlskdTimeout means a search or transfer outlived its budget.
ErrSlskdTimeout = errors.New("slskd timed out")
)
// slskd tuning.
const (
// slskdSearchPoll is how often an in-flight search is polled.
slskdSearchPoll = 1 * time.Second
// slskdSearchWait bounds a single search. Soulseek searches return
// results progressively; waiting the full budget gets noticeably
// more peers than bailing at the first response. 12s was measured
// to miss real, available peers on real-world queries (roughly 4 of
// 5 attempts for a live search came back empty before this many
// responses had a chance to arrive), so this is generous rather
// than tight. Kept a few seconds under Manager's per-provider
// searchTimeout (25s) so the request/cleanup round-trips around it
// 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
// slskdMinFiles is the fewest audio files a folder needs before it
// is offered as a candidate for an album. Soulseek returns a lot of
// one-file noise for common queries. A single-track request takes
// one (see minFilesFor).
slskdMinFiles = 2
// slskdHTTPTimeout bounds one API call.
slskdHTTPTimeout = 20 * time.Second
// millisPerSecond converts slskd's whole-second file lengths.
millisPerSecond = 1000
// 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 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
// 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
// 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
)
func init() {
Register(
Descriptor{
Kind: KindSlskd,
Name: "Soulseek (slskd)",
Summary: "Search and download from the Soulseek network " +
"through your own slskd daemon.",
RequiresExternal: "slskd",
Caps: Caps{
CanSearch: true,
CanTransport: true,
CanCancel: true,
ReportsSize: true,
},
Fields: []Field{
{
Key: "url",
Label: "slskd URL",
Placeholder: "http://localhost:5030",
Required: true,
Default: "http://localhost:5030",
},
{
Key: "apiKey",
Label: "API key",
Secret: true,
Required: true,
Help: "From your slskd configuration under web.authentication.",
},
{
Key: "downloadsPath",
Label: "slskd downloads folder",
Placeholder: "/var/lib/slskd/downloads",
Path: true,
Required: true,
Help: "The folder slskd saves to, as this machine sees it. " +
"If slskd runs elsewhere, this must be a mounted share.",
},
},
},
newSlskd,
)
}
// slskd is the Soulseek provider.
type slskd struct {
info ProviderInfo
logger *slog.Logger
client *apiClient
downloadsPath string
// Poll intervals are fields rather than constants so tests can run
// the full search-and-transfer flow without sleeping through it.
searchPoll time.Duration
searchWait time.Duration
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.
func newSlskd(
cfg Config,
secrets SecretLookup,
logger *slog.Logger,
) (Provider, error) {
base := strings.TrimRight(cfg.Setting("url", ""), "/")
if base == "" {
return nil, fmt.Errorf("%w: slskd URL is required", ErrNotConfigured)
}
downloads := cfg.Setting("downloadsPath", "")
if downloads == "" {
return nil, fmt.Errorf(
"%w: slskd downloads folder is required", ErrNotConfigured,
)
}
apiKey := ""
if secrets != nil {
key, err := secrets("apiKey")
if err != nil {
return nil, fmt.Errorf("%w: no API key stored", ErrNotConfigured)
}
apiKey = key
}
return &slskd{
info: ProviderInfo{
ID: cfg.ID,
Kind: KindSlskd,
Name: cfg.Name,
Enabled: cfg.Enabled,
Priority: cfg.Priority,
Caps: Caps{
CanSearch: true,
CanTransport: true,
CanCancel: true,
ReportsSize: true,
},
},
logger: logger.With("provider", "slskd"),
client: newAPIClient(
base, "X-Api-Key", apiKey, slskdHTTPTimeout,
ErrSlskdUnreachable, ErrSlskdAuth,
),
downloadsPath: downloads,
searchPoll: slskdSearchPoll,
searchWait: slskdSearchWait,
transferPoll: slskdTransferPoll,
stallAfter: slskdStallAfter,
absentGrace: slskdAbsentGrace,
positionPoll: slskdPositionPoll,
queueCeiling: slskdQueueCeiling,
}, nil
}
// Info returns the provider's identity.
func (s *slskd) Info() ProviderInfo {
return s.info
}
// Close is a no-op; the HTTP client holds no session.
func (s *slskd) Close() error {
return nil
}
// Check verifies the daemon answers, the key is accepted, and the
// downloads directory is readable from this machine.
func (s *slskd) Check(ctx context.Context) error {
var app map[string]any
if err := s.client.get(ctx, "/api/v0/application", &app); err != nil {
return err
}
info, err := os.Stat(s.downloadsPath)
if err != nil || !info.IsDir() {
return fmt.Errorf("%w: %s", ErrSlskdDownloadsPath, s.downloadsPath)
}
return nil
}
// ---------------------------------------------------------------------------
// API types
// ---------------------------------------------------------------------------
// slskdSearch is a search as slskd reports it.
type slskdSearch struct {
ID string `json:"id"`
IsComplete bool `json:"isComplete"`
Responses []slskdResponse `json:"responses"`
}
// slskdResponse is one peer's answer to a search.
type slskdResponse struct {
Username string `json:"username"`
HasFreeUploadSlot bool `json:"hasFreeUploadSlot"`
QueueLength int `json:"queueLength"`
UploadSpeed int64 `json:"uploadSpeed"`
Files []slskdFile `json:"files"`
LockedFileCount int `json:"lockedFileCount"`
FileCount int `json:"fileCount"`
}
// slskdFile is one file a peer is offering.
type slskdFile struct {
Filename string `json:"filename"`
Size int64 `json:"size"`
BitRate int `json:"bitRate"`
// Length is the duration in whole seconds.
Length int `json:"length"`
}
// slskdTransfer is one download's state.
type slskdTransfer struct {
ID string `json:"id"`
Username string `json:"username"`
Filename string `json:"filename"`
State string `json:"state"`
Size int64 `json:"size"`
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".
func (t slskdTransfer) done() (finished, ok bool) {
if !strings.Contains(t.State, "Completed") {
return false, false
}
return true, strings.Contains(t.State, "Succeeded")
}
// ---------------------------------------------------------------------------
// Search
// ---------------------------------------------------------------------------
// Search runs a Soulseek search and groups the results into per-peer,
// per-folder candidates. A folder from one peer is the unit a user
// actually wants: Soulseek has no album concept, but people organise
// their shares by album directory.
//
// Up to two queries run at once — the request as written and a
// normalised form of it (see slskdQueries) — and their candidates are
// merged. They run concurrently rather than as a fallback because the
// manager gives a provider one search budget, and a Soulseek search
// spends most of it waiting for peers to answer; a second query after
// the first would not fit.
func (s *slskd) Search(ctx context.Context, dl Download) ([]Candidate, error) {
queries := slskdQueries(dl)
if len(queries) == 0 {
return nil, nil
}
type found struct {
candidates []Candidate
err error
}
results := make(chan found, len(queries))
for _, q := range queries {
go func(q string) {
c, err := s.searchOnce(ctx, q, minFilesFor(dl))
results <- found{candidates: c, err: err}
}(q)
}
var (
out []Candidate
seen = map[string]bool{}
firstErr error
answered int
)
for range queries {
r := <-results
if r.err != nil {
s.logger.Debug("slskd search failed", "error", r.err)
if firstErr == nil {
firstErr = r.err
}
continue
}
answered++
// The same peer's folder turns up under both queries; the ID is
// peer and folder, so it is the same candidate.
for _, c := range r.candidates {
if seen[c.ID] {
continue
}
seen[c.ID] = true
out = append(out, c)
}
}
if answered == 0 {
return nil, firstErr
}
return out, nil
}
// searchOnce runs one query to completion and returns its candidates.
func (s *slskd) searchOnce(
ctx context.Context,
text string,
minFiles int,
) ([]Candidate, error) {
// slskd's search endpoint deserializes id as a .NET Guid server-side,
// so it must be a dashed UUID — the app's own newID() (a plain hex
// string, used for request/item IDs elsewhere) is rejected with an
// HTTP 400 before any search happens.
searchID := uuid.NewString()
if err := s.client.post(
ctx, "/api/v0/searches", s.searchRequest(searchID, text, minFiles), nil,
); err != nil {
return nil, err
}
// Best effort cleanup; a left-behind search is harmless but clutters
// the slskd UI.
defer func() {
_ = s.client.delete(
context.WithoutCancel(ctx), "/api/v0/searches/"+searchID,
)
}()
if err := s.awaitSearch(ctx, searchID); err != nil {
return nil, err
}
responses, err := s.searchResponses(ctx, searchID)
if err != nil {
return nil, err
}
return s.candidatesFrom(responses, minFiles), nil
}
// searchRequest is the body that starts a search.
//
// Every option is stated rather than left to the daemon, because
// slskd's defaults are its own and not ours. Its search timeout in
// particular has to finish inside our wait: a search that slskd is still
// running when we stop polling is results we asked for and discarded.
// The response and file limits are raised well above what a popular
// album produces, and the peer filters let slskd drop answers this
// provider would only score down to nothing — a folder too small to be
// a candidate, a peer with a queue it will not reach today.
func (s *slskd) searchRequest(id, text string, minFiles int) map[string]any {
const (
responseLimit = 500
fileLimit = 20_000
maximumPeerQueueLength = 100
)
// 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,
"responseLimit": responseLimit,
"fileLimit": fileLimit,
"filterResponses": true,
"minimumResponseFileCount": minFiles,
"maximumPeerQueueLength": maximumPeerQueueLength,
}
}
// slskdQueries is what is searched for a request: the request's own
// search text, and a normalised form of it when that differs.
//
// Soulseek matches every term against the file's full path, so each
// extra word is a filter, and some words filter wrongly:
//
// - edition qualifiers — "(Deluxe Edition)", "[2011 Remaster]" — are
// in the catalog's title and rarely in anyone's folder name;
// - punctuation splits a term oddly, and a term that starts with "-"
// is an *exclusion*, so an album called "-ism" searches for
// everything without it;
// - "Various Artists" is in no one's path for a compilation.
//
// A query the user typed is theirs and is searched exactly as written.
func slskdQueries(dl Download) []string {
primary := strings.TrimSpace(dl.SearchText())
if primary == "" {
return nil
}
out := []string{primary}
if dl.Query != "" {
return out
}
artist := dl.Artist
if isVariousArtists(artist) {
artist = ""
}
normal := Download{
Artist: normalizeSearchTerms(artist),
Album: normalizeSearchTerms(editionPattern.ReplaceAllString(dl.Album, " ")),
}
if alt := strings.TrimSpace(normal.SearchText()); alt != "" &&
!strings.EqualFold(alt, primary) {
out = append(out, alt)
}
return out
}
var (
// editionPattern finds an edition qualifier: a bracketed group that
// names an edition, or a trailing " - 2011 Remaster".
editionPattern = regexp.MustCompile(
`(?i)\s*[(\[][^)\]]*\b(?:deluxe|edition|remaster(?:ed)?|expanded|` +
`anniversary|bonus|explicit|reissue|special|collector'?s?|` +
`version|mono|stereo)\b[^)\]]*[)\]]` +
`|\s+-\s+(?:\d{4}\s+)?remaster(?:ed)?\b.*$`,
)
// nonWordPattern is everything that is not a letter or a digit.
nonWordPattern = regexp.MustCompile(`[^\p{L}\p{N}]+`)
)
// normalizeSearchTerms reduces text to plain words.
func normalizeSearchTerms(s string) string {
return strings.Join(strings.Fields(nonWordPattern.ReplaceAllString(s, " ")), " ")
}
// isVariousArtists reports whether an artist credit is a compilation's
// placeholder rather than an artist.
func isVariousArtists(artist string) bool {
switch strings.ToLower(strings.TrimSpace(artist)) {
case "various artists", "various", "va":
return true
default:
return false
}
}
// minFilesFor is the fewest audio files a folder must offer to be a
// candidate for this request.
//
// Soulseek answers a search with the files that match it, not with the
// folders they sit in. An album query matches every file in the album's
// folder, because the folder name carries the terms; a *track* query
// 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.
func minFilesFor(dl Download) int {
if dl.RecordingMBID != "" {
return 1
}
return slskdMinFiles
}
// awaitSearch polls until the search completes or the budget runs out.
// A timeout is not an error: partial Soulseek results are normal and
// often good enough.
//
// The poll asks for the search's state only. It used to ask for every
// response on every one-second tick, which for a popular album is the
// same few thousand file entries serialised twenty times to be read
// once; searchResponses fetches them once at the end.
func (s *slskd) awaitSearch(ctx context.Context, searchID string) error {
deadline := time.Now().Add(s.searchWait)
for time.Now().Before(deadline) {
select {
case <-ctx.Done():
return fmt.Errorf("%w: search cancelled", ErrSlskdTimeout)
case <-time.After(s.searchPoll):
}
var search slskdSearch
if err := s.client.get(
ctx, "/api/v0/searches/"+searchID, &search,
); err != nil {
return err
}
if search.IsComplete {
return nil
}
}
return nil
}
// searchResponses fetches a search's responses once.
//
// `/searches/{id}/responses` is the endpoint for that; a daemon that
// does not answer it is asked the older way, with the search itself
// carrying its responses, so an older slskd degrades to the previous
// behaviour rather than to no results at all.
func (s *slskd) searchResponses(
ctx context.Context,
searchID string,
) ([]slskdResponse, error) {
var responses []slskdResponse
err := s.client.get(
ctx, "/api/v0/searches/"+searchID+"/responses", &responses,
)
if err == nil {
return responses, nil
}
s.logger.Debug(
"slskd responses endpoint failed; asking with the search",
"error", err,
)
var search slskdSearch
if err := s.client.get(
ctx,
"/api/v0/searches/"+searchID+"?includeResponses=true",
&search,
); err != nil {
return nil, err
}
return search.Responses, nil
}
// candidatesFrom groups a search's responses into candidates, dropping
// folders with fewer than minFiles audio files.
func (s *slskd) candidatesFrom(
responses []slskdResponse,
minFiles int,
) []Candidate {
out := make([]Candidate, 0, len(responses))
for _, resp := range responses {
for folder, files := range groupByFolder(resp.Files) {
audio := 0
cfiles := make([]CandidateFile, 0, len(files))
var total int64
for _, f := range files {
format, isAudio := FormatForPath(f.Filename)
if isAudio {
audio++
}
cfiles = append(cfiles, CandidateFile{
Path: f.Filename,
Size: f.Size,
Format: format,
Bitrate: f.BitRate,
IsAudio: isAudio,
LengthMillis: int64(f.Length) * millisPerSecond,
})
total += f.Size
}
if audio < minFiles {
continue
}
out = append(out, Candidate{
ID: "slskd:" + resp.Username + ":" + folder,
Kind: KindSlskd,
Protocol: ProtocolDirect,
Title: path.Base(strings.ReplaceAll(folder, `\`, "/")),
Origin: resp.Username,
Files: cfiles,
TotalSize: total,
Health: peerHealth(resp),
Payload: map[string]string{"username": resp.Username},
})
}
}
return out
}
// groupByFolder buckets a peer's files by the album directory they sit
// in — the containing directory, or the one above it for a disc folder
// (see AlbumDir), so a multi-disc rip is one candidate and not two.
func groupByFolder(files []slskdFile) map[string][]slskdFile {
out := map[string][]slskdFile{}
for _, f := range files {
dir := AlbumDir(f.Filename)
out[dir] = append(out[dir], f)
}
return out
}
// peerHealth scores how likely a peer is to actually deliver, in 0..1.
// On Soulseek this matters more than it does for torrents: a queue of
// 40 behind a single upload slot means the transfer starts tomorrow,
// and that is the difference between a good candidate and a bad one no
// matter how good the files look.
func peerHealth(r slskdResponse) float64 {
score := 0.35
if r.HasFreeUploadSlot {
score += 0.4
}
switch {
case r.QueueLength == 0:
score += 0.15
case r.QueueLength <= 3:
score += 0.08
case r.QueueLength > 20:
score -= 0.2
}
// Anything above roughly 1 MB/s is fast enough that more speed does
// not change the experience.
const fastEnough = 1_000_000
if r.UploadSpeed > 0 {
ratio := float64(r.UploadSpeed) / fastEnough
if ratio > 1 {
ratio = 1
}
score += 0.1 * ratio
}
return clamp01(score)
}
// ---------------------------------------------------------------------------
// Transfer
// ---------------------------------------------------------------------------
// Grab enqueues a candidate's files with slskd, waits for the peer to
// send them, then moves them out of slskd's download directory into the
// staging directory.
func (s *slskd) Grab(
ctx context.Context,
c Candidate,
dst string,
onProgress ProgressFunc,
) (Result, error) {
username := c.Payload["username"]
if username == "" {
return Result{}, fmt.Errorf(
"%w: candidate has no peer username", ErrSlskdTransferFailed,
)
}
// 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. 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
}
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 {
wanted = append(wanted, map[string]any{
"filename": f.Path,
"size": f.Size,
})
}
if err := s.client.post(
ctx, slskdDownloadsPath(username), wanted, nil,
); err != nil {
return "", 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,
})
}
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.
//
// slskd names a download's directory after the remote *leaf* folder, so
// two different albums both shared as "Greatest Hits" — or any two
// multi-disc rips, whose leaves are "CD1" and "CD2" — are written into
// one directory, and collect finds files by name there. Run at once,
// a file one peer never sent is filled by the other peer's file of the
// same name. One grab per peer made that impossible; several peers at
// once makes it likely. It is package-level and keyed on the full
// path because two configured clients can share one daemon.
var slskdFolders keyedLock[string]
// localFolders returns the directories under downloadsPath a candidate's
// files will be written to, sorted so every grab takes them in the same
// order and two cannot each hold what the other waits for.
func (s *slskd) localFolders(c Candidate) []string {
var out []string
for _, f := range c.Files {
norm := strings.ReplaceAll(f.Path, `\`, "/")
out = append(out, filepath.Join(s.downloadsPath, path.Base(path.Dir(norm))))
}
slices.Sort(out)
return slices.Compact(out)
}
// lockSlskdFolders takes every folder in order, releasing what it holds
// if the context ends part way.
func lockSlskdFolders(ctx context.Context, folders []string) (func(), error) {
releases := make([]func(), 0, len(folders))
releaseAll := func() {
for _, r := range slices.Backward(releases) {
r()
}
}
for _, f := range folders {
r, err := slskdFolders.acquire(ctx, f)
if err != nil {
releaseAll()
return nil, err
}
releases = append(releases, r)
}
return releaseAll, nil
}
// 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, 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*. 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
// 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 {
wanted := make(map[string]bool, len(c.Files))
for _, f := range c.Files {
wanted[f.Path] = true
}
var (
started = time.Now()
lastProgress = started
lastBytes int64
live []slskdTransfer
queue queueWatch
)
for {
select {
case <-ctx.Done():
s.cancelTransfers(username, live)
return fmt.Errorf("%w: transfer cancelled", ErrSlskdTimeout)
case <-time.After(s.transferPoll):
}
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 — 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
}
tally := tallyTransfers(
transfers, wanted, stale,
time.Since(started) >= s.absentGrace,
)
live = tally.live
if tally.bytes > lastBytes {
lastBytes = tally.bytes
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,
Total: c.TotalSize,
Phase: fmt.Sprintf(
"Transferring from %s (%d/%d)",
username, tally.done, 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
}
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 fmt.Errorf(
"%w: %s sent nothing in %s",
ErrSlskdTimeout, username, s.stallAfter,
)
}
}
// 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
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,
)
}
}
}
// transfersFor returns a peer's current downloads. slskd nests
// transfers under directories, so this flattens them.
func (s *slskd) transfersFor(
ctx context.Context,
username string,
) ([]slskdTransfer, error) {
var raw struct {
Directories []struct {
Files []slskdTransfer `json:"files"`
} `json:"directories"`
}
if err := s.client.get(ctx, slskdDownloadsPath(username), &raw); err != nil {
return nil, err
}
out := make([]slskdTransfer, 0, len(raw.Directories))
for _, d := range raw.Directories {
out = append(out, d.Files...)
}
return out, nil
}
// collect moves finished files out of slskd's download directory into
// 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 {
norm := strings.ReplaceAll(f.Path, `\`, "/")
folder := path.Base(path.Dir(norm))
base := path.Base(norm)
var (
src string
info os.FileInfo
ok bool
)
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
}
// A multi-disc candidate keeps its disc folders in staging.
// 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 _, disc := discFolder(folder); disc {
target = filepath.Join(dst, folder, base)
}
if err := movePath(src, target); err != nil {
return Result{}, fmt.Errorf("collect %s: %w", base, err)
}
result.Files = append(result.Files, target)
result.BytesTransferred += info.Size()
}
if len(result.Files) == 0 {
return Result{}, fmt.Errorf(
"%w: nothing found under %s",
ErrSlskdDownloadsPath, s.downloadsPath,
)
}
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
}