refactor(subscribe): persist fallback search queue

app/application/chain/context.py

@@ -33,6 +33,7 @@ if TYPE_CHECKING:
         SubscriptionMutationScope,
         SyncSubscriptionMutationScope,
     )
+    from app.application.subscription.execution import SubscriptionSearchRepository
     from app.application.transfer.execution import TransferExecutionRepository
     from app.application.transfer.workflow import TransferAdmissionRepository
 
@@ -74,6 +75,7 @@ class ChainRuntimeContext:
     media_server_repository: MediaServerRepository
     download_failure_repository: DownloadFailureRepository
     user_repository: ChainUserRepository
+    subscription_search_repository: Optional[SubscriptionSearchRepository] = None
     legacy_transfer_command: Optional[LegacyTransferCommand] = None
     durable_event_writer: Optional[ChainDurableEventWriter] = None
     configuration: ChainRuntimeConfig = field(

app/application/subscription/execution.py

@@ -0,0 +1,105 @@
+"""订阅搜索执行批次、任务与持久队列端口。"""
+
+from dataclasses import dataclass
+from typing import Optional, Protocol
+
+
+@dataclass(frozen=True, slots=True)
+class SearchBatchSnapshot:
+    """订阅搜索批次的持久业务状态快照。"""
+
+    batch_id: str
+    source: str
+    state: str
+    priority: int
+    total_count: int
+    finished_count: int
+    failed_count: int
+    cancelled_count: int
+    cancel_requested: bool
+    created_at: str
+    updated_at: str
+    started_at: Optional[str] = None
+    finished_at: Optional[str] = None
+    last_error: Optional[str] = None
+
+
+@dataclass(frozen=True, slots=True)
+class SearchTaskSnapshot:
+    """一个可认领、恢复和取消的订阅搜索任务快照。"""
+
+    task_id: str
+    batch_id: str
+    subscription_id: int
+    source: str
+    priority: int
+    position: int
+    state: str
+    attempt_count: int
+    cancel_requested: bool
+    lease_token: Optional[str]
+    created_at: str
+    updated_at: str
+    started_at: Optional[str] = None
+    finished_at: Optional[str] = None
+    last_error: Optional[str] = None
+
+
+@dataclass(frozen=True, slots=True)
+class SearchEnqueueResult:
+    """一次批次入队结果，区分新任务与 single-flight 合并。"""
+
+    batch: SearchBatchSnapshot
+    created_count: int
+    coalesced_count: int
+
+
+class SubscriptionSearchRepository(Protocol):
+    """订阅搜索批次与任务的持久队列端口。"""
+
+    def enqueue(
+        self,
+        *,
+        subscription_ids: tuple[int, ...],
+        source: str,
+        priority: int,
+    ) -> SearchEnqueueResult:
+        """按订阅 ID 建立批次，并合并已存在的活动任务。"""
+        ...
+
+    def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]:
+        """按优先级和稳定游标认领下一条可执行任务。"""
+        ...
+
+    def finish_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        state: str,
+        error: Optional[str] = None,
+    ) -> bool:
+        """以租约令牌收口任务，并推进所属批次聚合状态。"""
+        ...
+
+    def release_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        cancelled: bool = False,
+    ) -> bool:
+        """释放尚未完成的任务租约，供停止或取消后恢复。"""
+        ...
+
+    def is_cancel_requested(self, task_id: str) -> bool:
+        """判断任务或所属批次是否已请求取消。"""
+        ...
+
+    def request_cancel(self, batch_id: str) -> bool:
+        """请求取消批次，并立即终止尚未发出的排队任务。"""
+        ...
+
+    def get_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]:
+        """按稳定批次 ID 返回当前聚合状态。"""
+        ...

app/chain/base.py

@@ -58,6 +58,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
         self.async_filecache = context.async_file_cache
         self.site_repository = context.site_repository
         self.subscription_repository = context.subscription_repository
+        self.subscription_search_repository = context.subscription_search_repository
         self.subscription_mutation_scope = context.subscription_mutation_scope
         self.sync_subscription_mutation_scope = context.sync_subscription_mutation_scope
         self.subscription_delete_scope = context.subscription_delete_scope

app/chain/subscribe/contract.py

@@ -25,6 +25,7 @@ if TYPE_CHECKING:
         site_repository: Any
         subscription_completion_scope: Callable[..., Any]
         subscription_repository: Any
+        subscription_search_repository: Any
         sync_subscription_delete_scope: Callable[..., Any]
         sync_subscription_mutation_scope: Callable[..., Any]
 
@@ -98,6 +99,7 @@ if TYPE_CHECKING:
         remote_list: Callable[..., Any]
         resolve_subscribe_missing: Callable[..., Any]
         reconcile_subscription_completion: Callable[..., Any]
+        resume_search_queue: Callable[..., Any]
 
     _SubscribeOwnerBase = _SubscribeOwnerHost
 else:

app/chain/subscribe/facade.py

@@ -49,6 +49,7 @@ class SubscribeChain(
 
     _interaction_handler_type = SubscribeInteractionHandler
     _rlock = threading.RLock()
+    _search_queue_lock = threading.Lock()
     _LOCK_TIMOUT = 3600 * 2
 
     @classmethod

app/chain/subscribe/search.py

@@ -4,6 +4,7 @@ import random
 import time
 from datetime import datetime
 from typing import Any, Callable, Optional, cast
+from uuid import uuid4
 
 from app.application.configuration import get_configured_system_config
 from app.application.subscription.contract import (
@@ -13,6 +14,7 @@ from app.application.subscription.contract import (
     subscribe_media_key,
 )
 from app.application.subscription.query import SubscriptionQueryService
+from app.application.subscription.execution import SearchBatchSnapshot, SubscriptionSearchRepository
 from app.chain.media import MediaChain
 from app.chain.search.facade import SearchChain
 from app.chain.subscribe.contract import _SubscribeOwnerBase
@@ -81,7 +83,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
         manual: Optional[bool] = False,
         progress_callback: Optional[Callable[..., None]] = None,
         sids: Optional[tuple[int, ...]] = None,
-    ) -> None:
+    ) -> Optional[str]:
         """
         执行订阅搜索。
 
@@ -102,7 +104,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
         manual: Optional[bool] = False,
         progress_callback: Optional[Callable[..., None]] = None,
         sids: Optional[tuple[int, ...]] = None,
-    ) -> None:
+    ) -> Optional[str]:
         """
         订阅搜索
         :param sid: 订阅ID，有值时只处理该订阅
