From 8255df4df43179d5dc366a1e498f228c341ff847 Mon Sep 17 00:00:00 2001
From: jxxghp <jxxghp@gmail.com>
Date: Tue, 8 Sep 2026 08:58:35 +0800
Subject: [PATCH] fix(subscribe): remove cumulative search delays and resume
 pending sites

---
 app/application/subscription/execution.py     | 10 ++--
 app/application/subscription/sitebudget.py    | 11 ++--
 app/chain/search/provider.py                  | 10 +++-
 app/chain/subscribe/search.py                 | 33 +++++++++---
 app/chain/subscribe/searchtask.py             |  1 +
 app/db/adapters/subscriptionsearch.py         |  5 +-
 app/db/models/subscriptionsearch.py           |  4 +-
 app/db/oper/subscriptionsearch.py             | 12 ++++-
 app/scheduler/catalog.py                      | 10 ++--
 app/startup/initializers/scheduler.py         |  2 +-
 database/versions/a9c3e7f1b5d8_3_0_31.py      | 48 +++++++++++++++++
 docs/mcp-api.md                               |  7 +++
 skills/moviepilot-api/SKILL.md                |  4 ++
 .../architecture/dependency-baseline.json     |  6 ++-
 tests/test_scheduler_lifecycle.py             | 43 ++++++++++++++++
 ...st_subscription_search_cursor_migration.py | 43 ++++++++++++++++
 tests/test_subscription_search_governance.py  | 29 ++++-------
 tests/test_subscription_search_queue.py       | 51 +++++++++++++++++++
 tests/test_subscription_site_budget.py        | 45 ++++++++++++++++
 19 files changed, 330 insertions(+), 44 deletions(-)
 create mode 100644 database/versions/a9c3e7f1b5d8_3_0_31.py
 create mode 100644 tests/test_subscription_search_cursor_migration.py

diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py
index 780718200..a401999f7 100644
--- a/app/application/subscription/execution.py
+++ b/app/application/subscription/execution.py
@@ -119,7 +119,8 @@ def raise_subscription_site_budget_deferral(
         return
     retry_at = min(deferrals, key=lambda item: item.retry_at).retry_at
     site_ids = tuple(dict.fromkeys(item.site_id for item in deferrals))
-    raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=site_ids)
+    wait_reason = "cooldown" if all(item.wait_reason == "cooldown" for item in deferrals) else "busy"
+    raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=site_ids, wait_reason=wait_reason)
 
 
 def handle_subscription_search_deferred(
@@ -135,7 +136,8 @@ def handle_subscription_search_deferred(
         lease_token=lease_token,
         available_at=deferred.retry_at,
         phase="waiting_site_budget",
-        message="站点暂时忙，系统会自动继续搜索",
+        message=str(deferred),
+        pending_site_ids=deferred.site_ids,
     )
     if requeued:
         record("requeued", "site_budget_deferred")
@@ -184,6 +186,7 @@ class SearchTaskSnapshot:
     finished_at: Optional[str] = None
     last_error: Optional[str] = None
     current_site_id: Optional[int] = None
+    pending_site_ids: Optional[tuple[int, ...]] = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -265,8 +268,9 @@ class SubscriptionSearchRepository(Protocol):
         available_at: str,
         phase: str = "waiting_site_budget",
         message: Optional[str] = None,
+        pending_site_ids: Optional[tuple[int, ...]] = None,
     ) -> bool:
-        """把临时不可执行任务退回队列，并保留用户可理解的等待原因。"""
+        """延后任务并保存等待原因与待搜站点；省略站点时保留已有游标。"""
         ...
 
     def is_cancel_requested(self, task_id: str) -> bool:
diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py
index b1ac30dad..21f1a2c8c 100644
--- a/app/application/subscription/sitebudget.py
+++ b/app/application/subscription/sitebudget.py
@@ -21,14 +21,17 @@ class SubscriptionSiteBudgetDeferral:
 
     site_id: int
     retry_at: str
+    wait_reason: Optional[str] = None
 
 
 class SubscriptionSearchDeferred(RuntimeError):
     """表示订阅搜索未失败，而是应在站点预算可用后重新入队。"""
 
-    def __init__(self, *, retry_at: str, site_ids: tuple[int, ...]) -> None:
+    def __init__(
+        self, *, retry_at: str, site_ids: tuple[int, ...], wait_reason: Optional[str] = None,
+    ) -> None:
         """保存队列恢复所需的时间和站点，避免把临时等待写成错误。"""
