fix(search): persist processed candidates across queue resumes

Refs #7183

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)

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`）的主动搜索始终使用洗版规则组，
 包括尚无下载记录或仍有缺集的订阅；未通过洗版规则的整包不会回落到普通订阅规则组。
 订阅显式指定的规则组仍优先。分集洗版中从未下载的剧集继续使用普通订阅规则。

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())