@@ -112,6 +114,37 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
         :param sids: 订阅ID集合，有值时按给定顺序处理
         :return: 更新订阅状态为R或删除订阅
         """
+        queue = cast(
+            Optional[SubscriptionSearchRepository],
+            getattr(self, "subscription_search_repository", None),
+        )
+        if queue is not None:
+            return self._execute_queued_search(
+                queue=queue,
+                sid=sid,
+                sids=sids,
+                state=state,
+                manual=manual,
+                progress_callback=progress_callback,
+            )
+        self._execute_inline_search(
+            sid=sid,
+            sids=sids,
+            state=state,
+            manual=manual,
+            progress_callback=progress_callback,
+        )
+        return None
+
+    def _execute_inline_search(
+        self,
+        sid: Optional[int],
+        sids: Optional[tuple[int, ...]],
+        state: Optional[str],
+        manual: Optional[bool],
+        progress_callback: Optional[Callable[..., None]],
+    ) -> None:
+        """保留未注入持久队列宿主的旧串行锁语义。"""
         lock_acquired = self._acquire_run_lock("search", progress_callback)
         if not lock_acquired:
             return
@@ -154,6 +187,216 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
             self._rlock.release()
             logger.debug(f"search Lock released at {datetime.now()}")
 
+    def _execute_queued_search(
+        self,
+        *,
+        queue: SubscriptionSearchRepository,
+        sid: Optional[int],
+        sids: Optional[tuple[int, ...]],
+        state: Optional[str],
+        manual: Optional[bool],
+        progress_callback: Optional[Callable[..., None]],
+    ) -> str:
+        """将搜索转为持久任务并在无 Match 长锁的短租约中串行消费。"""
+        subscribes = self._load_search_subscriptions(sid=sid, sids=sids, state=state)
+        source, priority = self._search_source_and_priority(
+            sid=sid,
+            sids=sids,
+            state=state,
+            manual=manual,
+        )
+        enqueued = queue.enqueue(
+            subscription_ids=tuple(subscribe.id for subscribe in subscribes),
+            source=source,
+            priority=priority,
+        )
+        total = len(subscribes)
+        if progress_callback:
+            progress_callback(
+                value=0,
+                text=f"开始订阅搜索，共 {total} 个订阅 ...",
+                data={
+                    "batch_id": enqueued.batch.batch_id,
+                    "total": total,
+                    "finished": 0,
+                    "coalesced": enqueued.coalesced_count,
+                },
+            )
+        processed = self._drain_search_queue(
+            queue=queue,
+            limit=max(1, enqueued.created_count + enqueued.coalesced_count),
+            progress_callback=progress_callback,
+        )
+        processed_subscribes = [item for item in subscribes if item.id in processed]
+        self._notify_manual_search(manual, sid, sids, subscribes, processed_subscribes)
+        if progress_callback:
+            batch = queue.get_batch(enqueued.batch.batch_id)
+            progress_callback(
+                value=100,
+                text=self._batch_progress_text(batch),
+                data={
+                    "batch_id": enqueued.batch.batch_id,
+                    "total": total,
+                    "finished": len(processed_subscribes),
+                    "coalesced": enqueued.coalesced_count,
+                },
+            )
+        return enqueued.batch.batch_id
+
+    def _drain_search_queue(
+        self,
+        *,
+        queue: SubscriptionSearchRepository,
+        limit: int,
+        progress_callback: Optional[Callable[..., None]],
+    ) -> set[int]:
+        """有界消费可恢复任务；单任务失败不得阻止后续订阅。"""
+        queue_lock = getattr(self, "_search_queue_lock", None)
+        if queue_lock is not None and not queue_lock.acquire(blocking=False):
+            logger.debug("订阅搜索队列已有消费者，本轮仅保留持久任务")
+            return set()
+        owner = f"subscribe-search:{uuid4().hex}"
+        processed: set[int] = set()
+        try:
+            searchchain = SearchChain()
+            for index in range(1, limit + 1):
+                if runtime_stop_state.is_system_stopped:
+                    break
+                task = queue.claim_next(owner=owner)
+                if task is None:
+                    break
+                if not task.lease_token:
+                    logger.error(f"订阅搜索任务 {task.task_id} 缺少租约令牌，跳过执行")
+                    continue
+                if queue.is_cancel_requested(task.task_id):
+                    queue.release_task(
+                        task_id=task.task_id,
+                        lease_token=task.lease_token,
+                        cancelled=True,
+                    )
+                    continue
+                subscribe = self.subscription_repository.get(task.subscription_id)
+                if subscribe is None:
+                    queue.finish_task(
+                        task_id=task.task_id,
+                        lease_token=task.lease_token,
+                        state="cancelled",
+                        error="订阅已不存在",
+                    )
+                    continue
+                self._report_search_progress(progress_callback, subscribe, index, limit)
+                if self._defer_recent_subscription(subscribe):
+                    queue.finish_task(
+                        task_id=task.task_id,
+                        lease_token=task.lease_token,
+                        state="completed",
+                    )
+                    continue
+                current = subscribe
+                try:
+                    current = self._process_search_subscription(subscribe, searchchain)
+                    if queue.is_cancel_requested(task.task_id):
+                        queue.release_task(
+                            task_id=task.task_id,
+                            lease_token=task.lease_token,
+                            cancelled=True,
+                        )
+                    else:
+                        queue.finish_task(
+                            task_id=task.task_id,
+                            lease_token=task.lease_token,
+                            state="completed",
+                        )
+                    processed.add(subscribe.id)
+                except Exception as err:
+                    logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True)
+                    queue.finish_task(
+                        task_id=task.task_id,
+                        lease_token=task.lease_token,
+                        state="failed",
+                        error=str(err),
+                    )
+                finally:
+                    if current and current.state == "N":
+                        try:
+                            self._SubscribeChain__apply_subscribe_update(
+                                current,
+                                {"state": "R"},
+                                scene="search_reset",
+                            )
+                        except Exception as err:
+                            logger.error(
+                                f"订阅 {current.name} 搜索后状态重置失败：{str(err)}",
+                                exc_info=True,
+                            )
+                    self._report_search_progress(progress_callback, subscribe, index, limit, finished=True)
+        finally:
+            if queue_lock is not None:
+                queue_lock.release()
+        return processed
+
+    def resume_search_queue(
+        self,
+        progress_callback: Optional[Callable[..., None]] = None,
+        limit: int = 50,
+    ) -> None:
+        """短周期恢复排队或租约过期任务，不创建新的 24 小时兜底批次。"""
+        queue = cast(
+            Optional[SubscriptionSearchRepository],
+            getattr(self, "subscription_search_repository", None),
+        )
+        if queue is None:
+            return
+        self._drain_search_queue(
+            queue=queue,
+            limit=max(1, limit),
+            progress_callback=progress_callback,
+        )
+
+    @staticmethod
+    def _search_source_and_priority(
+        *,
+        sid: Optional[int],
+        sids: Optional[tuple[int, ...]],
+        state: Optional[str],
+        manual: Optional[bool],
+    ) -> tuple[str, int]:
+        """把兼容入口归一为持久来源和公平队列优先级。"""
+        if manual:
+            return "manual", 100
+        if sid or sids is not None:
+            return "targeted", 80
+        if state in {"R", "P"}:
+            return "fallback", 10
+        return "new", 50
+
+    @staticmethod
+    def _batch_progress_text(batch: Optional[SearchBatchSnapshot]) -> str:
+        """把批次聚合终态转为兼容进度文案。"""
+        if batch is None:
+            return "订阅搜索任务已提交"
+        if batch.state == "failed":
+            return "订阅搜索完成，部分任务失败"
+        if batch.state == "cancelled":
+            return "订阅搜索已取消"
+        return "订阅搜索完成"
+
+    def cancel_search_batch(self, batch_id: str) -> bool:
+        """请求取消持久搜索批次；未注入队列时返回失败。"""
+        queue = cast(
+            Optional[SubscriptionSearchRepository],
+            getattr(self, "subscription_search_repository", None),
+        )
+        return bool(queue and queue.request_cancel(batch_id))
+
+    def get_search_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]:
+        """返回持久搜索批次状态；未注入队列时返回空。"""
+        queue = cast(
+            Optional[SubscriptionSearchRepository],
+            getattr(self, "subscription_search_repository", None),
+        )
+        return queue.get_batch(batch_id) if queue else None
+
     def _load_search_subscriptions(
         self,
         sid: Optional[int],

app/db/adapters/subscriptionsearch.py

@@ -0,0 +1,169 @@
+"""订阅搜索持久队列的 SQLAlchemy 适配器。"""
+
+from collections.abc import Callable
+from typing import Optional, TypeVar
+
+from sqlalchemy.orm import Session
+
+from app.application.subscription.execution import (
+    SearchBatchSnapshot,
+    SearchEnqueueResult,
+    SearchTaskSnapshot,
+)
+from app.db.models.subscriptionsearch import (
+    SubscriptionSearchBatch,
+    SubscriptionSearchTask,
+)
+from app.db.oper.subscriptionsearch import SubscriptionSearchOper
+from app.db.uow import SqlAlchemyUnitOfWork
+
+T = TypeVar("T")
+
+
+def _batch(record: SubscriptionSearchBatch) -> SearchBatchSnapshot:
+    """在 Session 内投影不可变搜索批次快照。"""
+    return SearchBatchSnapshot(
+        batch_id=record.batch_id,
+        source=record.source,
+        state=record.state,
+        priority=record.priority,
+        total_count=record.total_count,
+        finished_count=record.finished_count,
+        failed_count=record.failed_count,
+        cancelled_count=record.cancelled_count,
+        cancel_requested=bool(record.cancel_requested),
+        created_at=record.created_at,
+        updated_at=record.updated_at,
+        started_at=record.started_at,
+        finished_at=record.finished_at,
+        last_error=record.last_error,
+    )
+
+
+def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot:
+    """在 Session 内投影不可变搜索任务快照。"""
+    return SearchTaskSnapshot(
+        task_id=record.task_id,
+        batch_id=record.batch_id,
+        subscription_id=record.subscription_id,
+        source=record.source,
+        priority=record.priority,
+        position=record.position,
+        state=record.state,
+        attempt_count=record.attempt_count,
+        cancel_requested=bool(record.cancel_requested),
+        lease_token=record.lease_token,
+        created_at=record.created_at,
+        updated_at=record.updated_at,
+        started_at=record.started_at,
+        finished_at=record.finished_at,
+        last_error=record.last_error,
+    )
+
+
+class TransactionalSubscriptionSearchRepository:
+    """使用短事务实现订阅搜索队列端口。"""
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """保存由组合根注入的同步 Session 工厂。"""
+        self._session_factory = session_factory
+
+    def _read(self, operation: Callable[[SubscriptionSearchOper], T]) -> T:
+        """在短 Session 中执行一次只读查询。"""
+        with self._session_factory() as session:
+            return operation(SubscriptionSearchOper(session))
+
+    def _write(self, operation: Callable[[SubscriptionSearchOper], T]) -> T:
+        """在短事务中执行一次队列状态变更。"""
+        with self._session_factory() as session:
+            unit_of_work = SqlAlchemyUnitOfWork(session)
+            try:
+                result = operation(SubscriptionSearchOper(session))
+                unit_of_work.commit()
+                return result
+            except Exception:
+                unit_of_work.rollback()
+                raise
+
+    def enqueue(
+        self,
+        *,
+        subscription_ids: tuple[int, ...],
+        source: str,
+        priority: int,
+    ) -> SearchEnqueueResult:
+        """创建批次并返回 single-flight 合并计数。"""
+        def operation(repository: SubscriptionSearchOper) -> SearchEnqueueResult:
+            """在同一事务内创建批次和任务。"""
+            record, created, coalesced = repository.enqueue(
+                subscription_ids=subscription_ids,
+                source=source,
+                priority=priority,
+            )
+            return SearchEnqueueResult(
+                batch=_batch(record),
+                created_count=created,
+                coalesced_count=coalesced,
+            )
+
+        return self._write(operation)
+
+    def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]:
+        """认领下一任务并返回脱离 Session 的快照。"""
+        return self._write(
+            lambda repository: (
+                _task(record)
+                if (record := repository.claim_next(owner=owner, lease_seconds=lease_seconds)) is not None
+                else None
+            )
+        )
+
+    def finish_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        state: str,
+        error: Optional[str] = None,
+    ) -> bool:
+        """以租约令牌收口任务。"""
+        return self._write(
+            lambda repository: repository.finish_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                state=state,
+                error=error,
+            )
+        )
+
+    def release_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        cancelled: bool = False,
+    ) -> bool:
+        """释放执行租约或收口取消。"""
+        return self._write(
+            lambda repository: repository.release_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                cancelled=cancelled,
+            )
+        )
+
+    def is_cancel_requested(self, task_id: str) -> bool:
+        """查询任务或批次的取消请求。"""
+        return self._read(lambda repository: repository.is_cancel_requested(task_id))
+
+    def request_cancel(self, batch_id: str) -> bool:
+        """请求取消批次。"""
+        return self._write(lambda repository: repository.request_cancel(batch_id))
+
+    def get_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]:
+        """查询批次聚合快照。"""
+        return self._read(
+            lambda repository: (
+                _batch(record) if (record := repository.get_batch(batch_id)) is not None else None
+            )
+        )

app/db/models/__init__.py

@@ -35,6 +35,14 @@ _MODEL_EXPORTS = {
         "app.db.models.subscribehistory",
         "SubscribeHistory",
     ),
+    "SubscriptionSearchBatch": (
+        "app.db.models.subscriptionsearch",
+        "SubscriptionSearchBatch",
+    ),
+    "SubscriptionSearchTask": (
+        "app.db.models.subscriptionsearch",
+        "SubscriptionSearchTask",
+    ),
     "SystemConfig": ("app.db.models.systemconfig", "SystemConfig"),
     "TransferHistory": ("app.db.models.transferhistory", "TransferHistory"),
     "TransferExecutionStep": (

app/db/models/subscriptionsearch.py

@@ -0,0 +1,72 @@
+"""订阅搜索持久批次与任务模型。"""
+
+from typing import Optional
+
+from sqlalchemy import Index, Integer, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base, get_id_column
+
+
+class SubscriptionSearchBatch(Base):
+    """记录一次订阅搜索请求及其可观察聚合终态。"""
+
+    id = get_id_column()
+    batch_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    source: Mapped[str] = mapped_column(String(32), nullable=False)
+    state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
+    priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    total_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    finished_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    failed_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    cancelled_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    cancel_requested: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    created_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    updated_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    started_at: Mapped[Optional[str]] = mapped_column(String(40))
+    finished_at: Mapped[Optional[str]] = mapped_column(String(40))
+    last_error: Mapped[Optional[str]] = mapped_column(Text)
+
+    __table_args__ = (
+        UniqueConstraint("batch_id", name="uq_subscriptionsearchbatch_batch_id"),
+        Index("ix_subscriptionsearchbatch_state_created", "state", "created_at", "id"),
+    )
+
+
+class SubscriptionSearchTask(Base):
+    """记录一个具有 single-flight、租约和恢复游标的订阅搜索任务。"""
+
+    id = get_id_column()
+    task_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    batch_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    subscription_id: Mapped[int] = mapped_column(Integer, nullable=False)
+    active_key: Mapped[Optional[str]] = mapped_column(String(128))
+    source: Mapped[str] = mapped_column(String(32), nullable=False)
+    priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    position: Mapped[int] = mapped_column(Integer, nullable=False)
+    state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
+    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))
+    lease_token: Mapped[Optional[str]] = mapped_column(String(64))
+    lease_expires_at: Mapped[Optional[str]] = mapped_column(String(40))
+    created_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    updated_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    started_at: Mapped[Optional[str]] = mapped_column(String(40))
+    finished_at: Mapped[Optional[str]] = mapped_column(String(40))
+    last_error: Mapped[Optional[str]] = mapped_column(Text)
+
+    __table_args__ = (
+        UniqueConstraint("task_id", name="uq_subscriptionsearchtask_task_id"),
+        UniqueConstraint("active_key", name="uq_subscriptionsearchtask_active_key"),
+        Index(
+            "ix_subscriptionsearchtask_claim",
+            "state",
+            "priority",
+            "lease_expires_at",
+            "created_at",
+            "id",
+        ),
+        Index("ix_subscriptionsearchtask_batch_position", "batch_id", "position", "id"),
+        Index("ix_subscriptionsearchtask_subscription", "subscription_id", "created_at", "id"),
+    )

app/db/oper/subscriptionsearch.py

@@ -0,0 +1,385 @@
+"""订阅搜索批次与任务的持久队列读写。"""
+
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+from uuid import uuid4
+
+from sqlalchemy import and_, case, func, or_, select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.db.base import DbOper, execute_dml
+from app.db.models.subscriptionsearch import (
+    SubscriptionSearchBatch,
+    SubscriptionSearchTask,
+)
+
+
+def utc_now_text() -> str:
+    """返回可按字符串稳定排序的 UTC ISO 时间。"""
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+class SubscriptionSearchOper(DbOper):
+    """在调用方事务中维护搜索队列、租约和批次聚合。"""
+
+    def enqueue(
+        self,
+        *,
+        subscription_ids: tuple[int, ...],
+        source: str,
+        priority: int,
+    ) -> tuple[SubscriptionSearchBatch, int, int]:
+        """创建批次，并以活动键合并同一订阅的重叠搜索入口。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅搜索入队需要调用方提供同步 Session")
+        now = utc_now_text()
+        batch = SubscriptionSearchBatch(
+            batch_id=uuid4().hex,
+            source=source,
+            state="queued",
+            priority=priority,
+            total_count=0,
+            created_at=now,
+            updated_at=now,
+        )
+        self._db.add(batch)
+        self._db.flush()
+        created = 0
+        coalesced = 0
+        for position, subscription_id in enumerate(dict.fromkeys(subscription_ids)):
+            active_key = f"subscription:{subscription_id}"
+            task = SubscriptionSearchTask(
+                task_id=uuid4().hex,
+                batch_id=batch.batch_id,
+                subscription_id=subscription_id,
+                active_key=active_key,
+                source=source,
+                priority=priority,
+                position=position,
+                state="queued",
+                created_at=now,
+                updated_at=now,
+            )
+            try:
+                with self._db.begin_nested():
+                    self._db.add(task)
+                    self._db.flush()
+                created += 1
+            except IntegrityError:
+                coalesced += 1
+                execute_dml(
+                    self._db,
+                    update(SubscriptionSearchTask)
+                    .where(SubscriptionSearchTask.active_key == active_key)
+                    .values(
+                        priority=case(
+                            (SubscriptionSearchTask.priority < priority, priority),
+                            else_=SubscriptionSearchTask.priority,
+                        ),
+                        updated_at=now,
+                    ),
+                    execution_options={"synchronize_session": False},
+                )
+        batch.total_count = created
+        if created == 0:
+            batch.state = "completed"
+            batch.finished_at = now
+        return batch, created, coalesced
+
+    def claim_next(self, *, owner: str, lease_seconds: int) -> Optional[SubscriptionSearchTask]:
+        """使用 CAS 认领最高优先级任务，过期 running 任务可被恢复。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅搜索认领需要调用方提供同步 Session")
+        now = utc_now_text()
+        lease_expires_at = (
+            datetime.now(timezone.utc) + timedelta(seconds=max(1, lease_seconds))
+        ).isoformat(timespec="seconds")
+        for _attempt in range(5):
+            candidate = self._db.execute(
+                select(SubscriptionSearchTask)
+                .where(
+                    SubscriptionSearchTask.cancel_requested == 0,
+                    or_(
+                        SubscriptionSearchTask.state == "queued",
+                        and_(
+                            SubscriptionSearchTask.state == "running",
+                            or_(
+                                SubscriptionSearchTask.lease_expires_at.is_(None),
+                                SubscriptionSearchTask.lease_expires_at <= now,
+                            ),
+                        ),
+                    ),
+                )
+                .order_by(
+                    SubscriptionSearchTask.priority.desc(),
+                    SubscriptionSearchTask.created_at.asc(),
+                    SubscriptionSearchTask.position.asc(),
+                    SubscriptionSearchTask.id.asc(),
+                )
+                .limit(1)
+            ).scalars().first()
+            if candidate is None:
+                return None
+            lease_token = uuid4().hex
+            claimed = execute_dml(
+                self._db,
+                update(SubscriptionSearchTask)
+                .where(
+                    SubscriptionSearchTask.id == candidate.id,
+                    SubscriptionSearchTask.cancel_requested == 0,
+                    or_(
+                        SubscriptionSearchTask.state == "queued",
+                        and_(
+                            SubscriptionSearchTask.state == "running",
+                            or_(
+                                SubscriptionSearchTask.lease_expires_at.is_(None),
+                                SubscriptionSearchTask.lease_expires_at <= now,
+                            ),
+                        ),
+                    ),
+                )
+                .values(
+                    state="running",
+                    lease_owner=owner,
+                    lease_token=lease_token,
+                    lease_expires_at=lease_expires_at,
+                    attempt_count=SubscriptionSearchTask.attempt_count + 1,
+                    started_at=func.coalesce(SubscriptionSearchTask.started_at, now),
+                    updated_at=now,
+                ),
+                execution_options={"synchronize_session": False},
+            )
+            if not claimed:
+                self._db.expire_all()
+                continue
+            execute_dml(
+                self._db,
+                update(SubscriptionSearchBatch)
+                .where(
+                    SubscriptionSearchBatch.batch_id == candidate.batch_id,
+                    SubscriptionSearchBatch.state.in_(("queued", "running")),
+                )
+                .values(
+                    state="running",
+                    started_at=func.coalesce(SubscriptionSearchBatch.started_at, now),
+                    updated_at=now,
+                ),
+                execution_options={"synchronize_session": False},
+            )
+            self._db.flush()
+            self._db.expire_all()
+            return self._db.execute(
+                select(SubscriptionSearchTask).where(
+                    SubscriptionSearchTask.id == candidate.id,
+                    SubscriptionSearchTask.lease_token == lease_token,
+                )
+            ).scalars().first()
+        return None
+
+    def finish_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        state: str,
+        error: Optional[str],
+    ) -> bool:
+        """以当前租约令牌收口任务，并重新计算批次终态。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅搜索收口需要调用方提供同步 Session")
+        if state not in {"completed", "failed", "cancelled"}:
+            raise ValueError(f"不支持的订阅搜索终态：{state}")
+        task = self._db.execute(
+            select(SubscriptionSearchTask).where(
+                SubscriptionSearchTask.task_id == task_id,
+                SubscriptionSearchTask.state == "running",
+                SubscriptionSearchTask.lease_token == lease_token,
+            )
+        ).scalars().first()
+        if task is None:
+            return False
+        now = utc_now_text()
+        updated = execute_dml(
+            self._db,
+            update(SubscriptionSearchTask)
+            .where(
+                SubscriptionSearchTask.id == task.id,
+                SubscriptionSearchTask.state == "running",
+                SubscriptionSearchTask.lease_token == lease_token,
+            )
+            .values(
+                state=state,
+                active_key=None,
+                lease_owner=None,
+                lease_token=None,
+                lease_expires_at=None,
+                updated_at=now,
+                finished_at=now,
+                last_error=error,
+            ),
+            execution_options={"synchronize_session": False},
+        )
+        if not updated:
+            return False
+        self._refresh_batch(task.batch_id, now=now, error=error)
+        return True
+
+    def release_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        cancelled: bool,
+    ) -> bool:
+        """取消时收口，停机时把任务退回队列并保留稳定游标。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅搜索释放需要调用方提供同步 Session")
+        task = self._db.execute(
+            select(SubscriptionSearchTask).where(
+                SubscriptionSearchTask.task_id == task_id,
+                SubscriptionSearchTask.state == "running",
+                SubscriptionSearchTask.lease_token == lease_token,
+            )
+        ).scalars().first()
+        if task is None:
+            return False
+        should_cancel = cancelled or bool(task.cancel_requested) or self._batch_cancel_requested(task.batch_id)
+        if should_cancel:
+            return self.finish_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                state="cancelled",
+                error=None,
+            )
+        now = utc_now_text()
+        return bool(execute_dml(
+            self._db,
+            update(SubscriptionSearchTask)
+            .where(
+                SubscriptionSearchTask.id == task.id,
+                SubscriptionSearchTask.state == "running",
+                SubscriptionSearchTask.lease_token == lease_token,
+            )
+            .values(
+                state="queued",
+                lease_owner=None,
+                lease_token=None,
+                lease_expires_at=None,
+                updated_at=now,
+            ),
+            execution_options={"synchronize_session": False},
+        ))
+
+    def is_cancel_requested(self, task_id: str) -> bool:
+        """读取任务和批次取消标记。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅搜索取消查询需要调用方提供同步 Session")
+        task = self._db.execute(
+            select(SubscriptionSearchTask).where(SubscriptionSearchTask.task_id == task_id)
+        ).scalars().first()
+        return bool(task and (task.cancel_requested or self._batch_cancel_requested(task.batch_id)))
+
+    def request_cancel(self, batch_id: str) -> bool:
+        """标记批次取消，并立即收口尚未发出的排队任务。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅搜索取消需要调用方提供同步 Session")
+        batch = self._db.execute(
+            select(SubscriptionSearchBatch).where(SubscriptionSearchBatch.batch_id == batch_id)
+        ).scalars().first()
+        if batch is None or batch.state in {"completed", "failed", "cancelled"}:
+            return False
+        now = utc_now_text()
+        execute_dml(
+            self._db,
+            update(SubscriptionSearchBatch)
+            .where(SubscriptionSearchBatch.id == batch.id)
+            .values(cancel_requested=1, state="cancelling", updated_at=now),
+            execution_options={"synchronize_session": False},
+        )
+        execute_dml(
+            self._db,
+            update(SubscriptionSearchTask)
+            .where(
+                SubscriptionSearchTask.batch_id == batch_id,
+                SubscriptionSearchTask.state == "queued",
+            )
+            .values(
+                state="cancelled",
+                active_key=None,
+                cancel_requested=1,
+                finished_at=now,
+                updated_at=now,
+            ),
+            execution_options={"synchronize_session": False},
+        )
+        execute_dml(
+            self._db,
+            update(SubscriptionSearchTask)
+            .where(
+                SubscriptionSearchTask.batch_id == batch_id,
+                SubscriptionSearchTask.state == "running",
+            )
+            .values(cancel_requested=1, updated_at=now),
+            execution_options={"synchronize_session": False},
+        )
+        self._refresh_batch(batch_id, now=now, error=None)
+        return True
+
+    def get_batch(self, batch_id: str) -> Optional[SubscriptionSearchBatch]:
+        """按稳定批次 ID 读取聚合记录。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅搜索批次查询需要调用方提供同步 Session")
+        return self._db.execute(
+            select(SubscriptionSearchBatch).where(SubscriptionSearchBatch.batch_id == batch_id)
+        ).scalars().first()
+
+    def _batch_cancel_requested(self, batch_id: str) -> bool:
+        """在当前事务中读取批次取消标记。"""
+        return bool(self._db.execute(
+            select(SubscriptionSearchBatch.cancel_requested).where(
+                SubscriptionSearchBatch.batch_id == batch_id
+            )
+        ).scalar())
+
+    def _refresh_batch(self, batch_id: str, *, now: str, error: Optional[str]) -> None:
+        """依据所属任务终态重新计算批次计数和聚合状态。"""
+        rows = self._db.execute(
+            select(SubscriptionSearchTask.state, func.count())  # pylint: disable=not-callable
+            .where(SubscriptionSearchTask.batch_id == batch_id)
+            .group_by(SubscriptionSearchTask.state)
+        ).all()
+        counts = {state: int(count) for state, count in rows}
+        completed = counts.get("completed", 0)
+        failed = counts.get("failed", 0)
+        cancelled = counts.get("cancelled", 0)
+        terminal = completed + failed + cancelled
+        batch = self.get_batch(batch_id)
+        if batch is None:
+            return
+        if terminal >= batch.total_count:
+            if failed:
+                state = "failed"
+            elif cancelled or batch.cancel_requested:
+                state = "cancelled"
+            else:
+                state = "completed"
+            finished_at = now
+        else:
+            state = "cancelling" if batch.cancel_requested else "running"
+            finished_at = None
+        execute_dml(
+            self._db,
+            update(SubscriptionSearchBatch)
+            .where(SubscriptionSearchBatch.id == batch.id)
+            .values(
+                state=state,
+                finished_count=completed,
+                failed_count=failed,
+                cancelled_count=cancelled,
+                updated_at=now,
+                finished_at=finished_at,
+                last_error=error or batch.last_error,
+            ),
+            execution_options={"synchronize_session": False},
+        )