-        super().__init__("站点暂时忙，系统会自动继续搜索")
+        super().__init__("站点冷却中" if wait_reason == "cooldown" else "等待站点")
         self.retry_at = retry_at
         self.site_ids = site_ids
 
@@ -176,8 +179,10 @@ class SubscriptionSiteBudget:
         clock: Callable[[], datetime] = _utc_now,
         phase_changed: Optional[Callable[[str, Optional[int]], None]] = None,
         metrics: Optional[SubscriptionSiteBudgetMetrics] = None,
+        pending_site_ids: Optional[tuple[int, ...]] = None,
     ) -> None:
-        """保存持久化端口及可注入的时钟和阶段回调。"""
+        """保存站点预算及恢复范围；首次搜索的 pending_site_ids 为 None。"""
+        self.pending_site_ids = pending_site_ids
         self._repository = repository
         self._owner = owner
         self._cancelled = cancelled
diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py
index 8506a7b60..0d542f615 100644
--- a/app/chain/search/provider.py
+++ b/app/chain/search/provider.py
@@ -262,6 +262,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
                 SubscriptionSiteBudgetDeferral(
                     site_id=error.site_id,
                     retry_at=error.retry_at,
+                    wait_reason=error.wait_reason,
                 )
             )
             logger.debug(str(error))
@@ -429,11 +430,16 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
         area: Optional[str] = "title",
         mtype: Optional[MediaType] = None,
     ) -> Optional[List[TorrentInfo]]:
-        """通过共享线程 owner 按站点顺序翻页并汇总同步 provider 结果。"""
+        """共享线程按站点翻页；恢复搜索仅查询待完成且仍启用的站点。"""
         indexer_sites = self._sync_indexers(sites)
         media_type = self._torrent_type(mediainfo, mtype)
         search_keyword = self._torrent_keyword(keyword, mediainfo, area)
-        plugin_results = self.search_plugin_torrents(
+        budget = getattr(self, "_subscription_site_budget", None)
+        pending_site_ids = budget.pending_site_ids if isinstance(budget, SubscriptionSiteBudget) else None
+        # 已完成的站点和插件源不再重复查询；仍按当前启用配置过滤被移除的站点。
+        if pending_site_ids is not None:
+            indexer_sites = [site for site in indexer_sites if site.get("id") in pending_site_ids]
+        plugin_results = [] if pending_site_ids is not None else self.search_plugin_torrents(
             keyword=search_keyword,
             mtype=media_type,
             page=page,
diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py
index 2c405b2a2..82fab8ea1 100644
--- a/app/chain/subscribe/search.py
+++ b/app/chain/subscribe/search.py
@@ -1,5 +1,6 @@
 """订阅主动搜索编排"""
 
+import asyncio
 import random
 import time
 from datetime import datetime, timedelta, timezone
@@ -46,8 +47,10 @@ from app.domain.context import (
     MusicInfo,
 )
 from app.domain.meta.metabase import MetaBase
+from app.runtime.execution import await_task_to_terminal
 from app.runtime.log import logger
 from app.runtime.stop import runtime_stop_state
+from app.runtime.tasks import get_task_registry
 from app.schemas.types import (
     MediaType,
     SystemConfigKey,
@@ -92,19 +95,14 @@ def _search_task_available_at(
     *,
     now: Optional[datetime] = None,
 ) -> dict[int, str]:
-    """把兜底搜索的随机节奏持久化为逐订阅到期时间。"""
+    """整批仅抖动一次；请求节奏由站点限流控制，不随订阅数量累计空等。"""
     ordered_ids = tuple(dict.fromkeys(subscription_ids))
     if not ordered_ids:
         return {}
     cursor = now or datetime.now(timezone.utc)
     if source == "fallback":
         cursor += timedelta(seconds=random.randint(0, 60))
-    available_at: dict[int, str] = {}
-    for position, subscription_id in enumerate(ordered_ids):
-        if source == "fallback" and position:
-            cursor += timedelta(seconds=random.randint(60, 300))
-        available_at[subscription_id] = cursor.isoformat(timespec="seconds")
-    return available_at
+    return dict.fromkeys(ordered_ids, cursor.isoformat(timespec="seconds"))
 
 
 class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
@@ -518,6 +516,27 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator):
         )
         return runner.execute()
 
+    async def async_resume_search_queue(
+        self,
+        progress_callback: Optional[Callable[..., None]] = None,
+        limit: int = 50,
+        manual_sids: Optional[tuple[int, ...]] = None,
+    ) -> None:
+        """托管恢复协程及手工反馈，等待同步消费者真实结束后才释放运行所有权。"""
+        task = get_task_registry().create_sync(
+            self.resume_search_queue,
+            owner="subscription.search_queue",
+            progress_callback=progress_callback,
+            limit=limit,
+            manual_sids=manual_sids,
+        )
+        try:
+            await asyncio.shield(task)
+        except asyncio.CancelledError:
+            # 同步线程不可强制取消；真实收尾前不能让下一次轮询取得消费者所有权。
+            await await_task_to_terminal(task)
+            raise
+
     def resume_search_queue(
         self,
         progress_callback: Optional[Callable[..., None]] = None,
diff --git a/app/chain/subscribe/searchtask.py b/app/chain/subscribe/searchtask.py
index 3a440208a..5175dc4b3 100644
--- a/app/chain/subscribe/searchtask.py
+++ b/app/chain/subscribe/searchtask.py
@@ -209,6 +209,7 @@ class SubscriptionSearchTaskRunner:
                     stop_state=self.stop_state,
                     phase_changed=phase_changed,
                     metrics=self.summary.site_metrics,
+                    pending_site_ids=self.task.pending_site_ids,
                 )
             )
             current = self.process_subscription(
diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py
index d30b63019..42d846fcf 100644
--- a/app/db/adapters/subscriptionsearch.py
+++ b/app/db/adapters/subscriptionsearch.py
@@ -67,6 +67,7 @@ def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot:
         finished_at=record.finished_at,
         last_error=record.last_error,
         current_site_id=record.current_site_id,
+        pending_site_ids=tuple(record.pending_site_ids) if record.pending_site_ids is not None else None,
     )
 
 
