mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-15 21:01:10 +08:00
fix(site): defer long local throttle waits instead of sleeping (#7399)
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user