app/scheduler/catalog.py

@@ -142,6 +142,13 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
                     "subscription",
                     kwargs={"state": "N"},
                 ),
+                JobSpec(
+                    "subscribe_search_queue",
+                    "恢复订阅搜索队列",
+                    services.resume_subscribe_search,
+                    "subscription",
+                    recovery=JobRecoveryPolicy.DURABLE_QUEUE,
+                ),
                 JobSpec("subscribe_refresh", "订阅刷新", services.refresh_subscribe, "subscription"),
                 JobSpec("subscribe_follow", "关注的订阅分享", services.follow_subscribe, "subscription"),
                 JobSpec(
@@ -246,6 +253,16 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
             )
 
         # 新增订阅时搜索（5分钟检查一次）
+        self._scheduler.add_job(
+            self.start,
+            "interval",
+            id="subscribe_search_queue",
+            name="恢复订阅搜索队列",
+            minutes=1,
+            next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=10),
+            kwargs={"job_id": "subscribe_search_queue"},
+        )
+
         self._scheduler.add_job(
             self.start,
             "interval",

app/scheduler/services.py

@@ -17,6 +17,7 @@ class SchedulerServices:
     sync_mediaserver: JobCallable
     check_subscribe: JobCallable
     search_subscribe: JobCallable
+    resume_subscribe_search: JobCallable
     refresh_subscribe: JobCallable
     follow_subscribe: JobCallable
     process_transfer: JobCallable

app/startup/composition/chain.py

@@ -91,6 +91,7 @@ def build_chain_runtime_context(
         module_dispatcher_factory=ModuleInvocationDispatcher,
         site_repository=dependencies.site,
         subscription_repository=dependencies.subscription,
+        subscription_search_repository=dependencies.subscription_search,
         subscription_mutation_scope=subscription_mutation_scope,
         sync_subscription_mutation_scope=sync_subscription_mutation_scope,
         subscription_delete_scope=delete_subscribe_scope,

app/startup/composition/runtime.py

@@ -48,6 +48,7 @@ from app.startup.composition.context import (
 )
 
 if TYPE_CHECKING:
+    from app.application.subscription.execution import SubscriptionSearchRepository
     from app.application.messaging.message import MessageHelper, MessageQueueManager
     from app.startup.composition.agent import AgentComposition
     from app.startup.composition.configuration import ConfigurationComposition
@@ -67,6 +68,7 @@ class RuntimeDependencies:
     transfer_execution: TransferExecutionRepository
     message_helper: MessageHelper
     message_queue: MessageQueueManager
+    subscription_search: SubscriptionSearchRepository | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -103,6 +105,7 @@ def compose_runtime_dependencies() -> RuntimeDependencies:
         TransactionalSubscriptionHistoryRepository,
         TransactionalSubscriptionRepository,
     )
+    from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository
     from app.db.adapters.transfer.execution import (
         TransactionalTransferExecutionRepository,
     )
@@ -132,6 +135,7 @@ def compose_runtime_dependencies() -> RuntimeDependencies:
         transfer_execution=TransactionalTransferExecutionRepository(SessionFactory),
         message_helper=message_helper_factory(),
         message_queue=MessageQueueManager(auto_start=False),
+        subscription_search=TransactionalSubscriptionSearchRepository(SessionFactory),
     )
 
 