@@ -248,8 +249,9 @@ class TransactionalSubscriptionSearchRepository:
         available_at: str,
         phase: str = "waiting_site_budget",
         message: Optional[str] = None,
+        pending_site_ids: Optional[tuple[int, ...]] = None,
     ) -> bool:
-        """按指定时间和可见原因重新排队任务。"""
+        """重新排队并保存待搜站点；未提供站点时保留原游标。"""
         return self._write(
             lambda repository: repository.defer_task(
                 task_id=task_id,
@@ -257,6 +259,7 @@ class TransactionalSubscriptionSearchRepository:
                 available_at=available_at,
                 phase=phase,
                 message=message,
+                pending_site_ids=pending_site_ids,
             )
         )
 
diff --git a/app/db/models/subscriptionsearch.py b/app/db/models/subscriptionsearch.py
index c5355a30b..49d38f15f 100644
--- a/app/db/models/subscriptionsearch.py
+++ b/app/db/models/subscriptionsearch.py
@@ -2,7 +2,7 @@
 
 from typing import Optional
 
-from sqlalchemy import Index, Integer, String, Text, UniqueConstraint
+from sqlalchemy import JSON, Index, Integer, String, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base, get_id_column
@@ -48,6 +48,8 @@ class SubscriptionSearchTask(Base):
     state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
     phase: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
     current_site_id: Mapped[Optional[int]] = mapped_column(Integer)
+    # None 表示首次完整搜索；重试只保留尚未完成的站点，跨重启不重复请求成功站点。
+    pending_site_ids: Mapped[Optional[list[int]]] = mapped_column(JSON(none_as_null=True))
     attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     cancel_requested: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     lease_owner: Mapped[Optional[str]] = mapped_column(String(128))
diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py
index 726004a40..2b48426b1 100644
--- a/app/db/oper/subscriptionsearch.py
+++ b/app/db/oper/subscriptionsearch.py
@@ -108,6 +108,11 @@ class SubscriptionSearchOper(DbOper):
                             (promote_queued_task, None),
                             else_=SubscriptionSearchTask.last_error,
                         ),
