diff --git a/backend/src/module/rss/engine.py b/backend/src/module/rss/engine.py index e126b5ef..9f2b6d13 100644 --- a/backend/src/module/rss/engine.py +++ b/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 diff --git a/backend/src/test/test_rss_engine_new.py b/backend/src/test/test_rss_engine_new.py index 14ea73c2..3512c756 100644 --- a/backend/src/test/test_rss_engine_new.py +++ b/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"