fix(slskd): try a batch enqueue once more when slskd answers 500 (#1590)

slskd answers a bare 500 when the peer doesn't reply to a queue request
inside its own 5 second wait (its log says "The wait timed out after
5000 milliseconds"). The peer often answers a moment later, so a
Soulseek download that would have worked failed on the first try.

The batch enqueue is now tried twice with 5 seconds between. A 4xx or a
connection failure is not retried, only a 5xx. While it waits the card
reads "The peer did not answer, trying again", and cancelling during the
wait stops without the second try. If the second try fails too, the
error message carries what slskd said (trimmed to 160 characters).
Before, that text was dropped and the card only showed the status line.

It's been running on my instance since mid-October, and the timeouts
still show up in slskd's log, so I can't say how many it saved. The five
new tests cover the retry that works, giving up after the second 500, a
400 and a connection error not being retried, and cancelling mid-wait.
Ruff, basedpyright and the slskd tests pass.

shelfmark/release_sources/slskd/handler.py

@@ -84,6 +84,32 @@ _SECONDS_PER_MINUTE = 60
 
 _SLSKD_ERRORS = (requests.exceptions.RequestException, SlskdError, ValueError, TypeError)
 
+# slskd gives up on a peer that does not answer a queue request after 5 seconds and returns a
+# 500. One more try, a few seconds later, costs little and often works.
+_ENQUEUE_ATTEMPTS = 2
+_ENQUEUE_RETRY_WAIT_SECONDS = 5.0
+_HTTP_SERVER_ERROR = 500
+_ERROR_BODY_CHARS = 160
+
+
+def _is_server_error(exc: BaseException) -> bool:
+    response = getattr(exc, "response", None)
+    return (
+        isinstance(exc, requests.exceptions.HTTPError)
+        and response is not None
+        and response.status_code >= _HTTP_SERVER_ERROR
+    )
+
+
+def _error_text(exc: BaseException) -> str:
+    """The error, with what slskd said when it answered with a body."""
+    response = getattr(exc, "response", None)
+    body = ""
+    if isinstance(exc, requests.exceptions.HTTPError) and response is not None:
+        with suppress(Exception):
+            body = " ".join(str(response.text or "").split())[:_ERROR_BODY_CHARS]
+    return f"{exc} ({body})" if body else str(exc)
+
 
 @dataclass(frozen=True)
 class SlskdDownloadSpec:
@@ -799,14 +825,33 @@ class SlskdHandler(DownloadHandler):
             _remove_files(leftovers)
 
         status_callback("resolving", "Sending to slskd")
-        try:
-            batch = client.enqueue_batch(spec.username, spec.files, destination)
-        except SlskdBatchUnsupportedError:
-            return _USE_ORIGINAL_ENQUEUE
-        except _SLSKD_ERRORS as e:
-            logger.exception("Failed to enqueue slskd batch")
-            status_callback("error", f"Failed to add to slskd: {e}")
-            return None
+        attempt = 0
+        while True:
+            attempt += 1
+            try:
+                batch = client.enqueue_batch(spec.username, spec.files, destination)
+                break
+            except SlskdBatchUnsupportedError:
+                return _USE_ORIGINAL_ENQUEUE
+            except _SLSKD_ERRORS as e:
+                if attempt < _ENQUEUE_ATTEMPTS and _is_server_error(e):
+                    # slskd answers 500 when the peer does not reply to the queue request
+                    # inside its own 5 second wait. A peer that missed it once often answers
+                    # a moment later.
+                    logger.warning(
+                        "slskd could not queue %s from %s (%s), trying again",
+                        task.task_id,
+                        spec.username,
+                        _error_text(e),
+                    )
+                    status_callback("resolving", "The peer did not answer, trying again")
+                    if cancel_flag.wait(timeout=_ENQUEUE_RETRY_WAIT_SECONDS):
+                        status_callback("cancelled", "Cancelled")
+                        return None
+                    continue
+                logger.error("Failed to enqueue slskd batch: %s", _error_text(e), exc_info=True)
+                status_callback("error", f"Failed to add to slskd: {_error_text(e)}")
+                return None
 
         ids = {
             str(t.get("filename") or ""): str(t["id"])

tests/slskd/test_handler_isolated.py

@@ -13,6 +13,7 @@ import json
 from threading import Event
 
 import pytest
+import requests
 
 import shelfmark.release_sources.slskd.handler as mod
 from shelfmark.core.models import DownloadTask
@@ -339,6 +340,100 @@ class TestFailuresAndCancel:
         assert fake.cancelled
 
 
+def _http_error(status, body=""):
+    response = requests.Response()
+    response.status_code = status
+    response._content = body.encode()
+    return requests.exceptions.HTTPError(f"{status} Server Error", response=response)
+
+
+class _FlakyEnqueue(FakeSlskd):
+    """A slskd whose first ``failures`` enqueue calls fail with the given error."""
+
+    def __init__(self, root, *, failures, error, **kwargs):
+        super().__init__(root, **kwargs)
+        self.failures = failures
+        self.error = error
+        self.attempts = 0
+
+    def enqueue_batch(self, username, files, destination):
+        self.attempts += 1
+        if self.attempts <= self.failures:
+            raise self.error
+        return super().enqueue_batch(username, files, destination)
+
+
+class TestEnqueueRetry:
+    @pytest.fixture(autouse=True)
+    def _no_wait(self, monkeypatch):
+        monkeypatch.setattr(mod, "_ENQUEUE_RETRY_WAIT_SECONDS", 0.0)
+
+    def test_a_500_from_a_peer_that_did_not_answer_is_tried_once_more(self, handler, cfg, tmp_path):
+        fake = _FlakyEnqueue(tmp_path, failures=1, error=_http_error(500, "timed out"))
+
+        result, _, rec = _run(handler, fake)
+
+        assert result is not None
+        assert fake.attempts == 2
+        assert ("resolving", "The peer did not answer, trying again") in rec.real
+        assert rec.real[-1][0] != "error"
+
+    def test_it_gives_up_after_the_second_500_and_says_what_slskd_said(
+        self, handler, cfg, tmp_path
+    ):
+        fake = _FlakyEnqueue(
+            tmp_path, failures=5, error=_http_error(500, "The wait timed out after 5000 ms")
+        )
+
+        result, _, rec = _run(handler, fake)
+
+        assert result is None
+        assert fake.attempts == 2
+        status, message = rec.real[-1]
+        assert status == "error"
+        assert message.startswith("Failed to add to slskd: 500 Server Error")
+        assert "The wait timed out after 5000 ms" in message
+
+    def test_a_client_error_is_not_retried(self, handler, cfg, tmp_path):
+        fake = _FlakyEnqueue(tmp_path, failures=5, error=_http_error(400, "bad request"))
+
+        result, _, rec = _run(handler, fake)
+
+        assert result is None
+        assert fake.attempts == 1
+        assert rec.real[-1][0] == "error"
+
+    def test_a_connection_error_is_not_retried(self, handler, cfg, tmp_path):
+        fake = _FlakyEnqueue(
+            tmp_path, failures=5, error=requests.exceptions.ConnectionError("refused")
+        )
+
+        result, _, _rec = _run(handler, fake)
+
+        assert result is None
+        assert fake.attempts == 1
+
+    def test_cancelling_during_the_wait_stops_without_a_second_try(
+        self, handler, cfg, tmp_path, monkeypatch
+    ):
+        monkeypatch.setattr(mod, "_ENQUEUE_RETRY_WAIT_SECONDS", 60.0)
+        cancel = Event()
+        fake = _FlakyEnqueue(tmp_path, failures=5, error=_http_error(500))
+        original = fake.enqueue_batch
+
+        def enqueue_then_cancel(username, files, destination):
+            cancel.set()
+            return original(username, files, destination)
+
+        fake.enqueue_batch = enqueue_then_cancel
+
+        result, _, rec = _run(handler, fake, cancel=cancel)
+
+        assert result is None
+        assert fake.attempts == 1
+        assert rec.real[-1][0] == "cancelled"
+
+
 class TestPeerQueue:
     def test_a_queued_transfer_asks_the_orchestrator_not_to_stall_it(self, handler, cfg, tmp_path):
         fake = FakeSlskd(