diff --git a/app/chain/search/scan.py b/app/chain/search/scan.py index 420027913..bb6c83bae 100644 --- a/app/chain/search/scan.py +++ b/app/chain/search/scan.py @@ -344,6 +344,8 @@ class SearchScan: source.error = observation.error self.checkpoint() return False + self.retry.pop(key, None) + self.retry_reasons.pop(key, None) if not observation.attempted or observation.outcome != "success": source.error = observation.error or "本页没有成功请求,保留页号" if source.head_refresh: @@ -524,10 +526,20 @@ class SearchScan: self.checkpoint() def _defer(self) -> None: + """按仍可搜索来源的有效冷却时间交还队列,避免过期检查点让任务立即重领。""" + now = datetime.now(timezone.utc) + now_at = now.isoformat(timespec="seconds") + active_sources = self.collection.active_sources(full=self.full) + active = set(active_sources) + self.retry = {key: retry_at for key, retry_at in self.retry.items() + if key in active and retry_at > now_at} + self.retry_reasons = {key: reason for key, reason in self.retry_reasons.items() if key in self.retry} self.checkpoint() - future = (datetime.now(timezone.utc) + timedelta(seconds=AUTOMATIC_REQUEST_INTERVAL)).isoformat(timespec="seconds") + # 检查点只保存到秒,向上留一秒可确保实际等待不短于请求间隔。 + future = (now + timedelta(seconds=AUTOMATIC_REQUEST_INTERVAL + 1)) + future = future.replace(microsecond=0).isoformat(timespec="seconds") retry_at = min(self.retry.values(), default=future) - site_ids = tuple(int(self.queries[key]["site"]) for key in self.collection.active_sources(full=self.full) + site_ids = tuple(int(self.queries[key]["site"]) for key in active_sources if self.queries[key]["site"] != "plugin") reason = "cooldown" if self.retry and all( self.retry_reasons.get(key) == "cooldown" for key in self.retry) else "busy" if self.retry else "slice" diff --git a/tests/test_search_scan.py b/tests/test_search_scan.py index 74c634b4b..2003491ef 100644 --- a/tests/test_search_scan.py +++ b/tests/test_search_scan.py @@ -1,6 +1,7 @@ """完整页驱动验证实际补集、过滤兜底和逐页识别成本。""" from dataclasses import replace +from datetime import datetime, timedelta, timezone import pytest @@ -8,7 +9,7 @@ from app.application.site.observation import report_site_search_outcome, report_ from app.chain.search.execution import MediaSearchPlan from app.chain.search.facade import SearchChain from app.chain.search.result import TorrentHelper -from app.chain.search.scan import SearchScan, SearchSources +from app.chain.search.scan import AUTOMATIC_REQUEST_INTERVAL, SearchScan, SearchSources from app.domain.context import MediaInfo, TorrentInfo from app.schemas.mediaserver import NotExistMediaInfo from app.schemas.types import MediaSource, MediaType @@ -331,6 +332,52 @@ def test_site_window_wait_keeps_cursor(owner, monkeypatch): assert scan.collection.sources[key].error == "站点窗口限流" +def test_defer_ignores_expired_retries_and_terminal_sources(owner, monkeypatch): + """失败或耗尽来源的检查点残留不能覆盖仍活跃来源的冷却时间。""" + from app.application.subscription.sitebudget import SubscriptionSearchDeferred + + monkeypatch.setattr(owner, "_sync_indexers", lambda _sites: [ + {"id": site_id, "name": f"test-{site_id}", "parser": "mTorrent"} for site_id in (1, 2, 3) + ]) + scan = SearchScan(owner=owner, plan=plan([1])) + failed_key, exhausted_key, active_key = (f"{site_id}:Show" for site_id in (1, 2, 3)) + scan.collection.sources[failed_key].failed = True + scan.collection.sources[exhausted_key].exhausted = True + now = datetime.now(timezone.utc) + expired = (now - timedelta(minutes=1)).isoformat(timespec="seconds") + active_retry = (now + timedelta(minutes=5)).isoformat(timespec="seconds") + scan.retry = {failed_key: expired, exhausted_key: expired, active_key: active_retry} + scan.retry_reasons = {failed_key: "cooldown", exhausted_key: "cooldown", active_key: "busy"} + + with pytest.raises(SubscriptionSearchDeferred) as error: + scan._defer() + + assert error.value.retry_at == active_retry + assert error.value.wait_reason == "busy" + assert scan.retry == {active_key: active_retry} + assert scan.retry_reasons == {active_key: "busy"} + + +def test_defer_uses_request_interval_when_all_retries_expired(owner): + """所有冷却都已过期时至少等待一个请求间隔再交还订阅队列。""" + from app.application.subscription.sitebudget import SubscriptionSearchDeferred + + scan = SearchScan(owner=owner, plan=plan([1])) + key = next(key for key in scan.collection.active_sources() if not key.startswith("plugin:")) + scan.retry[key] = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat(timespec="seconds") + scan.retry_reasons[key] = "cooldown" + before = datetime.now(timezone.utc) + + with pytest.raises(SubscriptionSearchDeferred) as error: + scan._defer() + + retry_at = datetime.fromisoformat(error.value.retry_at) + assert retry_at >= before + timedelta(seconds=AUTOMATIC_REQUEST_INTERVAL) + assert error.value.wait_reason == "slice" + assert not scan.retry + assert not scan.retry_reasons + + def test_all_sources_failing_is_not_reported_as_exhausted(owner, monkeypatch): from app.application.subscription.execution import SubscriptionSiteSearchFailed def request(**_params):