fix(worker): smarter Soulseek peer selection + stall/progress-aware transfers

Draining the queue via Soulseek was stalling: slow peers (~100KB/s) never finished a
full FLAC album within the fixed 10-min timeout, got cancelled at ~89%, and the
fall-through restarted from scratch on the next peer — while single-song folders named
like the album were also being tried.

- Stall detection: abandon a peer with no byte progress for ~90s (queued/dead) instead
  of waiting the full timeout; a peer that IS progressing keeps its slot up to a 60-min
  backstop, so a slow-but-working transfer can finish.
- Rank search candidates by peer: free upload slot, then speed, then shortest queue.
- Drop candidates with far fewer tracks than the release (single-song folders) up front,
  so no download attempt is wasted on them.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

worker/lyra_worker/adapters/_slskd.py

@@ -11,7 +11,8 @@ _DEFAULT_DOWNLOADS_ROOT = "/slskd-downloads"
 _AUDIO_EXT = {"flac", "mp3", "m4a", "ogg", "opus", "wav", "aac"}
 _LOSSLESS_EXT = {"flac", "wav"}
 _SEARCH_POLLS = 40    # * 3s ≈ 2 min
-_XFER_POLLS = 200    # * 3s ≈ 10 min
+_MAX_XFER_POLLS = 1200  # * 3s ≈ 60 min absolute backstop for a slow-but-progressing peer
+_STALL_POLLS = 30     # * 3s ≈ 90s of no byte progress → abandon a queued/dead peer and fall through
 _POLL_SECONDS = 3
 
 
@@ -27,6 +28,52 @@ def _dirname(path: str) -> str:
     return ""
 
 
