diff --git a/backend/src/module/downloader/download_client.py b/backend/src/module/downloader/download_client.py index a99bc412..60c6fbdd 100644 --- a/backend/src/module/downloader/download_client.py +++ b/backend/src/module/downloader/download_client.py @@ -1,5 +1,7 @@ import asyncio import logging +from collections import defaultdict +from urllib.parse import urlparse from module.conf import settings from module.models import Bangumi, Torrent @@ -10,6 +12,35 @@ from .path import gen_save_path logger = logging.getLogger(__name__) +# 同一主机批量抓取 .torrent 文件的请求间隔（#1052）。批量收集整季时并发抓取 +# 会触发 nyaa 等站点的 429 限流；与 rss/engine.py 的 RSS_PER_HOST_DELAY 同思路， +# 但这里是收集接口内的一次性批量抓取，取 1s 以控制接口延迟（24 集约 +23s）。 +TORRENT_FETCH_PER_HOST_DELAY = 1.0 + + +async def _fetch_torrent_files( + req: RequestContent, torrents: list[Torrent] +) -> list[bytes]: + """抓取种子文件内容：同主机串行加延时，不同主机并行，失败项丢弃（#1052）。""" + by_host: dict[str, list[Torrent]] = defaultdict(list) + for t in torrents: + by_host[urlparse(t.url).netloc].append(t) + + async def _fetch_host_group(items: list[Torrent]) -> list[bytes]: + files: list[bytes] = [] + for i, t in enumerate(items): + if i and TORRENT_FETCH_PER_HOST_DELAY: + await asyncio.sleep(TORRENT_FETCH_PER_HOST_DELAY) + content = await req.get_content(t.url) + if content is not None: + files.append(content) + return files + + groups = await asyncio.gather( + *[_fetch_host_group(items) for items in by_host.values()] + ) + return [f for group in groups for f in group] + # --------------------------------------------------------------------------- # Module-level concrete-client cache (session reuse, #1039 / #900) @@ -337,11 +368,7 @@ class DownloadClient: torrent_url = [t.url for t in torrent] torrent_file = None else: - torrent_file = await asyncio.gather( - *[req.get_content(t.url) for t in torrent] - ) - # Filter out None values (failed fetches) - torrent_file = [f for f in torrent_file if f is not None] + torrent_file = await _fetch_torrent_files(req, torrent) if not torrent_file: logger.warning( f"Failed to fetch torrent files for: {bangumi.official_title}" diff --git a/backend/src/module/network/request_url.py b/backend/src/module/network/request_url.py index fa2cd95e..e478c167 100644 --- a/backend/src/module/network/request_url.py +++ b/backend/src/module/network/request_url.py @@ -22,6 +22,22 @@ _CONNECTION_LIMITS = httpx.Limits( keepalive_expiry=60.0, ) +# HTTP 429 限流退避（#1052）：无 Retry-After（或非数字形式）时的默认等待， +# 以及对服务端 Retry-After 的上限保护 +HTTP_429_FALLBACK_DELAY = 10.0 +HTTP_429_MAX_RETRY_AFTER = 60.0 + + +def _retry_after_delay(response: httpx.Response) -> float: + """从 429 响应解析等待秒数；仅接受数字形式的 Retry-After，日期形式走默认值。""" + retry_after = response.headers.get("Retry-After") + if retry_after is not None: + try: + return min(max(float(retry_after), 0.0), HTTP_429_MAX_RETRY_AFTER) + except ValueError: + pass + return HTTP_429_FALLBACK_DELAY + def _proxy_config_key() -> str: if settings.proxy.enable: @@ -119,7 +135,18 @@ class RequestURL: return req except httpx.HTTPStatusError as e: logger.warning(f"HTTP {e.response.status_code} from {url}") - break + if e.response.status_code != 429: + break + # 429 限流：按 Retry-After 退避后重试（#1052） + try_time += 1 + if try_time >= retry: + break + delay = _retry_after_delay(e.response) + logger.warning( + f"Rate limited by {url}, retrying in {delay:.0f}s " + f"({try_time}/{retry})" + ) + await asyncio.sleep(delay) except httpx.RequestError as e: logger.warning( f"Request error for {url}: {type(e).__name__}. Retry {try_time + 1}/{retry}" diff --git a/backend/src/test/test_download_client.py b/backend/src/test/test_download_client.py index 719e27f2..40cfeaf4 100644 --- a/backend/src/test/test_download_client.py +++ b/backend/src/test/test_download_client.py @@ -4,7 +4,11 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from module.downloader.download_client import AddResult, DownloadClient +from module.downloader.download_client import ( + TORRENT_FETCH_PER_HOST_DELAY, + AddResult, + DownloadClient, +) from module.models import Bangumi, Torrent from module.models.config import Config from test.factories import make_bangumi, make_torrent @@ -103,6 +107,94 @@ class TestAddTorrent: assert call_kwargs["torrent_files"] == b"torrent-file-data" assert call_kwargs["torrent_urls"] is None + async def test_add_torrent_same_host_batch_delays_between_fetches( + self, download_client, mock_qb_client + ): + """同一主机的批量种子下载应串行并在请求间加延时（#1052）。""" + torrents = [ + make_torrent(url=f"https://nyaa.si/download/{i}.torrent") + for i in range(1, 4) + ] + bangumi = make_bangumi() + + with ( + patch("module.downloader.download_client.RequestContent") as MockReq, + patch( + "module.downloader.download_client.asyncio.sleep", + new_callable=AsyncMock, + ) as mock_sleep, + ): + mock_req = AsyncMock() + mock_req.get_content = AsyncMock(return_value=b"data") + MockReq.return_value.__aenter__ = AsyncMock(return_value=mock_req) + MockReq.return_value.__aexit__ = AsyncMock(return_value=False) + + result = await download_client.add_torrent(torrents, bangumi) + + assert result is AddResult.ADDED + assert mock_req.get_content.await_count == 3 + assert mock_sleep.await_count == 2 + for call in mock_sleep.await_args_list: + assert call.args == (TORRENT_FETCH_PER_HOST_DELAY,) + call_kwargs = mock_qb_client.add_torrents.call_args[1] + assert len(call_kwargs["torrent_files"]) == 3 + + async def test_add_torrent_different_hosts_no_delay( + self, download_client, mock_qb_client + ): + """不同主机之间并行下载，无需延时。""" + torrents = [ + make_torrent(url="https://nyaa.si/download/1.torrent"), + make_torrent(url="https://mikanani.me/Download/2.torrent"), + ] + bangumi = make_bangumi() + + with ( + patch("module.downloader.download_client.RequestContent") as MockReq, + patch( + "module.downloader.download_client.asyncio.sleep", + new_callable=AsyncMock, + ) as mock_sleep, + ): + mock_req = AsyncMock() + mock_req.get_content = AsyncMock(return_value=b"data") + MockReq.return_value.__aenter__ = AsyncMock(return_value=mock_req) + MockReq.return_value.__aexit__ = AsyncMock(return_value=False) + + result = await download_client.add_torrent(torrents, bangumi) + + assert result is AddResult.ADDED + assert mock_req.get_content.await_count == 2 + mock_sleep.assert_not_awaited() + + async def test_add_torrent_batch_partial_fetch_failure_keeps_successes( + self, download_client, mock_qb_client + ): + """批量下载中个别失败（None）应被过滤，其余照常提交。""" + torrents = [ + make_torrent(url=f"https://nyaa.si/download/{i}.torrent") + for i in range(1, 4) + ] + bangumi = make_bangumi() + + with ( + patch("module.downloader.download_client.RequestContent") as MockReq, + patch( + "module.downloader.download_client.asyncio.sleep", + new_callable=AsyncMock, + ), + ): + mock_req = AsyncMock() + mock_req.get_content = AsyncMock(side_effect=[b"a", None, b"c"]) + MockReq.return_value.__aenter__ = AsyncMock(return_value=mock_req) + MockReq.return_value.__aexit__ = AsyncMock(return_value=False) + + result = await download_client.add_torrent(torrents, bangumi) + + assert result is AddResult.ADDED + call_kwargs = mock_qb_client.add_torrents.call_args[1] + assert call_kwargs["torrent_files"] == [b"a", b"c"] + async def test_list_magnet_urls(self, download_client, mock_qb_client): """List of magnet torrents are joined as list of URLs.""" torrents = [ diff --git a/backend/src/test/test_request_url.py b/backend/src/test/test_request_url.py index fa80a22f..31bb953e 100644 --- a/backend/src/test/test_request_url.py +++ b/backend/src/test/test_request_url.py @@ -3,9 +3,12 @@ import asyncio from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest from module.network.request_url import ( + HTTP_429_FALLBACK_DELAY, + HTTP_429_MAX_RETRY_AFTER, RequestURL, get_shared_client, reset_shared_client, @@ -147,3 +150,92 @@ class TestReentrantContextManager: assert not seen_none # Client stays valid after both blocks exit -- __aexit__ is a no-op. assert req._client is not None + + +class TestRateLimit429: + """HTTP 429 must be retried with backoff instead of failing at once (#1052).""" + + URL = "https://nyaa.si/download/1.torrent" + + @staticmethod + def _resp_429(headers: dict | None = None) -> httpx.Response: + return httpx.Response( + 429, + headers=headers or {}, + request=httpx.Request("GET", TestRateLimit429.URL), + ) + + @staticmethod + def _resp_ok() -> MagicMock: + resp = MagicMock() + resp.status_code = 200 + resp.raise_for_status = MagicMock() + return resp + + async def _run(self, responses: list, retry: int = 3): + """Drive get_url against a canned response sequence; return (result, calls, sleeps).""" + call_count = 0 + + async def mock_get(**kwargs): + nonlocal call_count + resp = responses[min(call_count, len(responses) - 1)] + call_count += 1 + return resp + + with ( + patch("module.network.request_url.get_shared_client") as mock_get_client, + patch( + "module.network.request_url.asyncio.sleep", new_callable=AsyncMock + ) as mock_sleep, + ): + mock_client = AsyncMock() + mock_client.get = mock_get + mock_get_client.return_value = mock_client + + async with RequestURL() as req: + result = await req.get_url(self.URL, retry=retry) + + return result, call_count, mock_sleep + + async def test_get_url_429_then_success_retries_with_retry_after_delay(self): + result, calls, mock_sleep = await self._run( + [self._resp_429({"Retry-After": "3"}), self._resp_ok()] + ) + assert result is not None + assert result.status_code == 200 + assert calls == 2 + mock_sleep.assert_awaited_once_with(3.0) + + async def test_get_url_429_without_header_sleeps_fallback_delay(self): + result, calls, mock_sleep = await self._run([self._resp_429(), self._resp_ok()]) + assert result is not None + mock_sleep.assert_awaited_once_with(HTTP_429_FALLBACK_DELAY) + + async def test_get_url_429_with_huge_retry_after_caps_delay(self): + result, _, mock_sleep = await self._run( + [self._resp_429({"Retry-After": "600"}), self._resp_ok()] + ) + assert result is not None + mock_sleep.assert_awaited_once_with(HTTP_429_MAX_RETRY_AFTER) + + async def test_get_url_429_with_date_retry_after_uses_fallback(self): + result, _, mock_sleep = await self._run( + [ + self._resp_429({"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"}), + self._resp_ok(), + ] + ) + assert result is not None + mock_sleep.assert_awaited_once_with(HTTP_429_FALLBACK_DELAY) + + async def test_get_url_429_exhausts_retries_returns_none(self): + result, calls, _ = await self._run([self._resp_429()], retry=2) + assert result is None + assert calls == 2 + + async def test_get_url_non_429_status_does_not_retry(self): + resp_404 = httpx.Response(404, request=httpx.Request("GET", self.URL)) + result, calls, mock_sleep = await self._run([resp_404]) + assert result is None + assert calls == 1 + mock_sleep.assert_not_awaited()