app/startup/initializers/scheduler.py

@@ -61,6 +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,
             refresh_subscribe=subscribe_chain.refresh,
             follow_subscribe=subscribe_chain.follow,
             process_transfer=transfer_chain.process,

database/versions/c1f4a8d2e6b9_3_0_19.py

@@ -0,0 +1,104 @@
+"""3.0.19 增加订阅搜索持久批次与任务队列。
+
+Revision ID: c1f4a8d2e6b9
+Revises: a9d4f2c7e6b1
+Create Date: 2026-09-01
+"""
+
+# Alembic 的 op 是运行期代理，静态分析无法看到实际操作方法。
+# pylint: disable=no-member
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "c1f4a8d2e6b9"
+down_revision = "a9d4f2c7e6b1"
+branch_labels = None
+depends_on = None
+
+_BATCH_TABLE = "subscriptionsearchbatch"
+_TASK_TABLE = "subscriptionsearchtask"
+
+
+def _table_names() -> set[str]:
+    """返回当前数据库表名集合。"""
+    return set(sa.inspect(op.get_bind()).get_table_names())
+
+
+def upgrade() -> None:
+    """创建跨 SQLite/PostgreSQL 一致的订阅搜索治理表。"""
+    tables = _table_names()
+    if _BATCH_TABLE not in tables:
+        op.create_table(
+            _BATCH_TABLE,
+            sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+            sa.Column("batch_id", sa.String(length=64), nullable=False),
+            sa.Column("source", sa.String(length=32), nullable=False),
+            sa.Column("state", sa.String(length=32), nullable=False),
+            sa.Column("priority", sa.Integer(), nullable=False),
+            sa.Column("total_count", sa.Integer(), nullable=False),
+            sa.Column("finished_count", sa.Integer(), nullable=False),
+            sa.Column("failed_count", sa.Integer(), nullable=False),
+            sa.Column("cancelled_count", sa.Integer(), nullable=False),
+            sa.Column("cancel_requested", sa.Integer(), nullable=False),
+            sa.Column("created_at", sa.String(length=40), nullable=False),
+            sa.Column("updated_at", sa.String(length=40), nullable=False),
+            sa.Column("started_at", sa.String(length=40), nullable=True),
+            sa.Column("finished_at", sa.String(length=40), nullable=True),
+            sa.Column("last_error", sa.Text(), nullable=True),
+            sa.UniqueConstraint("batch_id", name="uq_subscriptionsearchbatch_batch_id"),
+        )
+        op.create_index(
+            "ix_subscriptionsearchbatch_state_created",
+            _BATCH_TABLE,
+            ["state", "created_at", "id"],
+        )
+    if _TASK_TABLE not in tables:
+        op.create_table(
+            _TASK_TABLE,
+            sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+            sa.Column("task_id", sa.String(length=64), nullable=False),
+            sa.Column("batch_id", sa.String(length=64), nullable=False),
+            sa.Column("subscription_id", sa.Integer(), nullable=False),
+            sa.Column("active_key", sa.String(length=128), nullable=True),
+            sa.Column("source", sa.String(length=32), nullable=False),
+            sa.Column("priority", sa.Integer(), nullable=False),
+            sa.Column("position", sa.Integer(), nullable=False),
+            sa.Column("state", sa.String(length=32), nullable=False),
+            sa.Column("attempt_count", sa.Integer(), nullable=False),
+            sa.Column("cancel_requested", sa.Integer(), nullable=False),
+            sa.Column("lease_owner", sa.String(length=128), nullable=True),
+            sa.Column("lease_token", sa.String(length=64), nullable=True),
+            sa.Column("lease_expires_at", sa.String(length=40), nullable=True),
+            sa.Column("created_at", sa.String(length=40), nullable=False),
+            sa.Column("updated_at", sa.String(length=40), nullable=False),
+            sa.Column("started_at", sa.String(length=40), nullable=True),
+            sa.Column("finished_at", sa.String(length=40), nullable=True),
+            sa.Column("last_error", sa.Text(), nullable=True),
+            sa.UniqueConstraint("task_id", name="uq_subscriptionsearchtask_task_id"),
+            sa.UniqueConstraint("active_key", name="uq_subscriptionsearchtask_active_key"),
+        )
+        op.create_index(
+            "ix_subscriptionsearchtask_claim",
+            _TASK_TABLE,
+            ["state", "priority", "lease_expires_at", "created_at", "id"],
+        )
+        op.create_index(
+            "ix_subscriptionsearchtask_batch_position",
+            _TASK_TABLE,
+            ["batch_id", "position", "id"],
+        )
+        op.create_index(
+            "ix_subscriptionsearchtask_subscription",
+            _TASK_TABLE,
+            ["subscription_id", "created_at", "id"],
+        )
+
+
+def downgrade() -> None:
+    """移除订阅搜索治理表。"""
+    tables = _table_names()
+    if _TASK_TABLE in tables:
+        op.drop_table(_TASK_TABLE)
+    if _BATCH_TABLE in tables:
+        op.drop_table(_BATCH_TABLE)