+                        # 用户重新指定搜索时按当前站点配置重搜；自动周期合并保留恢复游标。
+                        pending_site_ids=case(
+                            (and_(SubscriptionSearchTask.state == "queued", source in {"manual", "targeted"}), None),
+                            else_=SubscriptionSearchTask.pending_site_ids,
+                        ),
                         available_at=case(
                             (
                                 or_(
@@ -138,7 +143,7 @@ class SubscriptionSearchOper(DbOper):
         return batch, created, coalesced, tuple(dict.fromkeys(active_batch_ids))
 
     def claim_next(self, *, owner: str, lease_seconds: int) -> Optional[SubscriptionSearchTask]:
-        """使用 CAS 认领最高优先级任务，过期 running 任务可被恢复。"""
+        """CAS 认领任务并清除旧等待提示，过期 running 任务保留站点游标恢复。"""
         if not isinstance(self._db, Session):
             raise RuntimeError("订阅搜索认领需要调用方提供同步 Session")
         now = utc_now_text()
@@ -209,6 +214,7 @@ class SubscriptionSearchOper(DbOper):
                 .values(
                     state="running",
                     phase="matching",
+                    last_error=None,
                     current_site_id=None,
                     lease_owner=owner,
                     lease_token=lease_token,
@@ -403,8 +409,9 @@ class SubscriptionSearchOper(DbOper):
         available_at: str,
         phase: str,
         message: Optional[str],
+        pending_site_ids: Optional[tuple[int, ...]] = None,
     ) -> bool:
-        """释放当前租约并在指定时间后按可见原因恢复同一任务。"""
+        """释放租约并保存重试站点；普通准入等待保留已有站点游标。"""
         if not isinstance(self._db, Session):
             raise RuntimeError("订阅搜索延后需要调用方提供同步 Session")
         task = self._db.execute(
@@ -443,6 +450,7 @@ class SubscriptionSearchOper(DbOper):
                 updated_at=now,
                 finished_at=None,
                 last_error=message,
+                pending_site_ids=list(pending_site_ids) if pending_site_ids is not None else task.pending_site_ids,
             ),
             execution_options={"synchronize_session": False},
         )
diff --git a/app/scheduler/catalog.py b/app/scheduler/catalog.py
index 8fab04323..f8e358962 100644
--- a/app/scheduler/catalog.py
+++ b/app/scheduler/catalog.py
@@ -150,16 +150,20 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
             replace_existing=True,
         )
 
+    def _poll_subscription_search_queue(self) -> None:
+        """仅在消费者空闲时唤醒托管协程，轮询不等待搜索也不重复报告重入。"""
+        if not self._is_job_active("subscribe_search_queue"):
+            self.start("subscribe_search_queue")
+
     def _register_subscription_search_queue_job(self, config: SchedulerRuntimeConfig) -> None:
-        """注册短周期持久搜索队列恢复任务。"""
+        """注册轻量轮询；搜索协程的真实生命周期由 Scheduler 持有。"""
         self._scheduler.add_job(
-            self.start,
+            self._poll_subscription_search_queue,
             "interval",
             id="subscribe_search_queue",
             name="恢复订阅搜索队列",
             seconds=10,
             next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=5),
-            kwargs={"job_id": "subscribe_search_queue"},
         )
 
     def _initialize_catalog(self, config: SchedulerRuntimeConfig) -> None:
diff --git a/app/startup/initializers/scheduler.py b/app/startup/initializers/scheduler.py
index 4e2e952c8..59e46b8de 100644
--- a/app/startup/initializers/scheduler.py
+++ b/app/startup/initializers/scheduler.py
@@ -61,7 +61,7 @@ def configure_scheduler_services() -> None:
             sync_mediaserver=mediaserver_chain.sync,
             check_subscribe=subscribe_chain.check_and_reconcile,
             search_subscribe=subscribe_chain.search,
-            resume_subscribe_search=subscribe_chain.resume_search_queue,
+            resume_subscribe_search=subscribe_chain.async_resume_search_queue,
             refresh_subscribe=subscribe_chain.refresh,
             follow_subscribe=subscribe_chain.follow,
             process_transfer=transfer_chain.process,