+def _parse_search_responses(responses: list) -> list[dict]:
+    """Turn slskd search responses into album candidates (one per peer+directory), ordered so the
+    peers most likely to deliver quickly come first: a free upload slot, then higher upload speed,
+    then a shorter queue. The pipeline tries candidates in this order, so ranking fast/free peers
+    first (over slow or queued ones) is what makes the Soulseek fall-through actually converge.
+    Pure — no I/O — so it is unit-tested offline."""
+    scored: list[tuple] = []
+    for resp in responses:
+        username = resp.get("username", "")
+        has_slot = 1 if resp.get("hasFreeUploadSlot") else 0
+        speed = int(resp.get("uploadSpeed") or 0)
+        queue = int(resp.get("queueLength") or 0)
+        by_dir: dict[str, list] = {}
+        for f in resp.get("files", []) or []:
+            name = f.get("filename", "")
+            ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
+            if ext not in _AUDIO_EXT:
+                continue
+            by_dir.setdefault(_dirname(name), []).append(
+                {"filename": name, "size": f.get("size", 0), "ext": ext}
+            )
+        for directory, dfiles in by_dir.items():
+            if not dfiles:
+                continue
+            lossless = all(f["ext"] in _LOSSLESS_EXT for f in dfiles)
+            dirname = _basename(directory)
+            if " - " in dirname:
+                guess_artist, guess_album = (p.strip() for p in dirname.split(" - ", 1))
+            else:
+                guess_artist, guess_album = "", dirname
+            candidate = {
+                "source_ref": json.dumps({
+                    "username": username,
+                    "files": [{"filename": f["filename"], "size": f["size"]} for f in dfiles],
+                }),
+                "title": guess_album,
+                "artist": guess_artist,
+                "track_count": len(dfiles),
+                "format": "FLAC" if lossless else "MP3",
+                "bitrate": None,
+            }
+            scored.append((has_slot, speed, -queue, candidate))
+    scored.sort(key=lambda t: (t[0], t[1], t[2]), reverse=True)  # free + fast + short-queue first
+    return [c for *_rest, c in scored]
+
+
 class SlskdClient:
     """Real Soulseek client talking to an slskd daemon's REST API.
 
@@ -75,43 +122,7 @@ class SlskdClient:
                 break
 
         responses = self._get(f"/api/v0/searches/{search_id}/responses")
-        candidates: list[dict] = []
-        for resp in responses:
-            username = resp.get("username", "")
-            by_dir: dict[str, list] = {}
-            for f in resp.get("files", []) or []:
-                name = f.get("filename", "")
-                ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
-                if ext not in _AUDIO_EXT:
-                    continue
-                by_dir.setdefault(_dirname(name), []).append(
-                    {"filename": name, "size": f.get("size", 0), "ext": ext}
-                )
-            for directory, dfiles in by_dir.items():
-                if not dfiles:
-                    continue
-                lossless = all(f["ext"] in _LOSSLESS_EXT for f in dfiles)
-                dirname = _basename(directory)
-                if " - " in dirname:
-                    guess_artist, guess_album = (p.strip() for p in dirname.split(" - ", 1))
-                else:
-                    guess_artist, guess_album = "", dirname
-                candidates.append(
-                    {
-                        "source_ref": json.dumps(
-                            {
-                                "username": username,
-                                "files": [{"filename": f["filename"], "size": f["size"]} for f in dfiles],
-                            }
-                        ),
-                        "title": guess_album,
-                        "artist": guess_artist,
-                        "track_count": len(dfiles),
-                        "format": "FLAC" if lossless else "MP3",
-                        "bitrate": None,
-                    }
-                )
-        return candidates
+        return _parse_search_responses(responses if isinstance(responses, list) else [])
 
     def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
         ref = json.loads(source_ref)
@@ -141,14 +152,18 @@ class SlskdClient:
 
         total = len(files)
         completed = False
-        for _ in range(_XFER_POLLS):
+        last_bytes = -1
+        stall = 0
+        for _ in range(_MAX_XFER_POLLS):
             time.sleep(_POLL_SECONDS)
             data = self._get(f"/api/v0/transfers/downloads/{username}")
             states: list[str] = []
+            bytes_now = 0
             for directory in data.get("directories", []) if isinstance(data, dict) else []:
                 for f in directory.get("files", []):
                     if f.get("filename") in wanted:  # only the files we enqueued
                         states.append(str(f.get("state", "")))
+                        bytes_now += int(f.get("bytesTransferred") or 0)
                         if f.get("id"):  # remember the transfer so we can cancel it if abandoned
                             xfer_ids.add(str(f["id"]))
             if states:
@@ -159,6 +174,16 @@ class SlskdClient:
                 if done >= total:
                     completed = True
                     break
+                # Stall detection: a peer making byte progress keeps its slot (up to the absolute
+                # backstop above) so a slow-but-working transfer can finish; one that sends no bytes
+                # for _STALL_POLLS (~90s) is queued/dead — abandon it fast and fall through.
+                if bytes_now > last_bytes:
+                    last_bytes = bytes_now
+                    stall = 0
+                else:
+                    stall += 1
+                    if stall >= _STALL_POLLS:
+                        raise TimeoutError(f"soulseek transfer stalled for {username}")
 
         if not completed:
             raise TimeoutError(f"soulseek download did not complete for {username}")

worker/lyra_worker/pipeline.py

@@ -395,6 +395,12 @@ def run_pipeline(
     for adapter, results in per_adapter:
         for c in results:
             found.append(replace(c, source_tier=adapter.tier, confidence=score_confidence(target, c)))
+    # Drop candidates that can't be the full album — a result with far fewer tracks than the release
+    # has (e.g. a single-song Soulseek folder named like the album). Same threshold as the
+    # completeness 'too small' guard, applied up front so we never waste a download attempt on it.
+    # Unknown counts (0) and legitimately smaller editions (e.g. 11 vs 12) are kept.
+    if target.track_count:
+        found = [c for c in found if c.track_count == 0 or c.track_count * 2 > target.track_count]
     _persist_candidates(conn, job_id, found)
 
     # Qobuz pacing: when the budget gate is closed (off-hours / daily cap / spacing), drop Qobuz

worker/tests/test_candidate_filter.py

@@ -0,0 +1,57 @@
+"""A search result with far fewer tracks than the release (e.g. a single-song Soulseek folder
+named like the album) can't be the full album — it must be dropped before we waste a download."""
+import os
+
+from lyra_worker.claim import claim_next
+from lyra_worker.pipeline import run_pipeline
+from lyra_worker.types import Candidate, DownloadResult, MBTarget, Quality
+from tests.conftest import insert_request
+
+_Q = Quality(fmt="FLAC", lossless=True, bit_depth=24, sample_rate=96000)
+
+
+class _TwoCandQobuz:
+    name = "qobuz"
+    tier = 0
+
+    def __init__(self):
+        self.downloaded = []
+
+    def health(self):
+        return True
+
+    def search(self, target):
+        return [
+            Candidate(source="qobuz", source_ref="single", matched_artist=target.artist,
+                      matched_album=target.album, quality=_Q, track_count=1, source_tier=0),
+            Candidate(source="qobuz", source_ref="full", matched_artist=target.artist,
+                      matched_album=target.album, quality=_Q, track_count=12, source_tier=0),
+        ]
+
+    def download(self, candidate, dest, on_progress):
+        self.downloaded.append(candidate.source_ref)
+        os.makedirs(dest, exist_ok=True)
+        for i in range(candidate.track_count):
+            with open(os.path.join(dest, f"{i + 1:02d}.flac"), "wb") as fh:
+                fh.write(b"a")
+        return DownloadResult(ok=True, path=dest, track_count=candidate.track_count)
+
+
+class _Resolver:
+    def __init__(self, n):
+        self.n = n
+
+    def resolve(self, artist, album):
+        return MBTarget(artist=artist, album=album, track_count=self.n)
+
+
+def test_single_song_candidate_is_dropped(conn):
+    job_id = insert_request(conn, artist="A", album="B")
+    claim_next(conn)
+    adapter = _TwoCandQobuz()
+    run_pipeline(conn, job_id, [adapter], resolver=_Resolver(12), dest_root="/tmp/lib-cf")
+
+    assert adapter.downloaded == ["full"]  # only the full album was ever attempted
+    with conn.cursor() as cur:
+        cur.execute('SELECT "trackCount" FROM "Candidate" WHERE "jobId" = %s', (job_id,))
+        assert [r[0] for r in cur.fetchall()] == [12]  # the single-song folder wasn't even persisted

