fix(downloader): throttle batch torrent fetches and back off on HTTP 429 (#1052)

Batch-collecting a season fired all .torrent file downloads at once via
an unbounded asyncio.gather, tripping nyaa's rate limiter; get_url then
gave up on the first 429 with no retry, so the whole batch silently
failed.

- add_torrent now fetches torrent files serially per host with a 1s
  delay between requests (parallel across hosts), mirroring the
  RSS_PER_HOST_DELAY pattern from #1026
- get_url retries 429 responses within the existing retry budget,
  honoring a numeric Retry-After header (clamped to 60s) and falling
  back to a 10s delay otherwise

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JhTduCYBGeHETU7dSWoh37

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}"

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}"

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 = [

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()