From 046d4758e6d3c8a8bc7bd56659e0cacd9d9a40cc Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 26 Oct 2026 08:50:04 +0800 Subject: [PATCH] fix(search): persist processed candidates across queue resumes Refs #7183 --- app/chain/search/scan.py | 15 +++--- docs/testing.md | 5 ++ tests/test_search_scan.py | 99 ++++++++++++++++++++++++++++++++++++--- 3 files changed, 107 insertions(+), 12 deletions(-) diff --git a/app/chain/search/scan.py b/app/chain/search/scan.py index 1038399cf..c6f75d0ca 100644 --- a/app/chain/search/scan.py +++ b/app/chain/search/scan.py @@ -240,8 +240,8 @@ class SearchScan: self.identity: dict[str, Any] = {} self.candidates: dict[str, Any] = {} self.live: dict[str, Context] = {} - # 本轮已处理（过滤淘汰、已提交或下载失败）的候选，只在内存中，重启后重新判断； - # 跨轮次的失败由原有下载失败冷却负责。 + # 本轮已处理（过滤淘汰、已提交或下载失败）的候选随检查点恢复， + # 避免重复重取原页抢占翻页预算；新的搜索轮次仍重新判断。 self.attempted: set[str] = set() self.disambiguation: DisambiguationCache = {} self.retry: dict[str, str] = {} @@ -310,7 +310,7 @@ class SearchScan: def _restore(self, data: dict[str, Any]) -> None: """ - 校验搜索合同并恢复候选、匹配前史和连续页进度，已提交目标仍受媒体身份约束。 + 校验搜索合同并恢复候选、处理记录和连续页进度，已提交目标仍受媒体身份约束。 """ # 搜索条件变化或进度太久未更新时重新搜索。 contract_matches = data.get("contract") == self._contract @@ -320,6 +320,7 @@ class SearchScan: or time.time() - data.get("saved_at", 0) > RESTART_AFTER_SECONDS) self.collection = restore_collection(data["collection"]) self.identity, self.candidates = data["identity"], data["candidates"] + self.attempted = set(data.get("attempted", [])) self.sources.queries = data["queries"] self.sources.keyword_index = data["keyword_index"] self.retry = data.get("retry", {}) @@ -346,11 +347,12 @@ class SearchScan: def _restore_target_scope(self) -> None: """ - 新增必要目标复用已识别元数据，补查头部后继续历史；已提交目标仍不重复。 + 新增必要目标复用已识别元数据并重新评估候选，补查头部后继续历史；已提交目标仍不重复。 """ fresh = search_targets(self.plan) - self.collection.targets if not fresh: return + self.attempted.clear() self.collection.targets.update(fresh) for record in self.identity.values(): if record["source"] and record["targets"] is not None: @@ -361,7 +363,7 @@ class SearchScan: def state(self) -> dict[str, Any]: """ - 页候选、关闭状态及下一页作为同一检查点保存。 + 页候选、处理记录、关闭状态及下一页作为同一检查点保存。 只持久化仍覆盖剩余缺集的候选及其识别结果；不相关资源的识别缓存只留在本进程， 恢复后再次遇到时重新识别，避免检查点随翻页无限增长。 @@ -372,6 +374,7 @@ class SearchScan: return {"format": _CHECKPOINT_FORMAT, "contract": self._contract, "media_identity": self._media_identity, "collection": collection_snapshot(self.collection), "identity": {key: self.identity[key] for key in candidates}, "candidates": candidates, + "attempted": sorted(self.attempted.intersection(candidates)), "queries": self.queries, "keywords": self.sources.keywords, "keyword_index": self.sources.keyword_index, "retry": self.retry, "retry_reasons": self.retry_reasons, @@ -681,7 +684,7 @@ class SearchScan: def choose(self, ready: set[str], submit: Callable[[list[Context], set[str]], set[str]]) -> None: """ - 合集覆盖目标全就绪才提交；剩余缓存先于共享游标兜底。 + 合集覆盖目标全就绪才提交；同轮已处理候选跨队列恢复不重试，剩余缓存先于共享游标兜底。 """ contexts, keys = self._contexts(ready) contexts = self._filter(contexts) diff --git a/docs/testing.md b/docs/testing.md index 24c969769..6de7cad14 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -168,6 +168,11 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)): `tests/test_search_scan.py` 与 `tests/test_subscribe_paged_search.py` 在离线边界验证恢复、过滤和 实际下载提交的先后顺序，同时保留普通补集按可用前缀提交的行为。 +同一轮分页搜索在站点冷却或时间片结束后恢复时，沿用已处理候选记录，避免无可用集数的 +候选反复重取原页并占用翻页预算。`tests/test_search_scan.py` 使用临时 SQLite、真实站点预算 +和虚拟时钟覆盖秒级冷却下的连续恢复，验证剩余来源继续翻页、末页候选处理后正常收尾， +且未收录缺集仍保留。新一轮搜索或缺集范围扩大时重新评估候选；旧检查点缺少该记录时兼容恢复。 + 全集洗版（`best_version=1`、`best_version_full=1`）的主动搜索始终使用洗版规则组， 包括尚无下载记录或仍有缺集的订阅；未通过洗版规则的整包不会回落到普通订阅规则组。 订阅显式指定的规则组仍优先。分集洗版中从未下载的剧集继续使用普通订阅规则。 diff --git a/tests/test_search_scan.py b/tests/test_search_scan.py index e9a6758d5..aa1c3bf88 100644 --- a/tests/test_search_scan.py +++ b/tests/test_search_scan.py @@ -490,9 +490,9 @@ def test_cross_site_barrier_keeps_late_opportunistic_candidate(owner, monkeypatc assert batches[0][2] == {"1:1"} -def test_rejected_candidate_is_not_retried_in_the_same_run_but_rejudged_after_resume(owner, monkeypatch): +def test_rejected_candidate_is_rejudged_only_in_a_new_round(owner, monkeypatch): """ - 同轮已拒绝候选不重复尝试，恢复搜索后允许重新判断。 + 同轮已拒绝候选跨检查点不重复尝试，新的搜索轮次仍按最新促销信息重新判断。 """ from app.application.search.session import SearchSessionSnapshot, encode_search_state @@ -510,13 +510,94 @@ def test_rejected_candidate_is_not_retried_in_the_same_run_but_rejudged_after_re # 本轮内已处理过的同一资源不再重复提交。 assert attempts == [1] assert scan.collection.remaining == {"1:1"} - # 已处理标记不入检查点：恢复后按最新信息（已免费）重新判断。 + # 恢复同一轮不能再次提交已拒绝资源，否则短冷却会让任务不断重取旧页。 resumed = SearchScan(owner=owner, plan=plan([1]), snapshot=SearchSessionSnapshot("task", 1, encode_search_state(scan.state()))) resumed.ended = False resumed.choose({"1:1"}, submit) + assert attempts == [1] + assert resumed.collection.remaining == {"1:1"} + pages(owner, monkeypatch, {0: [free]}) + fresh = SearchScan(owner=owner, plan=plan([1])) + fresh.run(submit) assert attempts == [1, 0] - assert not resumed.collection.remaining + assert not fresh.collection.remaining + + +@pytest.mark.parametrize("source_active", [False, True]) +def test_rejected_ticket_candidates_finish_across_short_budget_waits( + owner, monkeypatch, scan_clock, tmp_path, source_active, +): + """ + 同轮候选跨检查点不重复抢占预算，已结束来源能够收尾，兜底来源能够继续翻页。 + """ + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + from app.db.adapters import subscriptionsearch + from app.db.base import Base + from app.db.oper import subscriptionsearch as search_oper + + data = {0: [torrent(1, 1)], 1: [torrent(1, 2)]} + for items in data.values(): + items[0].enclosure = "https://site.example/download?ticket=do-not-store" + calls = pages(owner, monkeypatch, data) + initial = SearchScan(owner=owner, plan=plan([1, 2])) + source_key = "1:Show" + assert initial.step(source_key) + assert initial.step(source_key) + assert initial.step("plugin:Show") + cursor = initial.collection.sources[source_key] + cursor.closed.add("1:1") + cursor.exhausted = not source_active + assert initial.collection.unreleased() == {"1:2"} + calls.clear() + + engine = create_engine(f"sqlite:///{tmp_path / 'scan-budget.db'}") + Base.metadata.create_all(engine) + repository = subscriptionsearch.TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)) + monkeypatch.setattr(subscriptionsearch, "datetime", scan_module.datetime) + monkeypatch.setattr(search_oper, "datetime", scan_module.datetime) + budget = SubscriptionSiteBudget( + repository=repository, owner="scan", cancelled=lambda: False, + stop_state=runtime_stop_state, clock=lambda: scan_clock["now"], + ) + monkeypatch.setattr(owner, "_subscription_site_budget", budget, raising=False) + monkeypatch.setattr(owner, "run_module", lambda _method, **params: TorrentInfo(**params["record"])) + submitted = [] + + def reject(items, _goals): + """ + 模拟种子文件实际没有需要的集，保留缺集并记录每个已处理候选。 + """ + submitted.extend(item.torrent_info.description for item in items) + return set() + + state = initial.state() + try: + for version in range(6): + scan = SearchScan(owner=owner, plan=plan([1, 2]), + snapshot=SearchSessionSnapshot("round", version, encode_search_state(state))) + try: + scan.run(reject) + except SubscriptionSearchDeferred as error: + assert error.wait_reason == "cooldown" + delay = (datetime.fromisoformat(error.retry_at) - scan_clock["now"]).total_seconds() + assert 0 < delay <= AUTOMATIC_REQUEST_INTERVAL + 1 + scan_clock["now"] = datetime.fromisoformat(error.retry_at) + timedelta(seconds=1) + else: + if scan.ended: + break + scan_clock["now"] = datetime.fromisoformat(min(scan.retry.values())) + timedelta(seconds=1) + state = scan.state() + assert "do-not-store" not in encode_search_state(state) + assert scan.ended + assert submitted == ["1-1", "1-2"] + assert calls == ([0, 1, 2] if source_active else [0, 1]) + assert scan.collection.remaining == {"1:1", "1:2"} + assert scan.collection.sources[source_key].budget_deferrals == 0 + finally: + engine.dispose() def test_site_window_wait_keeps_cursor(owner, monkeypatch): @@ -844,9 +925,11 @@ def test_restored_scope_recomputes_targets_for_resources_without_season(owner, m scan.step(next(key for key in scan.queries if not key.startswith("plugin:"))) key = next(iter(scan.candidates)) assert scan.identity[key]["targets"] == ["1:1"] + scan.attempted.add(key) snapshot = SearchSessionSnapshot("scope", 1, encode_search_state(scan.state())) restored = SearchScan(owner=owner, plan=plan([1, 2]), snapshot=snapshot) assert restored.identity[key]["targets"] == ["1:1", "1:2"] + assert not restored.attempted def test_checkpoint_keeps_only_candidates_covering_remaining_episodes(owner, monkeypatch): @@ -858,9 +941,11 @@ def test_checkpoint_keeps_only_candidates_covering_remaining_episodes(owner, mon scan = SearchScan(owner=owner, plan=plan([20])) scan.step(next(key for key in scan.queries if not key.startswith("plugin:"))) assert len(scan.identity) == 30 + scan.attempted.update(scan.identity) state = scan.state() assert len(state["candidates"]) == 1 assert set(state["identity"]) == set(state["candidates"]) + assert set(state["attempted"]) == set(state["candidates"]) assert state["identity"][next(iter(state["candidates"]))]["targets"] == ["1:20"] @@ -1568,14 +1653,16 @@ def test_budget_wait_count_restores_and_success_resets(owner, monkeypatch, wait_ 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): +def test_legacy_checkpoint_defaults_budget_wait_count_and_processed_candidates(owner): """ - 没有预算计数的旧检查点仍可恢复，不会重启或提前结算来源。 + 缺少预算计数和候选处理记录的旧检查点仍可恢复，不会重启或提前结算来源。 """ state = SearchScan(owner=owner, plan=plan([1])).state() + state.pop("attempted") 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 not scan.attempted assert all(source.budget_deferrals == 0 and not source.failed for source in scan.collection.sources.values())