diff --git a/README.md b/README.md index da4e7c1..b794857 100644 --- a/README.md +++ b/README.md @@ -258,6 +258,20 @@ A download that sits with no bytes for `MUSE_STALL_MINUTES` is abandoned the same way, as is one slskd has forgotten about. Both used to leave the job in `downloading` forever, with a cheerful pulsing dot and no log line. +**Not every failure is about the file.** "File not shared" concerns one file and +the next candidate is unaffected, so the ladder advances immediately. A timeout +says only that slskd could not talk to anyone just then, and the next candidate +would fail identically. Walking the whole list to learn that once is waste -- +worse, it is harmful: each enqueue blocks about five seconds on a peer +connection, so eight back-to-back attempts is forty seconds of pressure on one +subsystem, and it takes other users' searches down with it. That is not +hypothetical; it is what a real request did. + +So attempts are spaced by `MUSE_ATTEMPT_BACKOFF`, and after +`MUSE_SYSTEMIC_LIMIT` consecutive connection failures muse stops and **keeps the +untried candidates on the row**. Retrying then resumes the ladder where it +stopped rather than paying for a search whose answer has not changed. + ## Review: when muse isn't sure Picking a file and *trusting* it are different questions, so they use different @@ -548,6 +562,8 @@ stat -c "%h" "/data/media/music// ()/ - .flac" | `MUSE_TITLE_FLOOR` | `0.5` | the title alone must clear this before the artist counts | | `MUSE_CANDIDATES` | `8` | ranked candidates kept per request, for fallback | | `MUSE_MAX_ATTEMPTS` | `5` | peers tried before a request is given up on | +| `MUSE_ATTEMPT_BACKOFF` | `4` | seconds between candidate attempts | +| `MUSE_SYSTEMIC_LIMIT` | `2` | consecutive connection failures before stopping | | `MUSE_STALL_MINUTES` | `20` | a transfer moving no bytes for this long is abandoned | | `MUSE_MAX_QUERIES` | `4` | rungs of the search ladder to try | | `MUSE_MIN_CANDIDATES` | `3` | below this, escalate to the next query | diff --git a/app.py b/app.py index 02f4f31..8b82e4c 100644 --- a/app.py +++ b/app.py @@ -100,6 +100,17 @@ SEARCH_RETRIES = int(os.getenv("MUSE_SEARCH_RETRIES", "3")) # Soulseek results trickle in: what you have at t=5s is much worse than t=15s. SEARCH_POLLS = int(os.getenv("MUSE_SEARCH_POLLS", "20")) +# Pause between candidate attempts. Without one, a ladder of 8 walks the whole +# list inside a single bad slskd window -- each enqueue blocks ~5s on a peer +# connection, so eight attempts is 40 seconds of hammering one subsystem, and +# it takes other people's searches down with it. +ATTEMPT_BACKOFF = float(os.getenv("MUSE_ATTEMPT_BACKOFF", "4")) + +# Consecutive systemic failures before we stop walking the ladder. A file that +# is not shared says something about that file; a timeout says only that slskd +# is having a moment, and burning eight candidates to learn it once is waste. +SYSTEMIC_LIMIT = int(os.getenv("MUSE_SYSTEMIC_LIMIT", "2")) + # How many Soulseek searches may run at once. # # This is a hard cap, not a suggestion. Each search takes 10-40s of polling @@ -911,6 +922,18 @@ def exhausted_detail(tried: list[dict], attempts: int) -> str: return f"tried {n_peers} peer{plural}: {joined}" +# Substrings in a failure that describe slskd's state rather than the file's. +# "File not shared" is about one file and the next candidate is unaffected; +# a timeout means the client could not talk to anyone, so the next candidate +# would fail for the same reason and only costs us the candidate. +SYSTEMIC = ("timed out", "timeout", "connection", "refused", "unreachable") + + +def is_systemic(reason: str) -> bool: + r = reason.lower() + return any(w in r for w in SYSTEMIC) + + async def try_next_candidate(job_id: str, artist: str, title: str, album: str | None = None) -> bool: """Enqueue the next-best candidate. False once the ladder is exhausted. @@ -930,7 +953,15 @@ async def try_next_candidate(job_id: str, artist: str, title: str, tried = json.loads(row["tried"] or "[]") attempts = row["attempts"] or 0 + systemic_run = 0 + first = True while cands and attempts < MAX_ATTEMPTS: + # Space the attempts out. The first one is immediate; the rest wait, + # because back-to-back enqueues are what saturated slskd. + if not first: + await asyncio.sleep(ATTEMPT_BACKOFF) + first = False + cand = cands.pop(0) attempts += 1 try: @@ -941,11 +972,33 @@ async def try_next_candidate(job_id: str, artist: str, title: str, "size": cand["size"]}]) er.raise_for_status() except Exception as e: + reason = f"could not reach peer: {errstr(e)}" tried.append({"username": cand["username"], "filename": cand["filename"], - "reason": f"could not reach peer: {errstr(e)}"}) + "reason": reason}) + if is_systemic(reason): + systemic_run += 1 + if systemic_run >= SYSTEMIC_LIMIT: + # Stop, and leave the untried candidates on the row so a + # retry resumes the ladder instead of re-running a search + # that already told us the right answer. + with db() as c: + c.execute("UPDATE jobs SET candidates=?, tried=?," + " attempts=? WHERE id=?", + (json.dumps(cands), json.dumps(tried), + attempts, job_id)) + set_state(job_id, "failed", + f"slskd could not reach any peer after " + f"{attempts} attempts -- {len(cands)} candidate" + f"{'' if len(cands) == 1 else 's'} left to try, " + f"so a retry resumes where this stopped") + return False + else: + systemic_run = 0 continue + systemic_run = 0 + conf, why = confidence(cand["filename"], artist, title, album) with db() as c: c.execute( @@ -1239,8 +1292,20 @@ async def search_worker(n: int): job = c.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone() if job and job["state"] in ("queued", "searching"): set_state(job_id, "searching", "") - await pick_and_enqueue(job_id, job["artist"], job["title"], - job["album"]) + # A ladder left over from a run that stopped on a slskd stall + # is still the right answer -- the search that produced it has + # not become wrong. Resume it rather than paying for another. + keys = job.keys() + left = json.loads( + (job["candidates"] if "candidates" in keys else None) or "[]") + if left: + log.info("job %s resuming ladder, %d candidate(s) left", + job_id[:8], len(left)) + await try_next_candidate(job_id, job["artist"], + job["title"], job["album"]) + else: + await pick_and_enqueue(job_id, job["artist"], job["title"], + job["album"]) except Exception as e: log.warning("worker %d: %s", n, e) try: @@ -2533,13 +2598,18 @@ async def retry(job_id: str, request: Request): if row["state"] != "failed": raise HTTPException(409, f"only failed jobs can be retried " f"(this one is {row['state']})") - # Clear the previous attempt, or the ladder resumes mid-way through a - # candidate list gathered for a search that has long since expired. + # Clear the last transfer, but keep any candidates the previous run did + # not get to: those are left behind deliberately when slskd could not + # reach anyone, and re-running the search would only rediscover them. with db() as c: + row = c.execute("SELECT candidates FROM jobs WHERE id=?", + (job_id,)).fetchone() + keep = bool(json.loads((row["candidates"] if row else None) or "[]")) c.execute("UPDATE jobs SET username=NULL, remote_file=NULL, size=NULL," - " confidence=NULL, match_why=NULL, candidates=NULL," - " tried=NULL, attempts=0, started_at=NULL WHERE id=?", - (job_id,)) + " confidence=NULL, match_why=NULL, attempts=0," + " started_at=NULL" + + ("" if keep else ", candidates=NULL, tried=NULL") + + " WHERE id=?", (job_id,)) set_state(job_id, "queued", "") await work_q.put(job_id) return {"ok": True} diff --git a/tests/test_acquisition.py b/tests/test_acquisition.py index d3b9ab1..7061840 100644 --- a/tests/test_acquisition.py +++ b/tests/test_acquisition.py @@ -218,6 +218,31 @@ def main() -> int: check("unreachable peers are reported as such", "could not reach peer" in (r["tried"] or ""), r["detail"]) + print("\nsystemic failures do not burn the ladder") + # slskd timing out says nothing about any particular file, so walking the + # whole list to learn it once is waste -- and it is what took a real + # request down: five enqueues in twenty-five seconds during one slskd stall. + fake4 = FakeSlskd(per_query={"Chicago": [ + peer(f"p{i}", r"x\Michael Jackson\Xscape\Chicago.flac") for i in range(8)]}) + + class Timeouts(FakeSlskd): + async def post(self, url, json=None): + if url != "/api/v0/searches": + raise RuntimeError("The wait timed out after 5000 milliseconds") + return await FakeSlskd.post(self, url, json=json) + + t = Timeouts(per_query=fake4.per_query) + app.slskd_client = t + make_job("j5", "Michael Jackson", "Chicago") + asyncio.run(app.pick_and_enqueue("j5", "Michael Jackson", "Chicago")) + r = job_row("j5") + check("stops after SYSTEMIC_LIMIT instead of walking the whole list", + r["attempts"] == app.SYSTEMIC_LIMIT, r["attempts"]) + left = json.loads(r["candidates"] or "[]") + check("untried candidates are preserved for a retry", len(left) > 0, len(left)) + check("the message says a retry will resume", + "retry resumes" in (r["detail"] or ""), r["detail"]) + print("\nempty results") fake3 = FakeSlskd(per_query={}) app.slskd_client = fake3