worker/tests/test_slskd_cancel.py

@@ -1,12 +1,12 @@
-"""When an slskd download is abandoned (times out or errors), the enqueued transfers must be
-cancelled — otherwise, as the pipeline falls through to other peers for the same album, abandoned
-transfers pile up in slskd (many downloads of one album from different sources)."""
+"""slskd download behaviour: abandoned transfers are cancelled (no pile-up), a stalled peer is
+dropped fast, a slow-but-progressing peer is kept, and search candidates are ranked fast/free
+peers first."""
 import json
 
 import pytest
 
 import lyra_worker.adapters._slskd as slskd_mod
-from lyra_worker.adapters._slskd import SlskdClient
+from lyra_worker.adapters._slskd import SlskdClient, _parse_search_responses
 
 
 class _Resp:
@@ -20,53 +20,91 @@ class _Resp:
         return self._p
 
 
-class _FakeRequests:
-    """Enqueue OK; polling always reports the file still in progress (never completes); records
-    DELETEs so the test can assert the abandoned transfer was cancelled."""
-    def __init__(self, state="InProgress"):
+class _SeqRequests:
+    """post OK; get() returns successive poll payloads (last one repeats); records DELETEs."""
+    def __init__(self, polls):
+        self._polls = list(polls)
+        self._i = 0
         self.deleted = []
-        self._poll = {"directories": [{"files": [
-            {"filename": "f1.flac", "size": 5, "state": state, "id": "t1"},
-        ]}]}
 
     def post(self, url, **kw):
         return _Resp({})
 
     def get(self, url, **kw):
-        return _Resp(self._poll)
+        p = self._polls[min(self._i, len(self._polls) - 1)]
+        self._i += 1
+        return _Resp(p)
 
     def delete(self, url, **kw):
         self.deleted.append(url)
         return _Resp({})
 
 
+def _poll(state, bytes_transferred, fid="t1"):
+    return {"directories": [{"files": [
+        {"filename": "f1.flac", "size": 100, "state": state,
+         "bytesTransferred": bytes_transferred, "id": fid},
+    ]}]}
+
+
 def _client(tmp_path):
     return SlskdClient({"slskd.url": "http://x", "slskd.api_key": "k",
                         "slskd.downloadsDir": str(tmp_path)})
 
 
