From 8b3eb22748dfd2fd97566f68509e9de7d6216403 Mon Sep 17 00:00:00 2001 From: Q Date: Sun, 18 Oct 2026 09:38:17 +0800 Subject: [PATCH] fix(search): bound long site cooldowns during paged searches (#7056) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 分页搜索在站点持续冷却时无法结束本轮收集，已缓存候选因此长期不能提交。 为来源持久化长冷却次数，仅在 cooldown 且等待至少 600 秒时累计；达到三次后仅停止当前轮次，下一轮重新搜索，成功取页后清零。 审查修复： - 冷却时间与当前时间采用同一秒精度，避免恰好 600 秒的等待漏计。 - 临时下载票据重取共享有限等待计数，保留末页来源的实际冷却时间；重启不清零，也不提前重复请求。超过上限后放弃无法恢复票据的候选，继续提交可用候选。 - 补充虚拟时钟、检查点恢复、同源多个候选回退、旧检查点兼容与成功后清零的回归覆盖及中文注释。 验证：本地相关 pytest 217 passed；changed-file Pylint、Ruff ratchet、结构复杂度、849 个源文件的 mypy 和宿主架构快照通过。合并前核对最终 PR head 的六片全量测试、覆盖率、Pylint 和架构门禁。 --- app/chain/search/scan.py | 61 +++++++- app/domain/search.py | 2 + tests/test_search_scan.py | 291 +++++++++++++++++++++++++++++++++++++- 3 files changed, 341 insertions(+), 13 deletions(-) diff --git a/app/chain/search/scan.py b/app/chain/search/scan.py index 581aaa691..2175a1c08 100644 --- a/app/chain/search/scan.py +++ b/app/chain/search/scan.py @@ -52,6 +52,11 @@ RESTART_AFTER_SECONDS = 36 * 3600 _CHECKPOINT_FORMAT = 3 # 重取临时下载票据所在原页连续失败达到该次数后放弃该候选，避免一个候选卡住整个订阅。 RECOVERY_ATTEMPTS = 3 +# 单个来源本轮因站点预算长冷却被推迟达到该次数后结算该来源：检查点随本轮结束删除，下一轮重新尝试， +# 既打破"全部来源结束才提交"的屏障死锁，也不永久拉黑偶尔不可用的站点。 +BUDGET_DEFERRAL_LIMIT = 3 +# 只累计不短于该秒数的冷却推迟；请求间隔与并发租约的秒级等待是健康站点噪声，不计入。 +BUDGET_DEFERRAL_SECONDS = 600.0 def search_targets(plan: MediaSearchPlan) -> set[str]: @@ -439,9 +444,7 @@ class SearchScan: result = self.sources.fetch(key, page, budget=self._budget()) observation = result.observation if result.retry_at: - self.retry[key] = result.retry_at - self.retry_reasons[key] = result.wait_reason or "busy" - source.error = observation.error + self._record_budget_wait(key, result) self.checkpoint() return False self.retry.pop(key, None) @@ -473,6 +476,7 @@ class SearchScan: logger.warning(f"{self.media.title_year} 搜索来源 {key} 连续 {UNMATCHED_PAGE_LIMIT} 页明确不匹配目标作品，" "本轮停止该来源，继续处理其他来源候选") source.failure_outcome = None + source.budget_deferrals = 0 self.retry.pop(key, None) self.retry_reasons.pop(key, None) self.live.update(contexts) @@ -485,6 +489,32 @@ class SearchScan: self.checkpoint() return bool(accepted) + def _record_budget_wait(self, key: str, result: ScanPage) -> None: + """ + 翻页与候选票据恢复共享来源长冷却计数，达到上限后仅停止本轮来源。 + + :param key: 当前页或待恢复候选所属的搜索来源 + :param result: 携带重试时间和等待原因的页面结果 + """ + source = self.collection.sources[key] + wait_reason = result.wait_reason or "busy" + self.retry[key] = cast(str, result.retry_at) + self.retry_reasons[key] = wait_reason + source.error = result.observation.error + if wait_reason == "cooldown": + # 预算时间保存到秒，当前时间使用相同精度，避免恰好 600 秒的冷却被小数部分排除。 + deferral_seconds = (datetime.fromisoformat(cast(str, result.retry_at)) + - datetime.now(timezone.utc).replace(microsecond=0)).total_seconds() + if deferral_seconds >= BUDGET_DEFERRAL_SECONDS: + source.budget_deferrals += 1 + if source.budget_deferrals >= BUDGET_DEFERRAL_LIMIT: + source.failure_outcome = "deferred" + source.failed = True + self.retry.pop(key, None) + self.retry_reasons.pop(key, None) + logger.warning(f"{self.media.title_year} 搜索来源 {key} 本轮 {source.budget_deferrals} 次等待站点预算未果，" + "跳过该来源以提交已收集候选，下一轮将重新搜索") + def _candidate_context(self, key: str) -> Optional[Context]: """ 取本进程的实时候选；重启后由站点模块按白名单快照恢复种子。 @@ -537,11 +567,23 @@ class SearchScan: """ record = self.candidates[key] source, page = record["source"], record["page"] + cursor = self.collection.sources[source] + if cursor.budget_deferrals >= BUDGET_DEFERRAL_LIMIT: + del self.candidates[key] + self.checkpoint() + return None + retry_at = self.retry.get(source) + if retry_at and retry_at > datetime.now(timezone.utc).isoformat(timespec="seconds"): + self._defer(waiting_source=source) result = self.sources.fetch(source, page, budget=self._budget()) if result.retry_at: - self.retry[source] = result.retry_at - self.retry_reasons[source] = result.wait_reason or "busy" - self._defer() + self._record_budget_wait(source, result) + if cursor.budget_deferrals < BUDGET_DEFERRAL_LIMIT: + self._defer(waiting_source=source) + # 临时票据尚未恢复时不能下载；放弃该候选，让仍可用的次优候选继续提交。 + del self.candidates[key] + self.checkpoint() + return None if result.observation.outcome != "success": failures = self.recovery_failures.get(key, 0) + 1 if failures < RECOVERY_ATTEMPTS: @@ -554,6 +596,7 @@ class SearchScan: f"{result.observation.error or '请求未成功'}") self.checkpoint() return None + cursor.budget_deferrals = 0 self.recovery_failures.pop(key, None) self.retry.pop(source, None) self.retry_reasons.pop(source, None) @@ -649,13 +692,17 @@ class SearchScan: self._add_sources(all_names=bool(self.owner.runtime_config.search_multiple_name)) self.checkpoint() - def _defer(self) -> None: + def _defer(self, *, waiting_source: Optional[str] = None) -> None: """ 按仍可搜索来源的有效冷却时间交还队列，避免过期检查点让任务立即重领。 + + :param waiting_source: 等待恢复候选票据的来源；即使已搜到末页也必须保留该来源的冷却 """ now = datetime.now(timezone.utc) now_at = now.isoformat(timespec="seconds") active_sources = self.collection.active_sources(full=self.full) + if waiting_source is not None and waiting_source not in active_sources: + active_sources.append(waiting_source) 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} diff --git a/app/domain/search.py b/app/domain/search.py index fb58e6b56..56e39bfd8 100644 --- a/app/domain/search.py +++ b/app/domain/search.py @@ -68,6 +68,7 @@ class SearchSourceCursor: :param page_limit: 当前来源页数上限；未指定时使用全局上限 :param matched_work: 作品匹配不局限于当前缺集；见过同作品的其他季集也不能按异常查询提前停止 :param unmatched_pages: 从未匹配目标作品时连续出现的明确不匹配页数 + :param budget_deferrals: 本轮因站点预算长冷却被推迟的次数；达到上限后本轮结算该来源，跨轮重新累计 """ next_page: int = 0 @@ -85,6 +86,7 @@ class SearchSourceCursor: page_limit: Optional[int] = None matched_work: bool = False unmatched_pages: int = 0 + budget_deferrals: int = 0 @property def limit_reached(self) -> bool: diff --git a/tests/test_search_scan.py b/tests/test_search_scan.py index e2f9f6110..4a46b9d52 100644 --- a/tests/test_search_scan.py +++ b/tests/test_search_scan.py @@ -7,12 +7,28 @@ from datetime import datetime, timedelta, timezone import pytest -from app.application.site.observation import report_site_search_outcome, report_site_search_page +from app.application.search.session import SearchSessionSnapshot, encode_search_state +from app.application.site.observation import SiteSearchObservation, report_site_search_outcome, report_site_search_page +from app.application.subscription.sitebudget import ( + SiteBudgetClaim, + SubscriptionSearchDeferred, + SubscriptionSiteBudget, + SubscriptionSiteBudgetUnavailable, +) +from app.chain.search import scan as scan_module 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 AUTOMATIC_REQUEST_INTERVAL, SearchScan, SearchSources +from app.chain.search.scan import ( + AUTOMATIC_REQUEST_INTERVAL, + BUDGET_DEFERRAL_LIMIT, + BUDGET_DEFERRAL_SECONDS, + ScanPage, + SearchScan, + SearchSources, +) from app.domain.context import MediaInfo, TorrentInfo +from app.runtime.stop import runtime_stop_state from app.schemas.mediaserver import NotExistMediaInfo from app.schemas.types import MediaSource, MediaType @@ -685,12 +701,12 @@ def test_language_or_capability_skip_keeps_existing_non_failure_semantics(owner, assert scan.collection.remaining == {"1:1"} -def test_opaque_ticket_recovery_submits_first_page_before_waiting_and_does_not_repeat_it_after_restart(owner, monkeypatch): +def test_opaque_ticket_recovery_submits_first_page_before_waiting_and_does_not_repeat_it_after_restart( + owner, monkeypatch, scan_clock, +): """ 不透明下载凭据恢复时先提交首页可用目标，重启后不重复提交。 """ - from datetime import datetime, timedelta, timezone - from app.application.search.session import SearchSessionSnapshot, encode_search_state from app.application.site.observation import SiteSearchObservation from app.application.subscription.sitebudget import SubscriptionSearchDeferred @@ -706,14 +722,20 @@ def test_opaque_ticket_recovery_submits_first_page_before_waiting_and_does_not_r assert initial.step(source) assert initial.step("plugin:Show") def restore(_method, **params): + """ + 恢复候选事实，缺失票据交给原页重取。 + """ item = TorrentInfo(**params["record"]) return item if item.enclosure or params.get("allow_missing") else None monkeypatch.setattr(owner, "run_module", restore) calls = [] cooling = {"active": True} def fetch(_sources, _key, page, **_params): + """ + 第二页等待一次冷却，其余页面直接返回票据。 + """ if page == 1 and cooling["active"]: - retry = (datetime.now(timezone.utc) + timedelta(seconds=10)).isoformat() + retry = (scan_clock["now"] + timedelta(seconds=10)).isoformat() return ScanPage([], SiteSearchObservation(), retry, "cooldown") calls.append(page) return ScanPage(data[page], SiteSearchObservation(True, "success", raw_count=1)) @@ -728,6 +750,7 @@ def test_opaque_ticket_recovery_submits_first_page_before_waiting_and_does_not_r payload = encode_search_state(restored.state()) assert "do-not-store" not in payload cooling["active"] = False + scan_clock["now"] = datetime.fromisoformat(restored.retry[source]) + timedelta(seconds=1) resumed = SearchScan(owner=owner, plan=plan([2]), full=True, snapshot=SearchSessionSnapshot("round", 1, payload)) resumed.run(lambda _items, goals: batches.append(goals) or goals) @@ -1185,3 +1208,259 @@ def test_recovered_candidate_is_filtered_again_before_submission(owner, monkeypa submitted = [] scan.choose({"1:1"}, lambda items, ready: submitted.extend(items) or set()) assert submitted == [] + + +class _StubSiteBudget(SubscriptionSiteBudget): + """ + 按站点模拟预算：故障站点始终长冷却推迟，健康站点立即放行。 + """ + + def __init__(self, deferred_site_id: int, retry_seconds: float) -> None: + """ + 保存故障站点和每次预算等待的秒数，不连接预算数据库。 + """ + super().__init__(repository=None, owner="test", cancelled=lambda: False, stop_state=runtime_stop_state) + self.deferred_site_id = deferred_site_id + self.retry_seconds = retry_seconds + + def acquire(self, site_id: int, *, minimum_interval: float = 0) -> SiteBudgetClaim: + """ + 故障站点返回冷却时间，其余站点直接取得测试租约。 + """ + if site_id == self.deferred_site_id: + retry_at = (scan_module.datetime.now(timezone.utc) + + timedelta(seconds=self.retry_seconds)).isoformat(timespec="seconds") + raise SubscriptionSiteBudgetUnavailable(site_id=site_id, retry_at=retry_at, wait_reason="cooldown") + return SiteBudgetClaim(site_id=site_id, acquired=True, retry_at="", + consecutive_failures=0, lease_token="stub") + + def record_request(self, site_id: int, candidate_count: int) -> None: + """ + 测试不落库，忽略请求指标。 + """ + return None + + def record_success(self, site_id: int) -> None: + """ + 测试不落库，忽略成功指标。 + """ + return None + + def finish(self, claim: SiteBudgetClaim, observation, *, deferral_seconds: float = 0) -> bool: + """ + 测试不落库，直接收口租约。 + """ + return True + + +@pytest.fixture +def scan_clock(monkeypatch): + """ + 使用可推进的 UTC 时钟验证冷却到期，不实际等待或清除重试状态。 + """ + clock = {"now": datetime.now(timezone.utc).replace(microsecond=500000)} + + class Clock(datetime): + """ + 返回测试控制的时刻，保留标准时间文本的解析行为。 + """ + + @classmethod + def now(cls, tz=None): + """ + 返回带指定时区的当前测试时间。 + """ + return clock["now"].astimezone(tz) + + monkeypatch.setattr(scan_module, "datetime", Clock) + return clock + + +def test_budget_deferred_dead_site_settles_and_releases_cached_candidates(owner, monkeypatch, scan_clock): + """ + 永久预算冷却的站点在本轮结算后放行"全部来源才提交"的屏障，缓存候选得以提交； + 结算只在本轮有效，下一轮该站点重新参与搜索。 + """ + monkeypatch.setattr(owner, "_sync_indexers", + lambda _sites: [{"id": 1, "name": "test", "parser": "mTorrent"}, + {"id": 2, "name": "dead", "parser": "mTorrent"}]) + site_pages = {(1, 0): [torrent(1)], (1, 1): []} + + def request(**params): + """ + 健康站点提供候选和末页证据，故障站点由预算拦截。 + """ + values = site_pages.get((params["site"]["id"], params["page"]), []) + report_site_search_outcome(attempted=True, outcome="success") + report_site_search_page(raw_count=len(values)) + return values + + monkeypatch.setattr(owner, "search_site_torrents", request) + monkeypatch.setattr(owner, "_subscription_site_budget", + _StubSiteBudget(deferred_site_id=2, retry_seconds=3600.0), raising=False) + + scan = SearchScan(owner=owner, plan=plan([1])) + submitted = [] + + def submit(contexts, ready): + """ + 记录屏障放行后真正提交的候选。 + """ + submitted.extend(contexts) + return ready + + # 按保存的重试时间推进时钟，并恢复检查点，模拟持久队列再次唤醒。 + for _ in range(5): + scan.run(submit) + if scan.ended: + break + if scan.retry: + scan_clock["now"] = datetime.fromisoformat(min(scan.retry.values())) + timedelta(seconds=1) + snapshot = SearchSessionSnapshot("round", 0, encode_search_state(scan.state())) + scan = SearchScan(owner=owner, plan=plan([1]), snapshot=snapshot) + + deferred_keys = [key for key in scan.collection.sources if key.startswith("2:")] + assert deferred_keys + assert all(scan.collection.sources[key].failed for key in deferred_keys) + assert all(scan.collection.sources[key].budget_deferrals >= BUDGET_DEFERRAL_LIMIT for key in deferred_keys) + assert scan.collection.settled == {"1:1"} + assert submitted + assert scan.ended is True + + +def test_short_budget_wait_does_not_settle_source(owner, monkeypatch): + """ + 秒级预算等待（请求间隔、并发租约）属于健康站点噪声，不得计数或结算来源。 + """ + monkeypatch.setattr(owner, "_subscription_site_budget", + _StubSiteBudget(deferred_site_id=1, retry_seconds=11.0), raising=False) + + scan = SearchScan(owner=owner, plan=plan([1])) + submitted = [] + + def submit(contexts, ready): + """ + 记录是否错误放行了仍在短等待的来源。 + """ + submitted.extend(contexts) + return ready + + for _ in range(5): + scan.run(submit) + scan.retry.clear() + + key = next(key for key in scan.collection.sources if key.startswith("1:")) + source = scan.collection.sources[key] + assert source.failed is False + assert source.budget_deferrals == 0 + assert submitted == [] + assert scan.ended is False + + +def test_exact_long_cooldown_counts_after_timestamp_truncation(owner, monkeypatch, scan_clock): + """ + 预算时间保存到秒，恰好达到长冷却门槛时不能因当前时间的小数部分漏计。 + """ + fixed_now = scan_clock["now"] + retry_at = (fixed_now + timedelta(seconds=BUDGET_DEFERRAL_SECONDS)).isoformat(timespec="seconds") + monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage( + [], SiteSearchObservation(), retry_at, "cooldown")) + scan = SearchScan(owner=owner, plan=plan([1])) + scan.step("1:Show") + assert scan.collection.sources["1:Show"].budget_deferrals == 1 + + +@pytest.mark.parametrize("ticket_count", [1, 2]) +def test_ticket_recovery_long_cooldown_is_bounded_across_checkpoints(owner, monkeypatch, scan_clock, ticket_count): + """ + 缓存候选重取票据也必须有限等待；恢复检查点后保留计数并继续提交可用候选。 + """ + scan, _ = _restored_ticket_scan(owner, monkeypatch, [torrent(1, version) + for version in range(1, ticket_count + 2)], []) + backup_key = next(key for key, saved in scan.candidates.items() + if saved["torrent"]["description"] == f"1-{ticket_count + 1}") + scan.candidates[backup_key]["torrent"]["enclosure"] = "https://site.example/download?id=backup" + requests = [] + + def fetch(_sources, _key, page, **_params): + """ + 模拟原页重取持续受站点长冷却阻塞。 + """ + requests.append(page) + retry_at = (scan_clock["now"] + timedelta(hours=1)).isoformat(timespec="seconds") + return ScanPage([], SiteSearchObservation(), retry_at, "cooldown") + + monkeypatch.setattr(SearchSources, "fetch", fetch) + submitted = [] + + def submit(items, goals): + """ + 记录实际可用候选并结算目标。 + """ + submitted.extend(item.torrent_info.description for item in items) + return goals + + for version in range(BUDGET_DEFERRAL_LIMIT - 1): + with pytest.raises(SubscriptionSearchDeferred) as error: + scan.run(submit) + assert error.value.wait_reason == "cooldown" + assert error.value.site_ids == (1,) + assert datetime.fromisoformat(error.value.retry_at) > scan_clock["now"] + timedelta(minutes=59) + snapshot = SearchSessionSnapshot("round", version, encode_search_state(scan.state())) + scan = SearchScan(owner=owner, plan=plan([1, 2]), full=True, snapshot=snapshot) + with pytest.raises(SubscriptionSearchDeferred): + scan.run(submit) + assert len(requests) == version + 1 + scan_clock["now"] = datetime.fromisoformat(error.value.retry_at) + timedelta(seconds=1) + scan.run(submit) + assert len(requests) == BUDGET_DEFERRAL_LIMIT + assert submitted == [f"1-{ticket_count + 1}"] + assert scan.ended + + +def test_successful_ticket_recovery_resets_budget_wait_count(owner, monkeypatch): + """ + 原页成功取回票据后清零等待计数，后续页不继承此前冷却。 + """ + scan, _ = _restored_ticket_scan(owner, monkeypatch, [torrent(1)], [torrent(1)]) + scan.collection.sources["1:Show"].budget_deferrals = BUDGET_DEFERRAL_LIMIT - 1 + scan.run(lambda _items, goals: goals) + assert scan.collection.sources["1:Show"].budget_deferrals == 0 + assert scan.collection.settled == {"1:1"} + + +@pytest.mark.parametrize("wait_reason, seconds, expected", [ + ("cooldown", 11, 1), ("busy", 3600, 1), ("cooldown", 3600, 2), +]) +def test_budget_wait_count_restores_and_success_resets(owner, monkeypatch, wait_reason, seconds, expected): + """ + 检查点保留长冷却累计，短等待与忙碌不加计数，成功取页后清零。 + """ + scan = SearchScan(owner=owner, plan=plan([1])) + scan.collection.sources["1:Show"].budget_deferrals = 1 + snapshot = SearchSessionSnapshot("round", 0, encode_search_state(scan.state())) + scan = SearchScan(owner=owner, plan=plan([1]), snapshot=snapshot) + retry_at = (datetime.now(timezone.utc) + timedelta(seconds=seconds)).isoformat(timespec="seconds") + monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage( + [], SiteSearchObservation(), retry_at, wait_reason)) + scan.step("1:Show") + assert scan.collection.sources["1:Show"].budget_deferrals == expected + pages(owner, monkeypatch, {0: []}) + monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage( + [], SiteSearchObservation(True, "success", raw_count=0))) + scan.step("1:Show") + assert scan.collection.sources["1:Show"].budget_deferrals == 0 + assert SearchScan(owner=owner, plan=plan([1])).collection.sources["1:Show"].budget_deferrals == 0 + + +def test_legacy_checkpoint_defaults_budget_wait_count_to_zero(owner): + """ + 没有预算计数的旧检查点仍可恢复，不会重启或提前结算来源。 + """ + state = SearchScan(owner=owner, plan=plan([1])).state() + for source in state["collection"]["sources"].values(): + source.pop("budget_deferrals") + snapshot = SearchSessionSnapshot("round", 0, encode_search_state(state)) + scan = SearchScan(owner=owner, plan=plan([1]), snapshot=snapshot) + assert not scan.restart + assert all(source.budget_deferrals == 0 and not source.failed for source in scan.collection.sources.values())