From 1fc61a0f5aa4c11c1741576d6d22201c3abba6ad Mon Sep 17 00:00:00 2001 From: Jonathan Date: Wed, 15 Jul 2026 19:03:05 +0200 Subject: [PATCH] fix(worker): cancel abandoned slskd transfers + cap download fall-through Two compounding causes of 'many downloads of one album from different sources' in slskd: (1) the slskd client raised on timeout/error WITHOUT cancelling the transfer it enqueued, so as the pipeline fell through to other peers each left a transfer running; (2) the incomplete-download fall-through was uncapped, grinding through every candidate of a popular album (~200 Soulseek peers), amplified now that Qobuz pacing pushes most jobs to Soulseek. - SlskdClient.download now cancels its enqueued transfers (DELETE) on any abandon. - The download loop caps at _MAX_DOWNLOAD_ATTEMPTS (6) candidates. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/adapters/_slskd.py | 25 +++++++ worker/lyra_worker/pipeline.py | 9 +++ worker/tests/test_incomplete_fallthrough.py | 34 ++++++++++ worker/tests/test_slskd_cancel.py | 72 +++++++++++++++++++++ 4 files changed, 140 insertions(+) create mode 100644 worker/tests/test_slskd_cancel.py diff --git a/worker/lyra_worker/adapters/_slskd.py b/worker/lyra_worker/adapters/_slskd.py index c668424..a8a9064 100644 --- a/worker/lyra_worker/adapters/_slskd.py +++ b/worker/lyra_worker/adapters/_slskd.py @@ -119,7 +119,17 @@ class SlskdClient: files = ref["files"] wanted = {f["filename"] for f in files} os.makedirs(dest, exist_ok=True) + xfer_ids: set[str] = set() # slskd transfer ids we started, for cancel-on-abandon + try: + return self._download(username, files, wanted, dest, on_progress, xfer_ids) + except Exception: + # Abandoning this peer (timeout / error / the pipeline moving on): cancel the transfers + # we enqueued so they don't keep running and pile up in slskd as the job falls through + # to other peers/sources for the same album. + self._cancel_transfers(username, xfer_ids) + raise + def _download(self, username, files, wanted, dest, on_progress, xfer_ids: set) -> dict: r = requests.post( f"{self._url}/api/v0/transfers/downloads/{username}", headers=self._headers(), @@ -139,6 +149,8 @@ class SlskdClient: for f in directory.get("files", []): if f.get("filename") in wanted: # only the files we enqueued states.append(str(f.get("state", ""))) + if f.get("id"): # remember the transfer so we can cancel it if abandoned + xfer_ids.add(str(f["id"])) if states: if any(bad in s for s in states for bad in ("Errored", "Rejected", "Cancelled")): raise RuntimeError("soulseek transfer failed") @@ -162,6 +174,19 @@ class SlskdClient: ) return {"track_count": moved, "path": dest} + def _cancel_transfers(self, username: str, xfer_ids: set) -> None: + """Cancel the given in-flight slskd transfers for a peer. Best-effort: a cleanup failure + must never mask the original download failure.""" + for tid in xfer_ids: + try: + requests.delete( + f"{self._url}/api/v0/transfers/downloads/{username}/{tid}", + headers=self._headers(), + timeout=15, + ) + except Exception: + pass + def _retrieve(self, files: list[dict], dest: str) -> int: """Move the just-finished slskd files out of its downloads dir into `dest`. diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index 2e49892..8a6363d 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -24,6 +24,11 @@ _AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"} # track COUNT right. Generous, so edition/encoding differences don't false-positive; bonus # tracks (over-long) are always fine. _DURATION_MIN_RATIO = 0.85 +# Cap how many ranked candidates a single job will actually download+verify before giving up. +# Without this, the incomplete-download fall-through can grind through dozens of sources (a popular +# album returns ~200 Soulseek candidates), each a slow attempt that also leaves an abandoned slskd +# transfer. The best candidates rank first, so 6 attempts is plenty. +_MAX_DOWNLOAD_ATTEMPTS = 6 def _measure_staged_duration_s(staging: str) -> float | None: @@ -397,10 +402,14 @@ def run_pipeline( _poller.start() try: try: + attempts = 0 for candidate in ranked: adapter = by_source.get(candidate.source) if adapter is None: continue + if attempts >= _MAX_DOWNLOAD_ATTEMPTS: + break # don't grind through every peer/source of a popular album + attempts += 1 _mark_chosen(conn, job_id, candidate.source_ref) # reflect the source now in flight shutil.rmtree(staging, ignore_errors=True) # clean slate per attempt _reports.clear() # this attempt hasn't reported yet → poller estimates until it does diff --git a/worker/tests/test_incomplete_fallthrough.py b/worker/tests/test_incomplete_fallthrough.py index aea3523..7bed420 100644 --- a/worker/tests/test_incomplete_fallthrough.py +++ b/worker/tests/test_incomplete_fallthrough.py @@ -89,3 +89,37 @@ def test_all_sources_incomplete_still_needs_attention(conn, tmp_path): run_pipeline(conn, job_id, [_Qobuz(deliver=2), _Soulseek(deliver=1)], dest_root=str(tmp_path)) assert _state_error(conn, job_id) == ("needs_attention", "incomplete download") assert not (tmp_path / "Radiohead" / "In Rainbows").exists() # nothing imported + + +class _CountingIncompleteQobuz: + """Offers `n` candidates that all download but come back incomplete; counts download calls.""" + name = "qobuz" + tier = 0 + + def __init__(self, n): + self.n = n + self.calls = 0 + + def health(self): + return True + + def search(self, target): + q = _HIRES + return [Candidate(source="qobuz", source_ref=f"r{i}", matched_artist=target.artist, + matched_album=target.album, quality=q, track_count=3, source_tier=0) + for i in range(self.n)] + + def download(self, candidate, dest, on_progress): + self.calls += 1 + _stage(dest, 1) # 1 of 3 → always incomplete + return DownloadResult(ok=True, path=dest, track_count=1) + + +def test_fall_through_is_capped(conn): + from lyra_worker.pipeline import _MAX_DOWNLOAD_ATTEMPTS + a = _CountingIncompleteQobuz(n=20) + job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") + claim_next(conn) + run_pipeline(conn, job_id, [a], dest_root="/tmp/lib-cap") + assert _state_error(conn, job_id) == ("needs_attention", "incomplete download") + assert a.calls == _MAX_DOWNLOAD_ATTEMPTS # capped, not all 20 peers tried diff --git a/worker/tests/test_slskd_cancel.py b/worker/tests/test_slskd_cancel.py new file mode 100644 index 0000000..c280c55 --- /dev/null +++ b/worker/tests/test_slskd_cancel.py @@ -0,0 +1,72 @@ +"""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).""" +import json + +import pytest + +import lyra_worker.adapters._slskd as slskd_mod +from lyra_worker.adapters._slskd import SlskdClient + + +class _Resp: + def __init__(self, payload=None): + self._p = payload if payload is not None else {} + + def raise_for_status(self): + pass + + def json(self): + 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"): + 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) + + def delete(self, url, **kw): + self.deleted.append(url) + return _Resp({}) + + +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}]}) + + +def test_cancels_transfer_on_timeout(tmp_path, monkeypatch): + fake = _FakeRequests(state="InProgress") + monkeypatch.setattr(slskd_mod, "requests", fake) + monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None) + monkeypatch.setattr(slskd_mod, "_XFER_POLLS", 2) + + 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 + + +def test_cancels_transfer_on_error_state(tmp_path, monkeypatch): + fake = _FakeRequests(state="Completed, Errored") + monkeypatch.setattr(slskd_mod, "requests", fake) + monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None) + monkeypatch.setattr(slskd_mod, "_XFER_POLLS", 5) + + 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), fake.deleted