From c88fc1c7c7f33a2c7297b62384831610d24b78cb Mon Sep 17 00:00:00 2001 From: Caleb Allen Date: Sat, 26 Sep 2026 22:07:18 -0400 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_017HJiuc3ZZhxsPXz3ozTirT --- backend/download/provider_slskd.go | 135 ++++++++++++++++++++++-- backend/download/provider_slskd_test.go | 28 ++++- backend/download/slskd_queue_test.go | 132 +++++++++++++++++++++++ 3 files changed, 287 insertions(+), 8 deletions(-) create mode 100644 backend/download/slskd_queue_test.go diff --git a/backend/download/provider_slskd.go b/backend/download/provider_slskd.go index 66f9ca2..a0fda64 100644 --- a/backend/download/provider_slskd.go +++ b/backend/download/provider_slskd.go @@ -95,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 @@ -104,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 @@ -168,6 +186,8 @@ 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). @@ -228,6 +248,8 @@ func newSlskd( transferPoll: slskdTransferPoll, stallAfter: slskdStallAfter, absentGrace: slskdAbsentGrace, + positionPoll: slskdPositionPoll, + queueCeiling: slskdQueueCeiling, }, nil } @@ -300,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". @@ -1028,11 +1056,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 @@ -1055,6 +1084,7 @@ func (s *slskd) awaitTransfers( lastProgress = started lastBytes int64 live []slskdTransfer + queue queueWatch ) for { @@ -1096,6 +1126,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, @@ -1148,6 +1200,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 diff --git a/backend/download/provider_slskd_test.go b/backend/download/provider_slskd_test.go index 5dd293d..a91ca1b 100644 --- a/backend/download/provider_slskd_test.go +++ b/backend/download/provider_slskd_test.go @@ -65,8 +65,13 @@ type slskdStub struct { // under downloads when it is enqueued, keyed by file base name. batches bool batchBodies []map[string]any - delivered map[string]string - downloads string + + // 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 { @@ -150,6 +155,23 @@ 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) @@ -379,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 } diff --git a/backend/download/slskd_queue_test.go b/backend/download/slskd_queue_test.go new file mode 100644 index 0000000..c5bfbae --- /dev/null +++ b/backend/download/slskd_queue_test.go @@ -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) + } +}