-_REF = json.dumps({"username": "peerA", "files": [{"filename": "f1.flac", "size": 5}]})
+_REF = json.dumps({"username": "peerA", "files": [{"filename": "f1.flac", "size": 100}]})
 
 
-def test_cancels_transfer_on_timeout(tmp_path, monkeypatch):
-    fake = _FakeRequests(state="InProgress")
-    monkeypatch.setattr(slskd_mod, "requests", fake)
+def test_stalled_peer_is_abandoned_and_cancelled(tmp_path, monkeypatch):
+    # No byte progress across polls → stall → TimeoutError → the transfer is cancelled.
+    monkeypatch.setattr(slskd_mod, "requests", _SeqRequests([_poll("InProgress", 0)]))
     monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
-    monkeypatch.setattr(slskd_mod, "_XFER_POLLS", 2)
+    monkeypatch.setattr(slskd_mod, "_STALL_POLLS", 3)
+    fake = slskd_mod.requests
 
     with pytest.raises(TimeoutError):
         _client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda _p: None)
-
-    assert any(u.endswith("/api/v0/transfers/downloads/peerA/t1") for u in fake.deleted), fake.deleted
+    assert any(u.endswith("/api/v0/transfers/downloads/peerA/t1") for u in fake.deleted)
 
 
-def test_cancels_transfer_on_error_state(tmp_path, monkeypatch):
-    fake = _FakeRequests(state="Completed, Errored")
-    monkeypatch.setattr(slskd_mod, "requests", fake)
+def test_error_state_is_cancelled(tmp_path, monkeypatch):
+    monkeypatch.setattr(slskd_mod, "requests", _SeqRequests([_poll("Completed, Errored", 0)]))
     monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
-    monkeypatch.setattr(slskd_mod, "_XFER_POLLS", 5)
-
+    fake = slskd_mod.requests
     with pytest.raises(RuntimeError):
         _client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda _p: None)
+    assert any(u.endswith("/peerA/t1") for u in fake.deleted)
 
-    assert any(u.endswith("/peerA/t1") for u in fake.deleted), fake.deleted
+
+def test_slow_but_progressing_peer_is_not_abandoned(tmp_path, monkeypatch):
+    # Bytes climb for more polls than _STALL_POLLS, then complete → success, not a premature stall.
+    (tmp_path / "01.flac").write_bytes(b"audio")  # so _retrieve finds the finished file
+    ref = json.dumps({"username": "peerA", "files": [{"filename": r"d\Album\01.flac", "size": 5}]})
+    polls = [
+        {"directories": [{"files": [{"filename": r"d\Album\01.flac", "size": 5,
+                                     "state": "InProgress", "bytesTransferred": b, "id": "t1"}]}]}
+        for b in (10, 20, 30, 40, 50)
+    ] + [
+        {"directories": [{"files": [{"filename": r"d\Album\01.flac", "size": 5,
+                                     "state": "Completed, Succeeded", "bytesTransferred": 5, "id": "t1"}]}]}
+    ]
+    monkeypatch.setattr(slskd_mod, "requests", _SeqRequests(polls))
+    monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
+    monkeypatch.setattr(slskd_mod, "_STALL_POLLS", 2)  # would fire early if progress were ignored
+
+    out = _client(tmp_path).download(ref, str(tmp_path / "dest"), lambda _p: None)
+    assert out["track_count"] == 1
+    assert slskd_mod.requests.deleted == []  # completed cleanly → nothing cancelled
+
+
+def test_parse_ranks_free_and_fast_peers_first():
+    responses = [
+        {"username": "slow", "hasFreeUploadSlot": True, "uploadSpeed": 100, "queueLength": 0,
+         "files": [{"filename": r"a\Artist - Album\01.flac", "size": 5}]},
+        {"username": "fast", "hasFreeUploadSlot": True, "uploadSpeed": 9000, "queueLength": 0,
+         "files": [{"filename": r"b\Artist - Album\01.flac", "size": 5}]},
+        {"username": "queued", "hasFreeUploadSlot": False, "uploadSpeed": 99999, "queueLength": 40,
+         "files": [{"filename": r"c\Artist - Album\01.flac", "size": 5}]},
+    ]
+    peers = [json.loads(c["source_ref"])["username"] for c in _parse_search_responses(responses)]
+    assert peers == ["fast", "slow", "queued"]  # free+fast first; no-slot peer last despite speed