docs/architecture-overview.md

@@ -754,8 +754,8 @@ flowchart LR
 
 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 919 |
-| 内部导入边 | 7,690 |
+| Python 模块 | 925 |
+| 内部导入边 | 7,728 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 53（债务已清零，53 条精确 containment） |

docs/architecture/optimization-checklist.md

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 919 / 7,690 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 925 / 7,728 | `dependency-baseline.json` 当前快照 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

docs/refactor/moviepilot-subscription-governance-roadmap.md

@@ -1,7 +1,7 @@
 # MoviePilot 订阅执行治理
 
 > 状态：`active（2026-09-01 已由 MoviePilot v3 接管）`
-> 当前叶：`SUB-GOV-002A`
+> 当前叶：`SUB-GOV-002B`
 > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、
 > `V3-RDY-009A1C`、`V3-RDY-009A1D`
 > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环
@@ -232,14 +232,14 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
 | `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `completed（2026-09-01）` |
 | `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `completed（2026-09-01）` |
 | `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `not_activated（2026-09-01）` |
-| `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `in_progress` |
-| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `pending` |
+| `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `completed（2026-09-01）` |
+| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `in_progress` |
 | `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `pending` |
 | `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `pending` |
 | `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` |
 | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` |
 
-当前只激活 `SUB-GOV-002A`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
+当前只激活 `SUB-GOV-002B`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。
 
 ### 6.1 SUB-GOV-001A 验收证据