diff --git a/database/versions/a9c3e7f1b5d8_3_0_31.py b/database/versions/a9c3e7f1b5d8_3_0_31.py
new file mode 100644
index 000000000..098c28a2c
--- /dev/null
+++ b/database/versions/a9c3e7f1b5d8_3_0_31.py
@@ -0,0 +1,48 @@
+"""3.0.31 保存订阅搜索重试站点并解除存量批次的累计空等。"""
+
+# Alembic 的 op 是运行期代理。
+# pylint: disable=no-member
+
+from datetime import datetime, timezone
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "a9c3e7f1b5d8"
+down_revision = "f8a2c6e9b1d4"
+branch_labels = None
+depends_on = None
+
+_TABLE = "subscriptionsearchtask"
+
+
+def upgrade() -> None:
+    """增加可恢复站点游标，只提前尚未执行的自动任务，不绕过冷却和用户停止。"""
+    inspector = sa.inspect(op.get_bind())
+    if _TABLE not in inspector.get_table_names():
+        return
+    columns = {column["name"] for column in inspector.get_columns(_TABLE)}
+    if "pending_site_ids" in columns:
+        return
+    op.add_column(_TABLE, sa.Column("pending_site_ids", sa.JSON(none_as_null=True), nullable=True))
+    tasks = sa.table(
+        _TABLE,
+        sa.column("source"), sa.column("state"), sa.column("phase"),
+        sa.column("attempt_count"), sa.column("cancel_requested"), sa.column("available_at"),
+    )
+    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    op.execute(
+        tasks.update().where(
+            tasks.c.source == "fallback", tasks.c.state == "queued", tasks.c.phase == "queued",
+            tasks.c.attempt_count == 0, tasks.c.cancel_requested == 0, tasks.c.available_at > now,
+        ).values(available_at=now)
+    )
+
+
+def downgrade() -> None:
+    """移除站点游标；恢复旧版本后未完成任务仍可完整重试。"""
+    inspector = sa.inspect(op.get_bind())
+    if _TABLE in inspector.get_table_names():
+        columns = {column["name"] for column in inspector.get_columns(_TABLE)}
+        if "pending_site_ids" in columns:
+            op.drop_column(_TABLE, "pending_site_ids")
diff --git a/docs/mcp-api.md b/docs/mcp-api.md
index 4102a9714..7b07a866e 100644
--- a/docs/mcp-api.md
+++ b/docs/mcp-api.md
@@ -433,6 +433,13 @@ TMDB 缓存查询响应的 `data` 包含 `count`、`recognized`、`unrecognized`
 旧订阅尚无搜索时间时，以添加时间计算到期。新订阅首次搜索、手动搜索和 RSS 刷新不受周期过滤影响；
 手动主动搜索会更新最近搜索时间。电影、电视剧和音乐均支持独立周期。
 
+自动批次只在启动时随机错峰 0–60 秒，不再按订阅数量累加分钟级等待；同站点访问间隔、唯一在途租约和错误冷却继续生效。
+站点暂不可用时，任务保存未完成站点并按 `next_run_at` 恢复，不重复查询已完成站点或插件源；队列和站点游标可跨重启恢复。
+`waiting_site_budget` 表示可恢复等待，`error` 中的“等待站点”或“站点冷却中”是原因提示，不表示搜索失败；重新执行时清除旧提示。
+卡片应按 `state` / `phase` 展示简短标签，原因放入详情提示。恢复调度每 10 秒检查空闲消费者，长搜索由调度器持续托管。
+实际吞吐仍受站点访问限制和网络响应时间影响，配置周期不是全部站点必须完成的截止时间。
+
+
 ### 插件补充接口
 
 **GET** `/api/v1/plugin/history/{plugin_id}`
diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md
index 3b60bb36f..57f7f02e4 100644
--- a/skills/moviepilot-api/SKILL.md
+++ b/skills/moviepilot-api/SKILL.md
@@ -1247,6 +1247,10 @@ Purpose: Reset one accessible subscription so it can be processed again.
 - `query`: none
 - `body`: none
 
+Subscription search uses a durable queue. Automatic batches share a single 0–60 second startup jitter; site rate limits and cooldowns still apply.
+Deferred attempts resume only pending sites, including after restart. Treat `waiting_site_budget` as recoverable waiting and use `next_run_at` for the next attempt;
+its `error` text is a waiting reason and is cleared when execution resumes. Do not interpret the configured search interval as a completion deadline.
+
 ### `subscription.search`
 `POST /api/v1/subscribe/search/{subscribe_id}`; policy effect: `external_side_effect`.
 Purpose: Run an immediate search for one existing subscription.
diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json
index cd01e7ee8..12d9ae2d0 100644
--- a/tests/fixtures/architecture/dependency-baseline.json
+++ b/tests/fixtures/architecture/dependency-baseline.json
@@ -1074,8 +1074,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8313,
-  "edge_sha256": "93fe54d658dcf90129d84cb2f61c2db39630a5ebfc7538fe589912fc540fda21",
+  "edge_count": 8315,
+  "edge_sha256": "798e408eb7130ed507124b1b775aee3fa04e592beb8c01f66394a264e8d37242",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -4904,8 +4904,10 @@
     "app.chain.subscribe.search -> app.domain.meta",
     "app.chain.subscribe.search -> app.domain.meta.metabase",
     "app.chain.subscribe.search -> app.runtime",
+    "app.chain.subscribe.search -> app.runtime.execution",
     "app.chain.subscribe.search -> app.runtime.log",
     "app.chain.subscribe.search -> app.runtime.stop",
