diff --git a/app/application/site/throttle.py b/app/application/site/throttle.py index b1171e902..e3d93f614 100644 --- a/app/application/site/throttle.py +++ b/app/application/site/throttle.py @@ -6,6 +6,7 @@ import asyncio import math import time from collections.abc import Mapping +from datetime import datetime, timedelta from typing import Optional from app.application.site.protocol import site_domain @@ -15,6 +16,9 @@ from app.runtime.rate import RequestRateLimiter from app.schemas.exception import RateLimitExceededException _site_requests = RequestRateLimiter() +# 本地额度最多原地等待的秒数;更长的间隔或计数窗口交还调用方延后, +# 避免订阅搜索消费者等串行执行者在单个请求上休眠到下一个额度。 +_MAX_LOCAL_WAIT_SECONDS = 60.0 def site_request_config(url: str, site_id: Optional[int] = None) -> dict[str, object]: @@ -56,7 +60,7 @@ def _positive_number(value: object) -> float: def _request_delay(site: Mapping[str, object]) -> float: """ - 原子申请一次站点请求,远端冷却期间直接返回原始原因。 + 原子申请一次站点请求,远端冷却或本地等待超过上限时直接抛出原因,不消耗额度。 """ interval = _positive_number(site.get("limit_seconds")) delay, reason = _site_requests.acquire( @@ -65,15 +69,29 @@ def _request_delay(site: Mapping[str, object]) -> float: ) if reason: raise RateLimitExceededException(reason) + if delay > _MAX_LOCAL_WAIT_SECONDS: + raise RateLimitExceededException(_local_limit_reason(delay)) return delay +def _local_limit_reason(delay: float) -> str: + """ + 描述超过原地等待上限的本地额度,保留限流语义和剩余秒数供下载冷却识别。 + + :param delay: 距离下一次本地许可的秒数 + :return: 给用户看的限流说明 + """ + seconds = math.ceil(delay) + ready_at = (datetime.now() + timedelta(seconds=seconds)).strftime("%H:%M") + return f"站点限流:站点设置的访问间隔或次数未到,还需等待 {seconds} 秒(约 {ready_at} 后可再访问)" + + def wait_for_site_request(site: Mapping[str, object]) -> None: """ 在同步请求前等待配置的间隔和次数窗口,唤醒后重新检查许可。 :param site: 本次请求所属站点的流控配置 - :raises RateLimitExceededException: 站点仍处于远端明确要求的冷却期 + :raises RateLimitExceededException: 站点仍处于远端明确要求的冷却期,或本地额度需等待超过上限 """ while delay := _request_delay(site): time.sleep(delay) @@ -84,7 +102,7 @@ async def async_wait_for_site_request(site: Mapping[str, object]) -> None: 异步等待同一站点额度,取消等待不会提前消耗请求次数。 :param site: 本次请求所属站点的流控配置 - :raises RateLimitExceededException: 站点仍处于远端明确要求的冷却期 + :raises RateLimitExceededException: 站点仍处于远端明确要求的冷却期,或本地额度需等待超过上限 """ while delay := _request_delay(site): await asyncio.sleep(delay) diff --git a/tests/test_site_request_throttle.py b/tests/test_site_request_throttle.py index a8eb20830..16f5cda97 100644 --- a/tests/test_site_request_throttle.py +++ b/tests/test_site_request_throttle.py @@ -19,6 +19,7 @@ from app.adapters.cache.backends import FileBackend from app.application.site.observation import capture_site_search_observation from app.chain.download import DownloadChain from app.chain.search import provider +from app.domain import torrent as torrent_rules from app.domain.context import TorrentInfo from app.modules.indexer import IndexerModule from app.runtime.rate import RequestRateLimiter @@ -321,3 +322,67 @@ def test_remote_cooldown_cannot_be_shortened(boundary): assert throttle.site_request_cooldown(site) == 590 with pytest.raises(RateLimitExceededException, match="600 seconds"): throttle.wait_for_site_request(site) + + +def test_long_local_interval_defers_instead_of_sleeping(boundary): + """ + 超过原地等待上限的访问间隔立即返回限流原因,不休眠、不占用额度,到期后恢复。 + """ + site = {"id": 42, "limit_seconds": 3600} + throttle.wait_for_site_request(site) + boundary.clock.now += 10 + + with pytest.raises(RateLimitExceededException, match="还需等待 3590 秒") as error: + throttle.wait_for_site_request(site) + + assert torrent_rules.rate_limit_cooldown(str(error.value)) == 3600 + assert boundary.clock.waits == [] + boundary.clock.now += 3590 + throttle.wait_for_site_request(site) + assert boundary.clock.waits == [] + + +def test_long_count_window_defers_and_short_wait_still_sleeps(boundary): + """ + 计数窗口剩余时间超过上限时延后,未超过上限的等待仍按原语义原地完成。 + """ + site = {"id": 42, "limit_interval": 3600, "limit_count": 2, "limit_seconds": 60} + throttle.wait_for_site_request(site) + throttle.wait_for_site_request(site) + + assert boundary.clock.waits == [60] + with pytest.raises(RateLimitExceededException, match="还需等待 3540 秒"): + throttle.wait_for_site_request(site) + assert boundary.clock.waits == [60] + + +def test_torrent_download_after_search_returns_without_waiting(boundary): + """ + 搜索刚用掉一小时一次的额度后,同站取种子立即失败并按限流冷却,不阻塞调用线程。 + """ + boundary.query.get_sync.side_effect = lambda site_id: Site(id=site_id, limit_seconds=3600) + provider._search_site_page( + boundary.owner, site={"id": 42, "limit_seconds": 3600}, keyword="Series", media_type=None, page=0, + ) + errors = [] + + content = boundary.chain.download_torrent(TorrentInfo(site=42, enclosure=URL, title="Series"), + on_failure=errors.append) + + assert content == (None, "", []) + assert errors and errors[0].startswith("站点限流:站点设置的访问间隔或次数未到") + assert DownloadChain._download_failure_ttl(errors[0]) == 3600 + boundary.port.request.assert_not_called() + assert boundary.clock.waits == [] + + +@pytest.mark.asyncio +async def test_async_long_local_interval_defers(boundary, monkeypatch): + """ + 异步搜索同样不为超过上限的本地额度长时间挂起。 + """ + monkeypatch.setattr(throttle.asyncio, "sleep", AsyncMock(side_effect=AssertionError("不应等待"))) + site = {"id": 42, "limit_seconds": 3600} + await throttle.async_wait_for_site_request(site) + with pytest.raises(RateLimitExceededException, match="还需等待 3600 秒"): + await throttle.async_wait_for_site_request(site)