@@ -299,6 +299,21 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
   等待仍主导批次的同 revision 证据；因此不激活 2/4 worker，不引入共享客户端线程安全与提交排序风险；
 - 后续若受控生产形态证据再次显示准备等待占主导，可从本条件叶重新开启，但不得绕过 SiteBudget 和串行提交。
 
+### 6.6 SUB-GOV-002A 验收证据
+
+- 新增持久 `SubscriptionSearchBatch` 与 `SubscriptionSearchTask`，通过 Alembic `c1f4a8d2e6b9` 同时支持
+  SQLite/PostgreSQL；批次记录聚合终态，任务记录稳定位置、优先级、attempt、租约和取消标记；
+- 可空唯一 `active_key=subscription:<id>` 是订阅级 single-flight owner：手工/定向任务只提高已排队任务优先级，
+  不为同一订阅创建第二个活动任务；终态清空 active key，允许后续正常周期重新搜索；
+- 配置了正式队列的 Search 路径不再取得 Match 类级锁，也不再执行逐订阅 `60–300s` 随机休眠；单任务异常
+  独立标记 failed 并继续后续订阅，批次最终暴露聚合失败；
+- running 任务使用 token fencing 与 UTC 到期租约，进程重启后以相同 task identity、递增 attempt 恢复；停止时
+  可退回 queued，取消会立即终止未发请求任务并在运行任务返回租约边界时收口；
+- 新增 `subscribe_search_queue` durable 调度，启动后 10 秒首次恢复、之后每分钟消费，且使用独立消费者锁保持
+  单进程串行；它不创建 24 小时兜底批次，也不阻塞日常 Match；
+- 验证：队列/编排/调度专项 `43 passed`，迁移与声明式模型 `35 passed`，复杂度/并发/架构/迁移组合
+  `119 passed`，错误级 Pylint 为 0，Alembic 唯一 head 为 `c1f4a8d2e6b9`。
+
 ## 7. 上线前验证与验收
 
 ### 7.1 场景

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 7711,
-  "edge_sha256": "f1ff788de28fa0c486e529afa4dc7af5038587f997bf155c5fc434754ec2a4a3",
+  "edge_count": 7728,
+  "edge_sha256": "33cdb555e19d7a5d15dbed065c98908269c3528ec4d61c2ac5515420c0b4cfa4",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -4586,6 +4586,7 @@
     "app.chain.subscribe.search -> app.application.configuration",
     "app.chain.subscribe.search -> app.application.subscription",
     "app.chain.subscribe.search -> app.application.subscription.contract",
