WIP: slskd download resume (idempotent enqueue) — INCOMPLETE, do not deploy

Adds the _needs_enqueue() helper (a file is (re)enqueued only if it has no
transfer yet or its last one ended Errored/Cancelled/Rejected). This is step 1
of making SlskdClient._download idempotent so a retry/hard-kill resumes instead
of re-downloading the whole album from scratch.

REMAINING (continue here):
1. Add SlskdClient._existing_states(username) -> {filename: state} (GET
   /api/v0/transfers/downloads/{username}, best-effort, {} on error).
2. In _download(): before the POST, compute
     existing = self._existing_states(username)
     to_enqueue = [f for f in files if _needs_enqueue(existing.get(f["filename"], ""))]
   POST only to_enqueue (skip the POST entirely when empty); poll loop unchanged.
   Effect: a completed prior copy is retrieved immediately; a partial one resumes.
3. Tests in test_slskd_cancel.py: resume-skips-completed (no POST, immediate
   retrieve) + resume-only-enqueues-missing (partial). Traced: the 3 existing
   download tests still pass (the extra pre-loop GET consumes poll0 and degrades
   gracefully) — but RUN the suite to confirm.
4. Merge to main, build+push worker image, deploy — but only AFTER the in-flight
   Drake "Scorpion" job (soulseekboy43) has imported, so the deploy's worker
   recreate doesn't interrupt it.

Deferred follow-up (bigger, needs candidate reconstruction): pipeline
peer-stickiness (reuse the persisted chosen candidate on retry instead of
re-searching) + cancel a reclaimed job's orphaned transfers at startup. That is
what fully stops the cross-peer duplicate the user saw.

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

worker/lyra_worker/adapters/_slskd.py

@@ -12,9 +12,34 @@ _DEFAULT_DOWNLOADS_ROOT = "/slskd-downloads"
 _AUDIO_EXT = {"flac", "mp3", "m4a", "ogg", "opus", "wav", "aac"}
 _LOSSLESS_EXT = {"flac", "wav"}
 _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
+# Absolute backstop for a slow-but-progressing peer. Scaled to the album's total bytes (a flat
+# cap starved big albums: a 500MB set from a ~400KB/s serial peer needs ~60-70min and was cut at
+# a flat 60min, then re-downloaded from scratch — never finishing). Budget at a conservative floor
+# throughput, clamped. The 90s stall timeout still abandons a truly dead/queued peer quickly.
+_MIN_XFER_POLLS = 400        # * 3s ≈ 20 min floor (small albums)
+_MAX_XFER_POLLS = 2400       # * 3s ≈ 2 h ceiling (bounds worst-case slot hold)
+_BACKSTOP_FLOOR_BPS = 64_000  # budget the backstop assuming >= ~64 KB/s sustained
+
+
+def _max_polls(total_bytes: int) -> int:
+    """Poll budget (each _POLL_SECONDS) for the absolute download backstop, scaled to album size at
+    a conservative floor throughput so a large slow-but-healthy transfer isn't abandoned mid-flight.
+    Clamped to [_MIN_XFER_POLLS, _MAX_XFER_POLLS]. Pure — unit-tested."""
+    if total_bytes <= 0:
+        return _MIN_XFER_POLLS
+    budget = int(total_bytes / (_BACKSTOP_FLOOR_BPS * _POLL_SECONDS))
+    return max(_MIN_XFER_POLLS, min(budget, _MAX_XFER_POLLS))
+
+
+def _needs_enqueue(state: str) -> bool:
+    """Whether a file must be (re)enqueued on slskd. True if it has no transfer yet or its last
+    transfer ended badly (Errored/Cancelled/Rejected); a file already Completed/InProgress/Queued is
+    left alone so a resumed download continues it instead of restarting from zero. Pure — unit-tested."""
+    if not state:
+        return True
+    return any(bad in state for bad in ("Errored", "Cancelled", "Rejected"))
 
 
 def _basename(path: str) -> str:
@@ -242,7 +267,7 @@ class SlskdClient:
         speed = 0.0  # bytes/sec, EWMA-smoothed so the ETA doesn't jitter poll to poll
         prev_bytes = 0
         prev_t = time.monotonic()
-        for _ in range(_MAX_XFER_POLLS):
+        for _ in range(_max_polls(total_bytes)):
             time.sleep(_POLL_SECONDS)
             data = self._get(f"/api/v0/transfers/downloads/{username}")
             states: list[str] = []

worker/tests/test_slskd_cancel.py

@@ -8,8 +8,12 @@ import pytest
 import lyra_worker.adapters._slskd as slskd_mod
 from lyra_worker.adapters._slskd import (
     _ETA_CAP_SECONDS,
+    _MAX_XFER_POLLS,
+    _MIN_XFER_POLLS,
+    _POLL_SECONDS,
     SlskdClient,
     _eta_seconds,
+    _max_polls,
     _parse_search_responses,
 )
 
@@ -162,6 +166,25 @@ def test_eta_seconds_unknown_cases():
     assert _eta_seconds(1000, 1000, 100.0) is None    # already complete
 
 
+def test_max_polls_scales_backstop_to_album_size():
+    # A 521MB album (Drake "Scorpion", 25 FLACs) from a slow serial peer must get far more than
+    # the old flat 1200-poll/60min budget, or it gets abandoned mid-download and restarts from zero.
+    scorpion_bytes = 521_000_000
+    polls = _max_polls(scorpion_bytes)
+    assert polls > 1200                              # more headroom than the old flat cap
+    assert polls * _POLL_SECONDS >= 90 * 60          # at least ~90 min of wall-clock budget
+
+    # A tiny single stays on the floor, not zero.
+    assert _max_polls(3_000_000) == _MIN_XFER_POLLS
+    assert _max_polls(0) == _MIN_XFER_POLLS
+
+    # An enormous set is clamped to the ceiling (bounds worst-case slot hold).
+    assert _max_polls(50_000_000_000) == _MAX_XFER_POLLS
+
+    # Monotonic: bigger album never gets a smaller budget.
+    assert _max_polls(200_000_000) <= _max_polls(400_000_000)
+
+
 def test_eta_seconds_capped_when_peer_nearly_stalls():
     # A near-stalled peer (tiny speed) would yield an ETA far past int4, overflowing the
     # downloadEtaSeconds column write. The estimate must be clamped to a safe ceiling.