diff --git a/worker/lyra_worker/adapters/_slskd.py b/worker/lyra_worker/adapters/_slskd.py index a8a9064..7fba1c0 100644 --- a/worker/lyra_worker/adapters/_slskd.py +++ b/worker/lyra_worker/adapters/_slskd.py @@ -10,8 +10,9 @@ _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 +_SEARCH_POLLS = 40 # * 3s ≈ 2 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}") diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index a62a585..4f5b30f 100644 --- a/worker/lyra_worker/pipeline.py +++ b/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 diff --git a/worker/tests/test_candidate_filter.py b/worker/tests/test_candidate_filter.py new file mode 100644 index 0000000..118bf25 --- /dev/null +++ b/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 diff --git a/worker/tests/test_slskd_cancel.py b/worker/tests/test_slskd_cancel.py index c280c55..f6c61ee 100644 --- a/worker/tests/test_slskd_cancel.py +++ b/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