fix(rss): throttle concurrent fetches per host to avoid HTTP 429 (#1026)

refresh_rss fired every feed at once, so sites with many subscribed feeds
(e.g. nyaa.si) rate-limited the whole batch. Feeds are now grouped by host:
serial with a 2s delay within a host, still parallel across hosts, with the
existing global concurrency cap kept.

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

backend/src/module/rss/engine.py

@@ -1,8 +1,10 @@
 import asyncio
 import logging
 import re
+from collections import defaultdict
 from datetime import datetime, timezone
 from typing import Optional
+from urllib.parse import urlparse
 
 from module.database import Database, engine
 from module.downloader import DownloadClient
@@ -11,6 +13,10 @@ from module.network import RequestContent
 
 logger = logging.getLogger(__name__)
 
+# Delay between consecutive requests to the same host. Firing all feeds of one
+# site at once gets the whole batch rate-limited with HTTP 429 (#1026).
+RSS_PER_HOST_DELAY = 2.0
+
 
 class RSSEngine(Database):
     def __init__(self, _engine=engine):
@@ -149,18 +155,35 @@ class RSSEngine(Database):
         else:
             rss_item = self.rss.search_id(rss_id)
             rss_items = [rss_item] if rss_item else []
-        # From RSS Items, fetch all torrents with concurrency limit
+        # From RSS Items, fetch all torrents: parallel across hosts, serial
+        # (with a delay) within one host so the site never sees a burst (#1026).
         logger.debug("[Engine] Get %s RSS items", len(rss_items))
         semaphore = asyncio.Semaphore(5)
 
-        async def _limited_pull(item):
-            async with semaphore:
-                return await self._pull_rss_with_status(item)
+        async def _pull_host_group(items: list[RSSItem]):
+            group_results = []
+            for i, item in enumerate(items):
+                if i and RSS_PER_HOST_DELAY:
+                    await asyncio.sleep(RSS_PER_HOST_DELAY)
+                async with semaphore:
+                    group_results.append(await self._pull_rss_with_status(item))
+            return group_results
 
-        results = await asyncio.gather(*[_limited_pull(item) for item in rss_items])
+        host_groups: dict[str, list[RSSItem]] = defaultdict(list)
+        for item in rss_items:
+            host_groups[urlparse(item.url).netloc].append(item)
+        group_lists = list(host_groups.values())
+        grouped_results = await asyncio.gather(
+            *[_pull_host_group(items) for items in group_lists]
+        )
+        item_results = [
+            (item, result)
+            for items, results in zip(group_lists, grouped_results)
+            for item, result in zip(items, results)
+        ]
         now = datetime.now(timezone.utc).isoformat()
         # Process results sequentially (DB operations)
-        for rss_item, (new_torrents, error) in zip(rss_items, results):
+        for rss_item, (new_torrents, error) in item_results:
             # Update connection status
             rss_item.connection_status = "error" if error else "healthy"
             rss_item.last_checked_at = now

backend/src/test/test_rss_engine_new.py

@@ -369,3 +369,66 @@ class TestRefreshRssConcurrency:
             await rss_engine.refresh_rss(client)
 
         assert max_active <= 5
+
+
+# ---------------------------------------------------------------------------
+# refresh_rss per-host throttling (#1026)
+# ---------------------------------------------------------------------------
+
+
+class TestRefreshRssPerHostThrottle:
+    async def test_same_host_requests_never_overlap(self, rss_engine):
+        """Feeds on the same host are fetched serially; other hosts stay parallel."""
+        for i in range(3):
+            rss_engine.rss.add(
+                make_rss_item(url=f"https://nyaa.example/rss/{i}", name=f"nyaa{i}")
+            )
+        rss_engine.rss.add(make_rss_item(url="https://mikan.example/rss", name="mikan"))
+
+        from urllib.parse import urlparse
+
+        active: dict[str, int] = {}
+        max_active: dict[str, int] = {}
+
+        async def fake_pull(item):
+            host = urlparse(item.url).netloc
+            active[host] = active.get(host, 0) + 1
+            max_active[host] = max(max_active.get(host, 0), active[host])
+            # Give concurrently-scheduled pulls a chance to overlap.
+            await asyncio.sleep(0.01)
+            active[host] -= 1
+            return [], None
+
+        client = AsyncMock()
+        with (
+            patch.object(
+                RSSEngine,
+                "_pull_rss_with_status",
+                new_callable=lambda: AsyncMock(side_effect=fake_pull),
+            ),
+            patch("module.rss.engine.RSS_PER_HOST_DELAY", 0),
+        ):
+            await rss_engine.refresh_rss(client)
+
+        assert max_active["nyaa.example"] == 1
+        assert max_active["mikan.example"] == 1
+
+    async def test_all_feeds_still_processed(self, rss_engine):
+        """Grouping by host must not drop any feed's status update."""
+        rss_engine.rss.add(make_rss_item(url="https://a.example/rss", name="a"))
+        rss_engine.rss.add(make_rss_item(url="https://b.example/rss", name="b"))
+
+        client = AsyncMock()
+        with (
+            patch.object(
+                RSSEngine,
+                "_pull_rss_with_status",
+                new_callable=lambda: AsyncMock(return_value=([], None)),
+            ),
+            patch("module.rss.engine.RSS_PER_HOST_DELAY", 0),
+        ):
+            await rss_engine.refresh_rss(client)
+
+        for rss_id in (1, 2):
+            item = rss_engine.rss.search_id(rss_id)
+            assert item.connection_status == "healthy"