+    "app.chain.subscribe.search -> app.runtime.tasks",
     "app.chain.subscribe.search -> app.schemas",
     "app.chain.subscribe.search -> app.schemas.types",
     "app.chain.subscribe.searchtask -> app.application",
diff --git a/tests/test_scheduler_lifecycle.py b/tests/test_scheduler_lifecycle.py
index 693ecd678..5f5c1faae 100644
--- a/tests/test_scheduler_lifecycle.py
+++ b/tests/test_scheduler_lifecycle.py
@@ -849,3 +849,46 @@ def test_cancelled_cross_thread_proxy_waits_for_target_loop_cleanup(
     assert scheduler._registry.is_active("cancel-before-start") is False
     assert scheduler._registry.handles() == ()
     assert not any("was never awaited" in str(item.message) for item in captured)
+
+
+@pytest.mark.anyio
+async def test_subscription_queue_poll_returns_while_owned_search_is_running(monkeypatch) -> None:
+    """长搜索不占用轮询调用，重复轮询不重入且取消必须等待同步 worker 结束。"""
+    from app.chain.subscribe.facade import SubscribeChain
+
+    _patch_progress(monkeypatch)
+    chain = object.__new__(SubscribeChain)
+    started = threading.Event()
+    release = threading.Event()
+    calls = []
+
+    def consume(**kwargs):
+        """模拟慢搜索并捕获手工搜索参数，禁止真实业务调用。"""
+        calls.append(kwargs)
+        started.set()
+        assert release.wait(timeout=5)
+
+    chain.resume_search_queue = consume
+    job_id = "subscribe_search_queue"
+    scheduler = _scheduler(job_id, chain.async_resume_search_queue)
+    scheduler._jobs[job_id]["kwargs"] = {"limit": 2, "manual_sids": (1, 2)}
+    scheduler._poll_subscription_search_queue()
+    try:
+        assert await asyncio.to_thread(started.wait, 2)
+        assert scheduler._is_job_active(job_id)
+        for _ in range(3):
+            scheduler._poll_subscription_search_queue()
+        assert len(calls) == 1
+        assert calls[0]["limit"] == 2
+        assert calls[0]["manual_sids"] == (1, 2)
+        handles = scheduler._registry.handles()
+        assert handles
+        stopping = asyncio.create_task(scheduler.stop_async())
+        await asyncio.sleep(0)
+        assert not stopping.done()
+        assert scheduler._is_job_active(job_id)
+    finally:
+        release.set()
+    await asyncio.wait_for(stopping, timeout=3)
+    assert not scheduler._is_job_active(job_id)
+    assert scheduler._registry.handles() == ()
diff --git a/tests/test_subscription_search_cursor_migration.py b/tests/test_subscription_search_cursor_migration.py
new file mode 100644
index 000000000..108b748db
--- /dev/null
+++ b/tests/test_subscription_search_cursor_migration.py
@@ -0,0 +1,43 @@
+"""搜索站点游标与旧排期升级回归。"""
+
+import importlib
+from datetime import datetime, timedelta, timezone
+
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+
+def test_search_cursor_migration_preserves_cooldown_and_running_tasks(monkeypatch):
+    """只提前未开始的自动排期，冷却、停止、新增保护期和在途任务保持原状。"""
+    migration = importlib.import_module("database.versions.a9c3e7f1b5d8_3_0_31")
+    engine = sa.create_engine("sqlite://")
+    metadata = sa.MetaData()
+    tasks = sa.Table(
+        "subscriptionsearchtask", metadata,
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("source", sa.String()), sa.Column("state", sa.String()),
+        sa.Column("phase", sa.String()), sa.Column("attempt_count", sa.Integer()),
+        sa.Column("cancel_requested", sa.Integer()), sa.Column("available_at", sa.String()),
+    )
+    future = (datetime.now(timezone.utc) + timedelta(days=2)).isoformat(timespec="seconds")
+    with engine.begin() as connection:
+        metadata.create_all(connection)
+        rows = [dict(id=index, source="fallback", state="queued", phase="queued", attempt_count=0,
+                     cancel_requested=0, available_at=future) for index in range(5)]
+        rows[1].update(phase="waiting_site_budget", attempt_count=1)
+        rows[2].update(cancel_requested=1)
+        rows[3].update(source="new", phase="scheduled")
+        rows[4].update(state="running", attempt_count=1)
+        connection.execute(tasks.insert(), rows)
+        monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection)))
+        migration.upgrade()
+        migration.upgrade()
+        stored = connection.execute(sa.select(tasks.c.id, tasks.c.available_at).order_by(tasks.c.id)).all()
+        assert stored[0].available_at < future
+        assert all(row.available_at == future for row in stored[1:])
+        assert "pending_site_ids" in {col["name"] for col in sa.inspect(connection).get_columns(tasks.name)}
+        migration.downgrade()
+        assert "pending_site_ids" not in {col["name"] for col in sa.inspect(connection).get_columns(tasks.name)}
+        assert connection.scalar(sa.select(sa.func.count()).select_from(tasks)) == 5
+    engine.dispose()
diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py
index 525fb7737..f7bfa3edf 100644
--- a/tests/test_subscription_search_governance.py
+++ b/tests/test_subscription_search_governance.py
@@ -100,25 +100,15 @@ def _make_tasks_ready(monkeypatch) -> None:
     )
 
 