+    "app.chain.subscribe.search -> app.application.subscription.execution",
     "app.chain.subscribe.search -> app.application.subscription.query",
     "app.chain.subscribe.search -> app.chain",
     "app.chain.subscribe.search -> app.chain.media",
@@ -5130,6 +5131,15 @@
     "app.db.adapters.subscription -> app.schemas",
     "app.db.adapters.subscription -> app.schemas.common",
     "app.db.adapters.subscription -> app.schemas.types",
+    "app.db.adapters.subscriptionsearch -> app.application",
+    "app.db.adapters.subscriptionsearch -> app.application.subscription",
+    "app.db.adapters.subscriptionsearch -> app.application.subscription.execution",
+    "app.db.adapters.subscriptionsearch -> app.db",
+    "app.db.adapters.subscriptionsearch -> app.db.models",
+    "app.db.adapters.subscriptionsearch -> app.db.models.subscriptionsearch",
+    "app.db.adapters.subscriptionsearch -> app.db.oper",
+    "app.db.adapters.subscriptionsearch -> app.db.oper.subscriptionsearch",
+    "app.db.adapters.subscriptionsearch -> app.db.uow",
     "app.db.adapters.transaction -> app.db",
     "app.db.adapters.transaction -> app.db.uow",
     "app.db.adapters.transfer.admission -> app.application",
@@ -5268,6 +5278,8 @@
     "app.db.models.subscribehistory -> app.db.models._constraints",
     "app.db.models.subscribehistory -> app.schemas",
     "app.db.models.subscribehistory -> app.schemas.types",
+    "app.db.models.subscriptionsearch -> app.db",
+    "app.db.models.subscriptionsearch -> app.db.base",
     "app.db.models.systemconfig -> app.db",
     "app.db.models.systemconfig -> app.db.base",
     "app.db.models.transferexecutionstep -> app.db",
@@ -5374,6 +5386,10 @@
     "app.db.oper.subscribehistory -> app.schemas.common",
     "app.db.oper.subscribehistory -> app.schemas.query",
     "app.db.oper.subscribehistory -> app.schemas.types",
+    "app.db.oper.subscriptionsearch -> app.db",
+    "app.db.oper.subscriptionsearch -> app.db.base",
+    "app.db.oper.subscriptionsearch -> app.db.models",
+    "app.db.oper.subscriptionsearch -> app.db.models.subscriptionsearch",
     "app.db.oper.systemconfig -> app.db",
     "app.db.oper.systemconfig -> app.db.base",
     "app.db.oper.systemconfig -> app.db.models",
@@ -8216,6 +8232,7 @@
     "app.startup.composition.runtime -> app.db.adapters.outbox",
     "app.startup.composition.runtime -> app.db.adapters.site",
     "app.startup.composition.runtime -> app.db.adapters.subscription",
+    "app.startup.composition.runtime -> app.db.adapters.subscriptionsearch",
     "app.startup.composition.runtime -> app.db.adapters.transfer",
     "app.startup.composition.runtime -> app.db.adapters.transfer.execution",
     "app.startup.composition.runtime -> app.db.oper",
