diff --git a/shelfmark/release_sources/slskd/handler.py b/shelfmark/release_sources/slskd/handler.py index 52431476..83d52205 100644 --- a/shelfmark/release_sources/slskd/handler.py +++ b/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"]) diff --git a/tests/slskd/test_handler_isolated.py b/tests/slskd/test_handler_isolated.py index a5f62dd1..c03aef35 100644 --- a/tests/slskd/test_handler_isolated.py +++ b/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(