-def test_fallback_task_schedule_staggers_each_subscription(monkeypatch):
-    """兜底批次首条抖动后，每条后续订阅都按独立随机间隔到期。"""
+def test_fallback_task_schedule_jitters_once_for_large_batch(monkeypatch):
+    """千条订阅共享一次启动抖动，不能再累计成数日空等。"""
     now = datetime(2026, 9, 3, 1, 2, 3, tzinfo=timezone.utc)
-    delays = iter((12, 60, 300))
-    monkeypatch.setattr(
-        "app.chain.subscribe.search.random.randint",
-        lambda _low, _high: next(delays),
-    )
-
-    schedule = _search_task_available_at(
-        "fallback",
-        (1, 2, 3),
-        now=now,
-    )
-
-    available = [datetime.fromisoformat(schedule[subscribe_id]) for subscribe_id in (1, 2, 3)]
-    assert (available[0] - now).total_seconds() == 12
-    assert (available[1] - available[0]).total_seconds() == 60
-    assert (available[2] - available[1]).total_seconds() == 300
+    jitter = Mock(return_value=12)
+    monkeypatch.setattr("app.chain.subscribe.search.random.randint", jitter)
+    schedule = _search_task_available_at("fallback", tuple(range(1000)), now=now)
+    assert len(schedule) == 1000
+    assert set(schedule.values()) == {(now + timedelta(seconds=12)).isoformat(timespec="seconds")}
+    jitter.assert_called_once_with(0, 60)
 
 
 def test_inline_fallback_search_preserves_site_pressure_stagger(tmp_path, monkeypatch):
@@ -349,7 +339,8 @@ def test_site_budget_conflict_requeues_task_without_batch_failure(tmp_path, monk
         assert task.state == "queued"
         assert task.phase == "waiting_site_budget"
         assert task.available_at == retry_at
-        assert task.last_error == "站点暂时忙，系统会自动继续搜索"
+        assert task.last_error == "等待站点"
+        assert task.pending_site_ids == [31]
     assert chain.subscription_search_repository.claim_next(owner="worker-after-retry") is None
 
 
diff --git a/tests/test_subscription_search_queue.py b/tests/test_subscription_search_queue.py
index f2f29f74e..310835306 100644
--- a/tests/test_subscription_search_queue.py
+++ b/tests/test_subscription_search_queue.py
@@ -362,3 +362,54 @@ def test_search_queue_keeps_manual_work_ahead_of_aged_fallback(tmp_path):
 
     assert claimed.subscription_id == 9
     assert claimed.source == "manual"
+
+
+def test_retry_sites_survive_reopen_and_admission_wait_without_stale_error(tmp_path):
+    """站点游标跨重建仓储和准入等待保留，真正恢复时清除上轮等待提示。"""
+    repository, engine = _repository(tmp_path)
+    repository.enqueue(subscription_ids=(701,), source="fallback", priority=10)
+    first = repository.claim_next(owner="before-restart")
+    ready_at = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat(timespec="seconds")
+    assert repository.defer_task(
+        task_id=first.task_id, lease_token=first.lease_token, available_at=ready_at,
+        message="站点冷却中", pending_site_ids=(8, 9),
+    )
+    reopened = TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine))
+    resumed = reopened.claim_next(owner="after-restart")
+    assert resumed.task_id == first.task_id
+    assert resumed.pending_site_ids == (8, 9)
+    assert resumed.state == "running"
+    assert resumed.last_error is None
+    assert reopened.defer_task(
+        task_id=resumed.task_id, lease_token=resumed.lease_token, available_at=ready_at,
+        phase="waiting_subscription", message="等待任务",
+    )
+    continued = reopened.claim_next(owner="after-match")
+    assert continued.pending_site_ids == (8, 9)
+    assert continued.last_error is None
+    assert reopened.defer_task(
+        task_id=continued.task_id, lease_token=first.lease_token, available_at=ready_at,
+        pending_site_ids=(99,),
+    ) is False
+    assert reopened.finish_task(task_id=continued.task_id, lease_token=continued.lease_token, state="completed")
+    engine.dispose()
+
+
+def test_manual_search_restarts_full_scope_but_automatic_merge_keeps_cursor(tmp_path):
+    """自动调度合并保留进度，用户主动重搜可包含新配置的站点。"""
+    repository, engine = _repository(tmp_path)
+    repository.enqueue(subscription_ids=(702,), source="fallback", priority=10)
+    first = repository.claim_next(owner="worker")
+    ready_at = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat(timespec="seconds")
+    assert repository.defer_task(
+        task_id=first.task_id, lease_token=first.lease_token, available_at=ready_at,
+        pending_site_ids=(9,),
+    )
+    repository.enqueue(subscription_ids=(702,), source="fallback", priority=10)
+    with Session(engine) as session:
+        assert session.scalar(select(SubscriptionSearchTask.pending_site_ids)) == [9]
+    repository.enqueue(subscription_ids=(702,), source="manual", priority=100)
+    resumed = repository.claim_next(owner="manual")
+    assert resumed.pending_site_ids is None
+    assert resumed.source == "manual"
+    engine.dispose()
diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py
index 4a09db9a1..4953a757f 100644
--- a/tests/test_subscription_site_budget.py
+++ b/tests/test_subscription_site_budget.py
@@ -549,3 +549,48 @@ def test_search_provider_logs_site_budget_release_failure(monkeypatch):
     assert result == ["torrent"]
     assert metrics.snapshot().release_failure_count == 1
     assert errors == ["订阅站点预算释放失败: site_id=14 site=Stale"]