@@ -8804,7 +8821,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 921,
+  "module_count": 925,
   "modules": [
     "app",
     "app.adapters",
@@ -9095,6 +9112,7 @@
     "app.application.subscription.complete",
     "app.application.subscription.contract",
     "app.application.subscription.delete",
+    "app.application.subscription.execution",
     "app.application.subscription.facts",
     "app.application.subscription.identity",
     "app.application.subscription.mutation",
@@ -9229,6 +9247,7 @@
     "app.db.adapters.query",
     "app.db.adapters.site",
     "app.db.adapters.subscription",
+    "app.db.adapters.subscriptionsearch",
     "app.db.adapters.transaction",
     "app.db.adapters.transfer",
     "app.db.adapters.transfer.admission",
@@ -9262,6 +9281,7 @@
     "app.db.models.siteuserdata",
     "app.db.models.subscribe",
     "app.db.models.subscribehistory",
+    "app.db.models.subscriptionsearch",
     "app.db.models.systemconfig",
     "app.db.models.transferexecutionstep",
     "app.db.models.transferhistory",
@@ -9284,6 +9304,7 @@
     "app.db.oper.site",
     "app.db.oper.subscribe",
     "app.db.oper.subscribehistory",
+    "app.db.oper.subscriptionsearch",
     "app.db.oper.systemconfig",
     "app.db.oper.transferexecutionstep",
     "app.db.oper.transferhistory",

tests/test_subscription_search_governance.py

@@ -0,0 +1,116 @@
+"""订阅搜索队列接入、锁隔离和批次失败治理测试。"""
+
+from dataclasses import replace
+from datetime import datetime, timedelta
+from unittest.mock import Mock, patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.application.subscription.contract import SubscriptionSnapshot
+from app.chain.subscribe.facade import SubscribeChain
+from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository
+from app.db.base import Base
+from app.schemas.types import MediaType
+
+
+class _ForbiddenLock:
+    """正式队列路径若仍访问 Match 全局锁则立即失败。"""
+
+    def acquire(self, **_kwargs):
+        """禁止搜索队列取得历史长锁。"""
+        raise AssertionError("持久搜索队列不应取得 Match 全局锁")
+
+    def release(self):
+        """禁止搜索队列释放从未取得的历史长锁。"""
+        raise AssertionError("持久搜索队列不应释放 Match 全局锁")
+
+
+class _SubscriptionRepository:
+    """为搜索队列返回稳定的多订阅快照。"""
+
+    def __init__(self, subscribes: list[SubscriptionSnapshot]) -> None:
+        self._subscribes = {subscribe.id: subscribe for subscribe in subscribes}
+
+    def list(self, _state: str = None) -> list[SubscriptionSnapshot]:
+        """返回全部测试订阅。"""
+        return list(self._subscribes.values())
+
+    def get(self, subscribe_id: int) -> SubscriptionSnapshot | None:
+        """按 ID 返回任务执行时重新读取的订阅。"""
+        return self._subscribes.get(subscribe_id)
+
+
+def _subscribe(subscribe_id: int) -> SubscriptionSnapshot:
+    """构造越过新增保护期的活动电影订阅。"""
+    return SubscriptionSnapshot(
+        id=subscribe_id,
+        name=f"治理电影 {subscribe_id}",
+        year="2026",
+        type=MediaType.MOVIE.value,
+        media_source="themoviedb",
+        media_id=str(1000 + subscribe_id),
+        state="R",
+        date=(datetime.now() - timedelta(minutes=2)).strftime("%Y-%m-%d %H:%M:%S"),
+    )
+
+
+def _chain(tmp_path, subscribes: list[SubscriptionSnapshot]):
+    """构造注入持久队列且不初始化其他 Chain 依赖的搜索实例。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'search-governance.db'}")
+    Base.metadata.create_all(engine)
+    chain = object.__new__(SubscribeChain)
+    chain.subscription_repository = _SubscriptionRepository(subscribes)
+    chain.subscription_search_repository = TransactionalSubscriptionSearchRepository(
+        sessionmaker(bind=engine)
+    )
+    chain.get_states_for_search = lambda state: state
+    chain._rlock = _ForbiddenLock()
+    return chain
+
+
+def test_fallback_queue_executes_without_match_global_lock(tmp_path, monkeypatch):
+    """R/P 兜底搜索在持久队列中执行，不受日常 Match 长锁阻塞。"""
+    subscribes = [_subscribe(1), _subscribe(2)]
+    chain = _chain(tmp_path, subscribes)
+    processed = []
+    monkeypatch.setattr(
+        chain,
+        "_process_search_subscription",
+        lambda subscribe, _searchchain: processed.append(subscribe.id) or subscribe,
+    )
+
+    with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()):
+        batch_id = chain.search(state="R")
+
+    batch = chain.get_search_batch(batch_id)
+    assert processed == [1, 2]
+    assert batch.state == "completed"
+    assert batch.finished_count == 2
+    assert batch.failed_count == 0
+
+
+def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monkeypatch):
+    """单订阅异常不得中止批次后续任务，聚合终态必须暴露失败。"""
+    subscribes = [_subscribe(3), _subscribe(4)]
+    chain = _chain(tmp_path, subscribes)
+    processed = []
+
+    def process(subscribe, _searchchain):
+        """让首条失败并保持第二条正常完成。"""
+        processed.append(subscribe.id)
+        if subscribe.id == 3:
+            raise RuntimeError("provider timeout")
+        return replace(subscribe)
+
+    monkeypatch.setattr(chain, "_process_search_subscription", process)
+
+    with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()):
+        batch_id = chain.search(state="R")
+
+    batch = chain.get_search_batch(batch_id)
+    assert processed == [3, 4]
+    assert batch.state == "failed"
+    assert batch.finished_count == 1
+    assert batch.failed_count == 1
+    assert batch.last_error == "provider timeout"

tests/test_subscription_search_queue.py

@@ -0,0 +1,132 @@
+"""订阅搜索持久队列、single-flight、租约和取消测试。"""
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import create_engine, select, update
+from sqlalchemy.orm import Session, sessionmaker
+
+from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository
+from app.db.base import Base
+from app.db.models.subscriptionsearch import SubscriptionSearchTask
+
+
+def _repository(tmp_path):
+    """构造使用独立 SQLite 文件的事务型搜索队列。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'search-queue.db'}")
+    Base.metadata.create_all(engine)
+    return TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)), engine
+
+
+def test_search_queue_coalesces_active_subscription_and_raises_priority(tmp_path):
+    """重叠入口只保留一个活动任务，手工请求可提高优先级。"""
+    repository, _engine = _repository(tmp_path)
+
+    scheduled = repository.enqueue(
+        subscription_ids=(1, 2),
+        source="fallback",
+        priority=10,
+    )
+    manual = repository.enqueue(
+        subscription_ids=(1,),
+        source="manual",
+        priority=100,
+    )
+
+    first = repository.claim_next(owner="worker-a")
+    second = repository.claim_next(owner="worker-b")
+
+    assert scheduled.created_count == 2
+    assert scheduled.coalesced_count == 0
+    assert manual.created_count == 0
+    assert manual.coalesced_count == 1
+    assert manual.batch.state == "completed"
+    assert first.subscription_id == 1
+    assert first.priority == 100
+    assert second.subscription_id == 2
+    assert first.task_id != second.task_id
+
+
+def test_search_queue_recovers_expired_lease_with_same_task_identity(tmp_path):
+    """进程遗留的过期 running 任务应以新 token 恢复且 attempt 单调递增。"""
+    repository, engine = _repository(tmp_path)
+    repository.enqueue(subscription_ids=(3,), source="fallback", priority=10)
+    first = repository.claim_next(owner="worker-a", lease_seconds=900)
+    expired_at = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat(timespec="seconds")
+    with Session(engine) as session:
+        session.execute(
+            update(SubscriptionSearchTask)
+            .where(SubscriptionSearchTask.task_id == first.task_id)
+            .values(lease_expires_at=expired_at)
+        )
+        session.commit()
+
+    recovered = repository.claim_next(owner="worker-b", lease_seconds=900)
+
+    assert recovered.task_id == first.task_id
+    assert recovered.subscription_id == first.subscription_id
+    assert recovered.lease_token != first.lease_token
+    assert recovered.attempt_count == 2
+
+
+def test_search_queue_cancel_finishes_queued_and_running_tasks(tmp_path):
+    """取消立即终止未发请求任务，运行中任务在租约边界收口。"""
+    repository, engine = _repository(tmp_path)
+    enqueued = repository.enqueue(
+        subscription_ids=(4, 5),
+        source="fallback",
+        priority=10,
+    )
+    running = repository.claim_next(owner="worker-a")
+
+    assert repository.request_cancel(enqueued.batch.batch_id) is True
+    assert repository.is_cancel_requested(running.task_id) is True
+    assert repository.release_task(
+        task_id=running.task_id,
+        lease_token=running.lease_token,
+        cancelled=True,
+    ) is True
+
+    batch = repository.get_batch(enqueued.batch.batch_id)
+    with Session(engine) as session:
+        states = list(
+            session.execute(
+                select(SubscriptionSearchTask.state)
+                .where(SubscriptionSearchTask.batch_id == enqueued.batch.batch_id)
+                .order_by(SubscriptionSearchTask.position)
+            ).scalars()
+        )
+
+    assert states == ["cancelled", "cancelled"]
+    assert batch.state == "cancelled"
+    assert batch.cancelled_count == 2
+    assert repository.claim_next(owner="worker-b") is None
+
+
+def test_search_queue_finishes_batch_with_aggregated_failure(tmp_path):
+    """单任务失败不阻止后续任务，但批次最终暴露聚合失败。"""
+    repository, _engine = _repository(tmp_path)
+    enqueued = repository.enqueue(
+        subscription_ids=(6, 7),
+        source="fallback",
+        priority=10,
+    )
+    first = repository.claim_next(owner="worker-a")
+    assert repository.finish_task(
+        task_id=first.task_id,
+        lease_token=first.lease_token,
+        state="failed",
+        error="site timeout",
+    ) is True
+    second = repository.claim_next(owner="worker-a")
+    assert repository.finish_task(
+        task_id=second.task_id,
+        lease_token=second.lease_token,
+        state="completed",
+    ) is True
+
+    batch = repository.get_batch(enqueued.batch.batch_id)
+
+    assert batch.state == "failed"
+    assert batch.finished_count == 1
+    assert batch.failed_count == 1
+    assert batch.last_error == "site timeout"