+
+
+def test_retry_provider_only_searches_pending_enabled_sites(monkeypatch):
+    """恢复时不重搜成功站点和插件源，也不重新启用已经移除的站点。"""
+    from unittest.mock import Mock
+
+    from app.chain.search.provider import SearchProviderOwner
+
+    chain = object.__new__(SearchChain)
+    chain.configure_subscription_site_budget(SubscriptionSiteBudget(
+        repository=_WaitingRepository(), owner="retry", cancelled=lambda: False,
+        stop_state=ProcessStopState(), pending_site_ids=(2, 3),
+    ))
+    chain._sync_indexers = lambda _sites: [{"id": 1}, {"id": 2}]
+    chain._torrent_type = lambda *_args: None
+    chain._torrent_keyword = lambda *_args: "movie"
+    chain._build_search_pages = lambda _page: [0]
+    chain.search_plugin_torrents = Mock(return_value=[])
+    captured = []
+
+    def collect(**kwargs):
+        """只观察 provider 选择，禁止真实站点和插件调用。"""
+        captured.extend(site["id"] for site in kwargs["indexer_sites"])
+        return {}
+
+    monkeypatch.setattr(SearchProviderOwner, "_collect_sync_site_results", lambda _self, **kwargs: collect(**kwargs))
+    monkeypatch.setattr("app.chain.search.provider.ProgressHelper", Mock())
+    SearchProviderOwner._search_all_sites(chain, keyword="movie", sites=[1, 2])
+    assert captured == [2]
+    chain.search_plugin_torrents.assert_not_called()
+
+
+def test_deferral_reports_cooldown_separately_from_busy():
+    """冷却提示不伪装成站点占用，并保留所有未完成站点。"""
+    from app.application.subscription.execution import raise_subscription_site_budget_deferral
+    from app.application.subscription.sitebudget import SubscriptionSearchDeferred, SubscriptionSiteBudgetDeferral
+
+    deferred_sites = (
+        SubscriptionSiteBudgetDeferral(site_id=1, retry_at="2026-09-08T10:00:00+00:00", wait_reason="cooldown"),
+        SubscriptionSiteBudgetDeferral(site_id=2, retry_at="2026-09-08T11:00:00+00:00", wait_reason="cooldown"),
+    )
+    with pytest.raises(SubscriptionSearchDeferred, match="站点冷却中") as caught:
+        raise_subscription_site_budget_deferral(deferred_sites, None)
+    assert caught.value.site_ids == (1, 2)
+    assert caught.value.retry_at == deferred_sites[0].retry_at