refactor(subscribe): govern fallback site pressure

app/application/site/search_observation.py

@@ -0,0 +1,50 @@
+"""站点搜索调用的轻量结果观察上下文。"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from typing import Iterator, Optional
+
+
+@dataclass(slots=True)
+class SiteSearchObservation:
+    """记录一次站点搜索是否发出请求及其可治理结果。"""
+
+    attempted: bool = False
+    outcome: str = "skipped"
+    error: Optional[str] = None
+
+
+_current_observation: ContextVar[Optional[SiteSearchObservation]] = ContextVar(
+    "site_search_observation",
+    default=None,
+)
+
+
+@contextmanager
+def capture_site_search_observation() -> Iterator[SiteSearchObservation]:
+    """为当前同步 worker 捕获一次索引器搜索结果。"""
+    observation = SiteSearchObservation()
+    token = _current_observation.set(observation)
+    try:
+        yield observation
+    finally:
+        _current_observation.reset(token)
+
+
+def report_site_search_outcome(
+    *,
+    attempted: bool,
+    outcome: str,
+    error: Optional[str] = None,
+) -> None:
+    """由索引器在不改变公开返回合同的前提下发布调用结果。"""
+    observation = _current_observation.get()
+    if observation is None:
+        return
+    observation.attempted = attempted
+    observation.outcome = outcome
+    observation.error = error
+

app/application/subscription/execution.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from typing import Optional, Protocol
 
+from app.application.subscription.sitebudget import SiteBudgetClaim
+
 
 @dataclass(frozen=True, slots=True)
 class SearchBatchSnapshot:
@@ -40,6 +42,7 @@ class SearchTaskSnapshot:
     lease_token: Optional[str]
     created_at: str
     updated_at: str
+    available_at: Optional[str] = None
     started_at: Optional[str] = None
     finished_at: Optional[str] = None
     last_error: Optional[str] = None
@@ -63,8 +66,9 @@ class SubscriptionSearchRepository(Protocol):
         subscription_ids: tuple[int, ...],
         source: str,
         priority: int,
+        available_at: Optional[str] = None,
     ) -> SearchEnqueueResult:
-        """按订阅 ID 建立批次，并合并已存在的活动任务。"""
+        """按订阅 ID 建立批次，在启动抖动后合并活动任务。"""
         ...
 
     def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]:
@@ -103,3 +107,25 @@ class SubscriptionSearchRepository(Protocol):
     def get_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]:
         """按稳定批次 ID 返回当前聚合状态。"""
         ...
+
+    def claim_site(
+        self,
+        *,
+        site_id: int,
+        owner: str,
+        lease_seconds: int,
+    ) -> SiteBudgetClaim:
+        """认领一个站点的唯一在途搜索预算。"""
+        ...
+
+    def finish_site(
+        self,
+        *,
+        site_id: int,
+        lease_token: str,
+        outcome: str,
+        next_allowed_at: str,
+        error: Optional[str] = None,
+    ) -> bool:
+        """释放站点预算并写入间隔、冷却和恢复状态。"""
+        ...

app/application/subscription/sitebudget.py

@@ -0,0 +1,156 @@
+"""订阅兜底搜索的站点级容量、间隔与冷却治理。"""
+
+from __future__ import annotations
+
+import random
+import time
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Callable, Optional, Protocol
+
+from app.application.site.search_observation import SiteSearchObservation
+from app.runtime.stop import StopState
+
+
+class SubscriptionSearchCancelled(RuntimeError):
+    """表示订阅搜索在可取消预算等待点终止。"""
+
+
+class SubscriptionSiteBudgetUnavailable(RuntimeError):
+    """表示站点预算超出本任务允许的短等待窗口。"""
+
+    def __init__(self, *, site_id: int, retry_at: str) -> None:
+        """保存站点和下一次可尝试时间，供批次聚合失败展示。"""
+        super().__init__(f"站点 {site_id} 冷却或已有在途搜索，最早可重试：{retry_at}")
+        self.site_id = site_id
+        self.retry_at = retry_at
+
+
+@dataclass(frozen=True, slots=True)
+class SiteBudgetClaim:
+    """一次站点预算认领结果或下一次可尝试时间。"""
+
+    site_id: int
+    acquired: bool
+    retry_at: str
+    consecutive_failures: int
+    lease_token: Optional[str] = None
+
+
+class SubscriptionSiteBudgetRepository(Protocol):
+    """站点预算租约和冷却状态的持久化端口。"""
+
+    def claim_site(
+        self,
+        *,
+        site_id: int,
+        owner: str,
+        lease_seconds: int,
+    ) -> SiteBudgetClaim:
+        """认领站点唯一在途租约，未就绪时返回重试时间。"""
+        ...
+
+    def finish_site(
+        self,
+        *,
+        site_id: int,
+        lease_token: str,
+        outcome: str,
+        next_allowed_at: str,
+        error: Optional[str] = None,
+    ) -> bool:
+        """收口当前站点租约并持久化下次允许时间。"""
+        ...
+
+
+def _utc_now() -> datetime:
+    """返回带时区的 UTC 当前时间。"""
+    return datetime.now(timezone.utc)
+
+
+class SubscriptionSiteBudget:
+    """在同步搜索 worker 内执行可取消的站点预算等待与反馈。"""
+
+    def __init__(
+        self,
+        *,
+        repository: SubscriptionSiteBudgetRepository,
+        owner: str,
+        cancelled: Callable[[], bool],
+        stop_state: StopState,
+        interval_range: tuple[float, float] = (60.0, 300.0),
+        lease_seconds: int = 900,
+        max_wait_seconds: float = 5.0,
+        random_uniform: Callable[[float, float], float] = random.uniform,
+        sleeper: Callable[[float], None] = time.sleep,
+        clock: Callable[[], datetime] = _utc_now,
+    ) -> None:
+        """保存持久化端口及可注入的时钟、随机数和等待实现。"""
+        self._repository = repository
+        self._owner = owner
+        self._cancelled = cancelled
+        self._stop_state = stop_state
+        self._interval_range = interval_range
+        self._lease_seconds = max(1, lease_seconds)
+        self._max_wait_seconds = max(0.0, max_wait_seconds)
+        self._random_uniform = random_uniform
+        self._sleeper = sleeper
+        self._clock = clock
+
+    def acquire(self, site_id: int) -> SiteBudgetClaim:
+        """循环认领指定站点，并在每秒边界检查取消与停机。"""
+        deadline = time.monotonic() + self._max_wait_seconds
+        while True:
+            self._raise_if_cancelled()
+            claim = self._repository.claim_site(
+                site_id=site_id,
+                owner=self._owner,
+                lease_seconds=self._lease_seconds,
+            )
+            if claim.acquired:
+                return claim
+            retry_at = datetime.fromisoformat(claim.retry_at)
+            remaining = max(0.0, (retry_at - self._clock()).total_seconds())
+            if remaining > max(0.0, deadline - time.monotonic()):
+                raise SubscriptionSiteBudgetUnavailable(
+                    site_id=site_id,
+                    retry_at=claim.retry_at,
+                )
+            self._sleeper(min(max(remaining, 0.05), 1.0))
+
+    def finish(self, claim: SiteBudgetClaim, observation: SiteSearchObservation) -> bool:
+        """依据调用结果计算随机间隔或错误冷却并释放租约。"""
+        if not claim.lease_token:
+            return False
+        outcome = observation.outcome if observation.attempted else "skipped"
+        delay = self._next_delay(outcome, claim.consecutive_failures)
+        next_allowed_at = (self._clock() + timedelta(seconds=delay)).isoformat(timespec="seconds")
+        return self._repository.finish_site(
+            site_id=claim.site_id,
+            lease_token=claim.lease_token,
+            outcome=outcome,
+            next_allowed_at=next_allowed_at,
+            error=observation.error,
+        )
+
+    def _next_delay(self, outcome: str, consecutive_failures: int) -> float:
+        """为成功渐进恢复、错误退避和本地跳过计算下一次等待。"""
+        if outcome == "skipped":
+            return 0.0
+        if outcome == "success":
+            low, high = self._interval_range
+            recovery_factor = 1.0 + min(max(consecutive_failures, 0), 3) * 0.5
+            return self._random_uniform(low, high) * recovery_factor
+        exponent = min(max(consecutive_failures, 0), 5)
+        base, ceiling = {
+            "rate_limited": (900.0, 21600.0),
+            "forbidden": (900.0, 21600.0),
+            "login_invalid": (900.0, 21600.0),
+            "timeout": (300.0, 7200.0),
+        }.get(outcome, (180.0, 3600.0))
+        return min(base * (2**exponent), ceiling)
+
+    def _raise_if_cancelled(self) -> None:
+        """在不持有业务锁的等待边界传播取消或停机。"""
+        if self._stop_state.is_system_stopped or self._cancelled():
+            raise SubscriptionSearchCancelled("订阅搜索已取消")

app/chain/search/contract.py

@@ -63,6 +63,7 @@ if TYPE_CHECKING:
         _restore_original_indices: Callable[..., Any]
         _save_results: Callable[..., Any]
         _search_all_sites: Callable[..., Any]
+        _search_site_torrents_with_budget: Callable[..., Any]
         _search_state: Callable[..., Any]
         _select_indexers: Callable[..., Any]
         _selected_site_ids: Callable[..., Any]
@@ -96,6 +97,8 @@ if TYPE_CHECKING:
         matches_music_resource: Callable[..., Any]
         music_site_keywords: Callable[..., Any]
         process: Callable[..., Any]
+        record_subscription_site_budget_failure: Callable[..., Any]
+        consume_subscription_site_budget_failures: Callable[..., Any]
         save_last_search_params: Callable[..., Any]
         search_by_id: Callable[..., Any]
         search_by_title: Callable[..., Any]

app/chain/search/facade.py

@@ -1,5 +1,6 @@
 """搜索处理链稳定 Facade。"""
 
+import threading
 from collections.abc import AsyncIterator, Callable
 from typing import Any, Optional, TypeVar, cast
 
@@ -15,6 +16,7 @@ from app.chain.search.result import SearchResultOwner
 from app.chain.search.site import SearchSiteOwner
 from app.chain.search.subtitle import SearchSubtitleOwner
 from app.chain.search.title import SearchTitleOwner
+from app.application.subscription.sitebudget import SubscriptionSiteBudget
 from app.domain.context import Context, MediaInfo, SubtitleInfo
 from app.runtime.events import Event, eventmanager
 from app.schemas.mediaserver import NotExistMediaInfo
@@ -40,6 +42,33 @@ class SearchChain(ChainBase):
     _SEARCH_PARAMS_CACHE_KEY = "__search_params__"
     _AI_INDICES_CACHE_KEY = "__ai_recommend_indices__"
 
+    def configure_subscription_site_budget(
+        self,
+        budget: Optional[SubscriptionSiteBudget],
+    ) -> None:
+        """仅为订阅搜索启用或清除站点预算，不影响其它搜索入口。"""
+        self._subscription_site_budget = budget
+        self._subscription_site_budget_failures: list[str] = []
+        self._subscription_site_budget_failure_lock = threading.Lock()
+
+    def record_subscription_site_budget_failure(self, error: str) -> None:
+        """线程安全地记录一个未执行站点，供订阅任务暴露聚合失败。"""
+        lock = getattr(self, "_subscription_site_budget_failure_lock", None)
+        if lock is None:
+            return
+        with lock:
+            self._subscription_site_budget_failures.append(error)
+
+    def consume_subscription_site_budget_failures(self) -> tuple[str, ...]:
+        """读取并清空当前订阅搜索积累的站点预算失败。"""
+        lock = getattr(self, "_subscription_site_budget_failure_lock", None)
+        if lock is None:
+            return ()
+        with lock:
+            failures = tuple(self._subscription_site_budget_failures)
+            self._subscription_site_budget_failures.clear()
+        return failures
+
     # owner descriptor 经类访问后被 mypy 视为普通 Callable；运行时仍需取回原始
     # classmethod 函数，才能保持 SearchChain 的直接 MRO 与既有绑定语义。
     music_site_keywords = classmethod(  # type: ignore[var-annotated]
@@ -456,6 +485,7 @@ class SearchChain(ChainBase):
     _async_indexers = SearchProviderOwner._async_indexers
     _torrent_keyword = staticmethod(SearchProviderOwner._torrent_keyword)
     _torrent_type = staticmethod(SearchProviderOwner._torrent_type)
+    _search_site_torrents_with_budget = SearchProviderOwner._search_site_torrents_with_budget
     _iter_provider_batches = SearchProviderOwner._iter_provider_batches
     _iter_provider_events = SearchProviderOwner._iter_provider_events
     _iter_torrent_events = SearchProviderOwner._iter_torrent_events

app/chain/search/provider.py

@@ -10,7 +10,15 @@ from datetime import datetime
 from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Iterator, List, Optional
 
 from app.application.configuration import get_configured_system_config
+from app.application.site.search_observation import (
+    capture_site_search_observation,
+    report_site_search_outcome,
+)
 from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
+from app.application.subscription.sitebudget import (
+    SubscriptionSiteBudget,
+    SubscriptionSiteBudgetUnavailable,
+)
 from app.chain.search.contract import _SearchOwnerBase
 from app.domain.context import MediaInfo, SubtitleInfo, TorrentInfo
 from app.runtime.log import logger
@@ -182,13 +190,64 @@ class SearchProviderOwner(_SearchOwnerBase):
         """向进程共享线程 owner 提交一页，并登记该站点的续页位置。"""
         page_number = search_pages[page_index]
         future = ThreadHelper().submit(
-            _search_site_page,
+            self._search_site_torrents_with_budget,
-            self,
+            site=site,
-            site=site, keyword=search_keyword,
+            keyword=search_keyword,
-            media_type=media_type, page=page_number,
+            mtype=media_type,
+            page=page_number,
         )
         pending[future] = (site, page_index, page_number)
 
+    def _search_site_torrents_with_budget(
+        self,
+        *,
+        site: SiteIndexer,
+        keyword: str,
+        mtype: Optional[MediaType],
+        page: int,
+    ) -> List[TorrentInfo]:
+        """在订阅专属站点预算内执行一页同步搜索。"""
+        budget = getattr(self, "_subscription_site_budget", None)
+        site_id = site.get("id")
+        if not isinstance(budget, SubscriptionSiteBudget) or not isinstance(site_id, int):
+            return _search_site_page(
+                self,
+                site=site,
+                keyword=keyword,
+                media_type=mtype,
+                page=page,
+            )
+        try:
+            claim = budget.acquire(site_id)
+        except SubscriptionSiteBudgetUnavailable as error:
+            self.record_subscription_site_budget_failure(str(error))
+            logger.info(str(error))
+            return []
+        with capture_site_search_observation() as observation:
+            try:
+                return _search_site_page(
+                    self,
+                    site=site,
+                    keyword=keyword,
+                    media_type=mtype,
+                    page=page,
+                )
+            except Exception as error:
+                report_site_search_outcome(
+                    attempted=True,
+                    outcome="error",
+                    error=str(error),
+                )
+                raise
+            finally:
+                try:
+                    budget.finish(claim, observation)
+                except Exception as error:  # noqa: BLE001
+                    logger.error(
+                        f"站点 {site.get('name') or site_id} 搜索预算收口失败：{str(error)}",
+                        exc_info=True,
+                    )
+
     def _submit_next_sync_site(
         self,
         *,

app/chain/subscribe/search.py

@@ -2,7 +2,7 @@
 
 import random
 import time
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, Optional, cast
 from uuid import uuid4
 
@@ -15,6 +15,10 @@ from app.application.subscription.contract import (
 )
 from app.application.subscription.query import SubscriptionQueryService
 from app.application.subscription.execution import SearchBatchSnapshot, SubscriptionSearchRepository
+from app.application.subscription.sitebudget import (
+    SubscriptionSearchCancelled,
+    SubscriptionSiteBudget,
+)
 from app.chain.media import MediaChain
 from app.chain.search.facade import SearchChain
 from app.chain.subscribe.contract import _SubscribeOwnerBase
@@ -209,6 +213,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
             subscription_ids=tuple(subscribe.id for subscribe in subscribes),
             source=source,
             priority=priority,
+            available_at=self._search_batch_available_at(source),
         )
         total = len(subscribes)
         if progress_callback:
@@ -293,6 +298,14 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                     )
                     continue
                 current = subscribe
+                searchchain.configure_subscription_site_budget(
+                    SubscriptionSiteBudget(
+                        repository=queue,
+                        owner=f"{owner}:{task.task_id}",
+                        cancelled=lambda task_id=task.task_id: queue.is_cancel_requested(task_id),
+                        stop_state=getattr(self, "stop_state", runtime_stop_state),
+                    )
+                )
                 try:
                     current = self._process_search_subscription(subscribe, searchchain)
                     if queue.is_cancel_requested(task.task_id):
@@ -308,6 +321,12 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                             state="completed",
                         )
                     processed.add(subscribe.id)
+                except SubscriptionSearchCancelled:
+                    queue.release_task(
+                        task_id=task.task_id,
+                        lease_token=task.lease_token,
+                        cancelled=True,
+                    )
                 except Exception as err:
                     logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True)
                     queue.finish_task(
@@ -317,6 +336,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                         error=str(err),
                     )
                 finally:
+                    searchchain.configure_subscription_site_budget(None)
                     if current and current.state == "N":
                         try:
                             self._SubscribeChain__apply_subscribe_update(
@@ -379,8 +399,18 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
             return "订阅搜索完成，部分任务失败"
         if batch.state == "cancelled":
             return "订阅搜索已取消"
+        if batch.state in {"queued", "running", "cancelling"}:
+            return "订阅搜索任务已排队"
         return "订阅搜索完成"
 
+    @staticmethod
+    def _search_batch_available_at(source: str) -> str:
+        """为自动兜底批次持久化一次全局启动抖动。"""
+        delay = random.randint(0, 60) if source == "fallback" else 0
+        return (
+            datetime.now(timezone.utc) + timedelta(seconds=delay)
+        ).isoformat(timespec="seconds")
+
     def cancel_search_batch(self, batch_id: str) -> bool:
         """请求取消持久搜索批次；未注入队列时返回失败。"""
         queue = cast(
@@ -491,6 +521,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
             custom_words=subscribe.custom_words.split("\n") if subscribe.custom_words else None,
             filter_params=self.get_params(subscribe),
         )
+        site_budget_failures = searchchain.consume_subscription_site_budget_failures()
         if not contexts:
             logger.warning(f"订阅 {subscribe.keyword or subscribe.name} 未搜索到资源")
             self.finish_subscribe_or_not(
@@ -499,11 +530,13 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                 mediainfo=mediainfo,
                 lefts=no_exists,
             )
+            self._raise_site_budget_failures(site_budget_failures)
             return subscribe
         matched = self._filter_search_contexts(subscribe, contexts)
         if not matched:
             logger.warning(f"订阅 {subscribe.name} 没有符合过滤条件的资源")
             self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo, lefts=no_exists)
+            self._raise_site_budget_failures(site_budget_failures)
             return subscribe
         downloads, lefts = self._SubscribeChain__download_best_version_with_full_pack_first(
             contexts=matched,
@@ -524,8 +557,15 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                 downloads=downloads,
                 lefts=lefts,
             )
+        self._raise_site_budget_failures(site_budget_failures)
         return cast(Optional[SubscriptionSnapshot], current)
 
+    @staticmethod
+    def _raise_site_budget_failures(failures: tuple[str, ...]) -> None:
+        """在成功站点结果完成处理后暴露未执行站点的聚合失败。"""
+        if failures:
+            raise RuntimeError("；".join(failures))
+
     def _filter_search_contexts(
         self,
         subscribe: SubscriptionSnapshot,

app/db/adapters/subscriptionsearch.py

@@ -10,7 +10,9 @@ from app.application.subscription.execution import (
     SearchEnqueueResult,
     SearchTaskSnapshot,
 )
+from app.application.subscription.sitebudget import SiteBudgetClaim
 from app.db.models.subscriptionsearch import (
+    SubscriptionSiteBudget,
     SubscriptionSearchBatch,
     SubscriptionSearchTask,
 )
@@ -55,6 +57,7 @@ def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot:
         lease_token=record.lease_token,
         created_at=record.created_at,
         updated_at=record.updated_at,
+        available_at=record.available_at,
         started_at=record.started_at,
         finished_at=record.finished_at,
         last_error=record.last_error,
@@ -91,6 +94,7 @@ class TransactionalSubscriptionSearchRepository:
         subscription_ids: tuple[int, ...],
         source: str,
         priority: int,
+        available_at: Optional[str] = None,
     ) -> SearchEnqueueResult:
         """创建批次并返回 single-flight 合并计数。"""
         def operation(repository: SubscriptionSearchOper) -> SearchEnqueueResult:
@@ -99,6 +103,7 @@ class TransactionalSubscriptionSearchRepository:
                 subscription_ids=subscription_ids,
                 source=source,
                 priority=priority,
+                available_at=available_at,
             )
             return SearchEnqueueResult(
                 batch=_batch(record),
@@ -167,3 +172,53 @@ class TransactionalSubscriptionSearchRepository:
                 _batch(record) if (record := repository.get_batch(batch_id)) is not None else None
             )
         )
+
+    def claim_site(
+        self,
+        *,
+        site_id: int,
+        owner: str,
+        lease_seconds: int,
+    ) -> SiteBudgetClaim:
+        """认领单站点预算并投影等待或租约事实。"""
+        def operation(repository: SubscriptionSearchOper) -> SiteBudgetClaim:
+            """在短事务中认领并复制站点预算状态。"""
+            record, acquired = repository.claim_site(
+                site_id=site_id,
+                owner=owner,
+                lease_seconds=lease_seconds,
+            )
+            retry_at = (
+                record.lease_expires_at
+                if record.lease_token and not acquired
+                else record.next_allowed_at
+            ) or record.next_allowed_at
+            return SiteBudgetClaim(
+                site_id=record.site_id,
+                acquired=acquired,
+                retry_at=retry_at,
+                consecutive_failures=record.consecutive_failures,
+                lease_token=record.lease_token if acquired else None,
+            )
+
+        return self._write(operation)
+
+    def finish_site(
+        self,
+        *,
+        site_id: int,
+        lease_token: str,
+        outcome: str,
+        next_allowed_at: str,
+        error: Optional[str] = None,
+    ) -> bool:
+        """释放站点租约并持久化间隔或冷却。"""
+        return self._write(
+            lambda repository: repository.finish_site(
+                site_id=site_id,
+                lease_token=lease_token,
+                outcome=outcome,
+                next_allowed_at=next_allowed_at,
+                error=error,
+            )
+        )

app/db/models/__init__.py

@@ -43,6 +43,10 @@ _MODEL_EXPORTS = {
         "app.db.models.subscriptionsearch",
         "SubscriptionSearchTask",
     ),
+    "SubscriptionSiteBudget": (
+        "app.db.models.subscriptionsearch",
+        "SubscriptionSiteBudget",
+    ),
     "SystemConfig": ("app.db.models.systemconfig", "SystemConfig"),
     "TransferHistory": ("app.db.models.transferhistory", "TransferHistory"),
     "TransferExecutionStep": (

app/db/models/subscriptionsearch.py

@@ -50,6 +50,7 @@ class SubscriptionSearchTask(Base):
     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))
+    available_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))
@@ -63,6 +64,7 @@ class SubscriptionSearchTask(Base):
             "ix_subscriptionsearchtask_claim",
             "state",
             "priority",
+            "available_at",
             "lease_expires_at",
             "created_at",
             "id",
@@ -70,3 +72,29 @@ class SubscriptionSearchTask(Base):
         Index("ix_subscriptionsearchtask_batch_position", "batch_id", "position", "id"),
         Index("ix_subscriptionsearchtask_subscription", "subscription_id", "created_at", "id"),
     )
+
+
+class SubscriptionSiteBudget(Base):
+    """记录兜底搜索对单个站点的唯一租约、间隔与错误冷却。"""
+
+    id = get_id_column()
+    site_id: Mapped[int] = mapped_column(Integer, nullable=False)
+    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))
+    next_allowed_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    consecutive_failures: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    success_streak: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    last_outcome: Mapped[Optional[str]] = mapped_column(String(32))
+    last_error: Mapped[Optional[str]] = mapped_column(Text)
+    updated_at: Mapped[str] = mapped_column(String(40), nullable=False)
+
+    __table_args__ = (
+        UniqueConstraint("site_id", name="uq_subscriptionsitebudget_site_id"),
+        Index(
+            "ix_subscriptionsitebudget_ready",
+            "next_allowed_at",
+            "lease_expires_at",
+            "site_id",
+        ),
+    )

app/db/oper/subscriptionsearch.py

@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
 
 from app.db.base import DbOper, execute_dml
 from app.db.models.subscriptionsearch import (
+    SubscriptionSiteBudget,
     SubscriptionSearchBatch,
     SubscriptionSearchTask,
 )
@@ -29,6 +30,7 @@ class SubscriptionSearchOper(DbOper):
         subscription_ids: tuple[int, ...],
         source: str,
         priority: int,
+        available_at: Optional[str],
     ) -> tuple[SubscriptionSearchBatch, int, int]:
         """创建批次，并以活动键合并同一订阅的重叠搜索入口。"""
         if not isinstance(self._db, Session):
@@ -58,6 +60,7 @@ class SubscriptionSearchOper(DbOper):
                 priority=priority,
                 position=position,
                 state="queued",
+                available_at=available_at or now,
                 created_at=now,
                 updated_at=now,
             )
@@ -77,6 +80,16 @@ class SubscriptionSearchOper(DbOper):
                             (SubscriptionSearchTask.priority < priority, priority),
                             else_=SubscriptionSearchTask.priority,
                         ),
+                        available_at=case(
+                            (
+                                or_(
+                                    SubscriptionSearchTask.available_at.is_(None),
+                                    SubscriptionSearchTask.available_at > (available_at or now),
+                                ),
+                                available_at or now,
+                            ),
+                            else_=SubscriptionSearchTask.available_at,
+                        ),
                         updated_at=now,
                     ),
                     execution_options={"synchronize_session": False},
@@ -100,6 +113,10 @@ class SubscriptionSearchOper(DbOper):
                 select(SubscriptionSearchTask)
                 .where(
                     SubscriptionSearchTask.cancel_requested == 0,
+                    or_(
+                        SubscriptionSearchTask.available_at.is_(None),
+                        SubscriptionSearchTask.available_at <= now,
+                    ),
                     or_(
                         SubscriptionSearchTask.state == "queued",
                         and_(
@@ -113,6 +130,7 @@ class SubscriptionSearchOper(DbOper):
                 )
                 .order_by(
                     SubscriptionSearchTask.priority.desc(),
+                    SubscriptionSearchTask.available_at.asc(),
                     SubscriptionSearchTask.created_at.asc(),
                     SubscriptionSearchTask.position.asc(),
                     SubscriptionSearchTask.id.asc(),
@@ -128,6 +146,10 @@ class SubscriptionSearchOper(DbOper):
                 .where(
                     SubscriptionSearchTask.id == candidate.id,
                     SubscriptionSearchTask.cancel_requested == 0,
+                    or_(
+                        SubscriptionSearchTask.available_at.is_(None),
+                        SubscriptionSearchTask.available_at <= now,
+                    ),
                     or_(
                         SubscriptionSearchTask.state == "queued",
                         and_(
@@ -334,6 +356,143 @@ class SubscriptionSearchOper(DbOper):
             select(SubscriptionSearchBatch).where(SubscriptionSearchBatch.batch_id == batch_id)
         ).scalars().first()
 
+    def claim_site(
+        self,
+        *,
+        site_id: int,
+        owner: str,
+        lease_seconds: int,
+    ) -> tuple[SubscriptionSiteBudget, bool]:
+        """以 CAS 认领单站点租约，返回当前或已认领预算记录。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅站点预算认领需要调用方提供同步 Session")
+        now = utc_now_text()
+        record = self._ensure_site_budget(site_id=site_id, now=now)
+        lease_busy = bool(
+            record.lease_token
+            and record.lease_expires_at
+            and record.lease_expires_at > now
+        )
+        if lease_busy or record.next_allowed_at > now:
+            return record, False
+        lease_token = uuid4().hex
+        lease_expires_at = (
+            datetime.now(timezone.utc) + timedelta(seconds=max(1, lease_seconds))
+        ).isoformat(timespec="seconds")
+        updated = execute_dml(
+            self._db,
+            update(SubscriptionSiteBudget)
+            .where(
+                SubscriptionSiteBudget.id == record.id,
+                or_(
+                    SubscriptionSiteBudget.lease_token.is_(None),
+                    SubscriptionSiteBudget.lease_expires_at.is_(None),
+                    SubscriptionSiteBudget.lease_expires_at <= now,
+                ),
+                SubscriptionSiteBudget.next_allowed_at <= now,
+            )
+            .values(
+                lease_owner=owner,
+                lease_token=lease_token,
+                lease_expires_at=lease_expires_at,
+                updated_at=now,
+            ),
+            execution_options={"synchronize_session": False},
+        )
+        if not updated:
+            self._db.expire_all()
+            current = self._db.execute(
+                select(SubscriptionSiteBudget).where(
+                    SubscriptionSiteBudget.site_id == site_id
+                )
+            ).scalar_one()
+            return current, False
+        self._db.flush()
+        self._db.expire_all()
+        claimed = self._db.execute(
+            select(SubscriptionSiteBudget).where(
+                SubscriptionSiteBudget.site_id == site_id,
+                SubscriptionSiteBudget.lease_token == lease_token,
+            )
+        ).scalar_one()
+        return claimed, True
+
+    def finish_site(
+        self,
+        *,
+        site_id: int,
+        lease_token: str,
+        outcome: str,
+        next_allowed_at: str,
+        error: Optional[str],
+    ) -> bool:
+        """释放当前站点租约，并按结果推进失败或恢复计数。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅站点预算收口需要调用方提供同步 Session")
+        record = self._db.execute(
+            select(SubscriptionSiteBudget).where(
+                SubscriptionSiteBudget.site_id == site_id,
+                SubscriptionSiteBudget.lease_token == lease_token,
+            )
+        ).scalars().first()
+        if record is None:
+            return False
+        if outcome == "success":
+            failures = max(0, record.consecutive_failures - 1)
+            success_streak = record.success_streak + 1
+        elif outcome == "skipped":
+            failures = record.consecutive_failures
+            success_streak = record.success_streak
+        else:
+            failures = record.consecutive_failures + 1
+            success_streak = 0
+        now = utc_now_text()
+        return bool(execute_dml(
+            self._db,
+            update(SubscriptionSiteBudget)
+            .where(
+                SubscriptionSiteBudget.id == record.id,
+                SubscriptionSiteBudget.lease_token == lease_token,
+            )
+            .values(
+                lease_owner=None,
+                lease_token=None,
+                lease_expires_at=None,
+                next_allowed_at=next_allowed_at,
+                consecutive_failures=failures,
+                success_streak=success_streak,
+                last_outcome=outcome,
+                last_error=error,
+                updated_at=now,
+            ),
+            execution_options={"synchronize_session": False},
+        ))
+
+    def _ensure_site_budget(self, *, site_id: int, now: str) -> SubscriptionSiteBudget:
+        """并发安全地创建站点预算初始记录。"""
+        record = self._db.execute(
+            select(SubscriptionSiteBudget).where(
+                SubscriptionSiteBudget.site_id == site_id
+            )
+        ).scalars().first()
+        if record is not None:
+            return record
+        try:
+            with self._db.begin_nested():
+                self._db.add(SubscriptionSiteBudget(
+                    site_id=site_id,
+                    next_allowed_at=now,
+                    updated_at=now,
+                ))
+                self._db.flush()
+        except IntegrityError:
+            self._db.expire_all()
+        return self._db.execute(
+            select(SubscriptionSiteBudget).where(
+                SubscriptionSiteBudget.site_id == site_id
+            )
+        ).scalar_one()
+
     def _batch_cancel_requested(self, batch_id: str) -> bool:
         """在当前事务中读取批次取消标记。"""
         return bool(self._db.execute(

app/modules/indexer/__init__.py

@@ -5,6 +5,7 @@ from typing import Any, Callable, List, Mapping, Optional, Tuple, Union, cast
 
 from app.application.site.health import get_configured_site_health_service
 from app.application.site.query import get_configured_site_query_service
+from app.application.site.search_observation import report_site_search_outcome
 from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
 from app.domain import site as site_rules
 from app.domain.context import Context, SubtitleInfo, TorrentInfo
@@ -64,6 +65,24 @@ class _IndexerSearchOutcome:
     error_flag: bool
     result: List[dict[str, Any]]
     seconds: int
+    error: Optional[Exception] = None
+
+
+def _classify_search_failure(error: Optional[Exception]) -> str:
+    """把索引器异常归一为站点预算可执行的冷却类别。"""
+    if error is None:
+        return "error"
+    status_code = getattr(error, "status_code", None)
+    text = f"{type(error).__name__}: {error}".lower()
+    if status_code == 429 or "429" in text or "rate limit" in text:
+        return "rate_limited"
+    if status_code == 403 or "403" in text or "forbidden" in text:
+        return "forbidden"
+    if any(token in text for token in ("未登录", "登录失效", "login", "cookie")):
+        return "login_invalid"
+    if isinstance(error, TimeoutError) or "timeout" in text or "timed out" in text or "超时" in text:
+        return "timeout"
+    return "error"
 
 
 class IndexerModule(_ModuleBase):
@@ -359,12 +378,14 @@ class IndexerModule(_ModuleBase):
         start_time: datetime,
         error_flag: bool,
         result: List[dict[str, Any]],
+        error: Optional[Exception] = None,
     ) -> _IndexerSearchOutcome:
         """把同步、异步 I/O 结果整理为共用的搜索完成状态"""
         return _IndexerSearchOutcome(
             error_flag=error_flag,
             result=result,
             seconds=(datetime.now() - start_time).seconds,
+            error=error,
         )
 
     @staticmethod
@@ -403,15 +424,18 @@ class IndexerModule(_ModuleBase):
             page=page,
         )
         if not request:
+            report_site_search_outcome(attempted=False, outcome="skipped")
             return []
 
         # 开始搜索
+        error: Optional[Exception] = None
         try:
             error_flag, result = self.__execute_search(site, request)
         except Exception as err:
+            error = err
             self.__log_search_error(site, "torrents", err)
 
-        outcome = self.__create_search_outcome(start_time, error_flag, result)
+        outcome = self.__create_search_outcome(start_time, error_flag, result, error)
 
         # 统计索引情况
         self.__indexer_statistic(
@@ -419,6 +443,15 @@ class IndexerModule(_ModuleBase):
             error_flag=outcome.error_flag,
             seconds=outcome.seconds,
         )
+        report_site_search_outcome(
+            attempted=True,
+            outcome=(
+                _classify_search_failure(outcome.error)
+                if outcome.error_flag or outcome.error is not None
+                else "success"
+            ),
+            error=str(outcome.error) if outcome.error is not None else None,
+        )
 
         # 返回结果
         return self.__parse_result(
@@ -498,15 +531,18 @@ class IndexerModule(_ModuleBase):
             page=page,
         )
         if not request:
+            report_site_search_outcome(attempted=False, outcome="skipped")
             return []
 
         # 开始搜索
+        error: Optional[Exception] = None
         try:
             error_flag, result = await self.__async_execute_search(site, request)
         except Exception as err:
+            error = err
             self.__log_search_error(site, "torrents", err)
 
-        outcome = self.__create_search_outcome(start_time, error_flag, result)
+        outcome = self.__create_search_outcome(start_time, error_flag, result, error)
 
         # 统计索引情况
         await self.__async_indexer_statistic(
@@ -514,6 +550,15 @@ class IndexerModule(_ModuleBase):
             error_flag=outcome.error_flag,
             seconds=outcome.seconds,
         )
+        report_site_search_outcome(
+            attempted=True,
+            outcome=(
+                _classify_search_failure(outcome.error)
+                if outcome.error_flag or outcome.error is not None
+                else "success"
+            ),
+            error=str(outcome.error) if outcome.error is not None else None,
+        )
 
         # 返回结果
         return self.__parse_result(

database/versions/d2a7c5e9f1b4_3_0_20.py

@@ -0,0 +1,95 @@
+"""3.0.20 增加订阅搜索站点预算与批次启动抖动。
+
+Revision ID: d2a7c5e9f1b4
+Revises: c1f4a8d2e6b9
+Create Date: 2026-09-01
+"""
+
+# Alembic 的 op 是运行期代理，静态分析无法看到实际操作方法。
+# pylint: disable=no-member
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "d2a7c5e9f1b4"
+down_revision = "c1f4a8d2e6b9"
+branch_labels = None
+depends_on = None
+
+_TASK_TABLE = "subscriptionsearchtask"
+_BUDGET_TABLE = "subscriptionsitebudget"
+
+
+def _table_names() -> set[str]:
+    """返回当前数据库表名集合。"""
+    return set(sa.inspect(op.get_bind()).get_table_names())
+
+
+def _column_names(table_name: str) -> set[str]:
+    """返回指定表的当前字段名集合。"""
+    return {
+        column["name"]
+        for column in sa.inspect(op.get_bind()).get_columns(table_name)
+    }
+
+
+def upgrade() -> None:
+    """创建单站点预算，并让搜索任务支持持久启动抖动。"""
+    tables = _table_names()
+    if _TASK_TABLE in tables and "available_at" not in _column_names(_TASK_TABLE):
+        with op.batch_alter_table(_TASK_TABLE) as batch_op:
+            batch_op.add_column(sa.Column("available_at", sa.String(length=40), nullable=True))
+            batch_op.drop_index("ix_subscriptionsearchtask_claim")
+            batch_op.create_index(
+                "ix_subscriptionsearchtask_claim",
+                [
+                    "state",
+                    "priority",
+                    "available_at",
+                    "lease_expires_at",
+                    "created_at",
+                    "id",
+                ],
+            )
+        op.execute(
+            sa.text(
+                "UPDATE subscriptionsearchtask "
+                "SET available_at = created_at WHERE available_at IS NULL"
+            )
+        )
+    if _BUDGET_TABLE not in tables:
+        op.create_table(
+            _BUDGET_TABLE,
+            sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+            sa.Column("site_id", 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("next_allowed_at", sa.String(length=40), nullable=False),
+            sa.Column("consecutive_failures", sa.Integer(), nullable=False),
+            sa.Column("success_streak", sa.Integer(), nullable=False),
+            sa.Column("last_outcome", sa.String(length=32), nullable=True),
+            sa.Column("last_error", sa.Text(), nullable=True),
+            sa.Column("updated_at", sa.String(length=40), nullable=False),
+            sa.UniqueConstraint("site_id", name="uq_subscriptionsitebudget_site_id"),
+        )
+        op.create_index(
+            "ix_subscriptionsitebudget_ready",
+            _BUDGET_TABLE,
+            ["next_allowed_at", "lease_expires_at", "site_id"],
+        )
+
+
+def downgrade() -> None:
+    """移除站点预算与任务启动抖动。"""
+    tables = _table_names()
+    if _BUDGET_TABLE in tables:
+        op.drop_table(_BUDGET_TABLE)
+    if _TASK_TABLE in tables and "available_at" in _column_names(_TASK_TABLE):
+        with op.batch_alter_table(_TASK_TABLE) as batch_op:
+            batch_op.drop_index("ix_subscriptionsearchtask_claim")
+            batch_op.drop_column("available_at")
+            batch_op.create_index(
+                "ix_subscriptionsearchtask_claim",
+                ["state", "priority", "lease_expires_at", "created_at", "id"],
+            )

docs/architecture-overview.md

@@ -754,8 +754,8 @@ flowchart LR
 
 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 925 |
+| Python 模块 | 927 |
-| 内部导入边 | 7,728 |
+| 内部导入边 | 7,745 |
 | 非平凡 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 模块 / 内部依赖边 | 925 / 7,728 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 927 / 7,745 | `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-002B`
+> 当前叶：`SUB-GOV-002C`
 > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、
 > `V3-RDY-009A1C`、`V3-RDY-009A1D`
 > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环
@@ -233,13 +233,13 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
 | `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 | `completed（2026-09-01）` |
-| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `in_progress` |
+| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `completed（2026-09-01）` |
-| `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `pending` |
+| `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `in_progress` |
 | `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `pending` |
 | `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` |
 | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` |
 
-当前只激活 `SUB-GOV-002B`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
+当前只激活 `SUB-GOV-002C`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。
 
 ### 6.1 SUB-GOV-001A 验收证据
@@ -314,6 +314,22 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
 - 验证：队列/编排/调度专项 `43 passed`，迁移与声明式模型 `35 passed`，复杂度/并发/架构/迁移组合
   `119 passed`，错误级 Pylint 为 0，Alembic 唯一 head 为 `c1f4a8d2e6b9`。
 
+### 6.7 SUB-GOV-002B 验收证据
+
+- 新增持久 `SubscriptionSiteBudget`，以 `site_id` 唯一租约保证每站点最多一个在途兜底搜索；不同站点使用
+  独立记录，可由 Search provider 的现有 worker 并行推进；
+- 每次真实站点页请求后写入 `next_allowed_at`：成功保持 `60–300s` 随机间隔，历史失败按一级一级递减实现
+  渐进恢复；`429/403`、登录失效、超时和通用错误进入有上限的指数冷却；
+- 自动 fallback 批次在任务 `available_at` 中持久化 `0–60s` 全局启动抖动；手工任务优先级更高且立即入队，
+  但进入 Search provider 后仍使用完全相同的站点租约、间隔和冷却，不能旁路；
+- 站点预算只包装订阅队列的同步搜索实例，不改变 RSS/Spider 刷新入口和普通手工资源搜索；索引器列表返回
+  合同、同步/异步结果及既有站点健康统计语义保持不变；
+- 等待每秒检查批次取消，超过 5 秒短窗口的冷却或在途站点不阻塞整批：该站点本轮返回空页，其他站点继续，
+  订阅在处理成功站点结果后以聚合失败收口，避免慢站点拖死后续订阅；
+- 验证：站点预算/队列/编排/provider 专项 `38 passed`，搜索和索引器兼容 `91 passed`，迁移与声明式模型
+  `37 passed, 1 skipped`，架构合同组合 `149 passed`，错误级 Pylint 为 0，Alembic 唯一 head 为
+  `d2a7c5e9f1b4`，Host 架构基线与 `git diff --check` 通过。
+
 ## 7. 上线前验证与验收
 
 ### 7.1 场景

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 7728,
+  "edge_count": 7745,
-  "edge_sha256": "33cdb555e19d7a5d15dbed065c98908269c3528ec4d61c2ac5515420c0b4cfa4",
+  "edge_sha256": "4c722884c3a228e2fc51d5389d4fec3e10436e797c4ae168b9306154a9179475",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -3442,6 +3442,9 @@
     "app.application.subscription.delete -> app.schemas",
     "app.application.subscription.delete -> app.schemas.common",
     "app.application.subscription.delete -> app.schemas.event",
+    "app.application.subscription.execution -> app.application",
+    "app.application.subscription.execution -> app.application.subscription",
+    "app.application.subscription.execution -> app.application.subscription.sitebudget",
     "app.application.subscription.facts -> app.application",
     "app.application.subscription.facts -> app.application.subscription",
     "app.application.subscription.facts -> app.application.subscription.contract",
@@ -3492,6 +3495,11 @@
     "app.application.subscription.search -> app.application",
     "app.application.subscription.search -> app.application.subscription",
     "app.application.subscription.search -> app.application.subscription.contract",
+    "app.application.subscription.sitebudget -> app.application",
+    "app.application.subscription.sitebudget -> app.application.site",
+    "app.application.subscription.sitebudget -> app.application.site.search_observation",
+    "app.application.subscription.sitebudget -> app.runtime",
+    "app.application.subscription.sitebudget -> app.runtime.stop",
     "app.application.subscription.write -> app.application",
     "app.application.subscription.write -> app.application.outbox",
     "app.application.subscription.write -> app.application.subscription",
@@ -4172,6 +4180,9 @@
     "app.chain.search.cache -> app.schemas.types",
     "app.chain.search.contract -> app.chain",
     "app.chain.search.contract -> app.chain.base",
+    "app.chain.search.facade -> app.application",
+    "app.chain.search.facade -> app.application.subscription",
+    "app.chain.search.facade -> app.application.subscription.sitebudget",
     "app.chain.search.facade -> app.chain",
     "app.chain.search.facade -> app.chain.base",
     "app.chain.search.facade -> app.chain.search",
@@ -4246,6 +4257,9 @@
     "app.chain.search.provider -> app.application",
     "app.chain.search.provider -> app.application.configuration",
     "app.chain.search.provider -> app.application.site",
+    "app.chain.search.provider -> app.application.site.search_observation",
+    "app.chain.search.provider -> app.application.subscription",
+    "app.chain.search.provider -> app.application.subscription.sitebudget",
     "app.chain.search.provider -> app.chain",
     "app.chain.search.provider -> app.chain.search",
     "app.chain.search.provider -> app.chain.search.contract",
@@ -4588,6 +4602,7 @@
     "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.application.subscription.sitebudget",
     "app.chain.subscribe.search -> app.chain",
     "app.chain.subscribe.search -> app.chain.media",
     "app.chain.subscribe.search -> app.chain.search",
@@ -5134,6 +5149,7 @@
     "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.application.subscription.sitebudget",
     "app.db.adapters.subscriptionsearch -> app.db",
     "app.db.adapters.subscriptionsearch -> app.db.models",
     "app.db.adapters.subscriptionsearch -> app.db.models.subscriptionsearch",
@@ -6133,6 +6149,7 @@
     "app.modules.indexer -> app.application.site",
     "app.modules.indexer -> app.application.site.health",
     "app.modules.indexer -> app.application.site.query",
+    "app.modules.indexer -> app.application.site.search_observation",
     "app.modules.indexer -> app.domain",
     "app.modules.indexer -> app.domain.context",
     "app.modules.indexer -> app.domain.site",
@@ -8821,7 +8838,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 925,
+  "module_count": 927,
   "modules": [
     "app",
     "app.adapters",
@@ -9106,6 +9123,7 @@
     "app.application.site.health",
     "app.application.site.mutation",
     "app.application.site.query",
+    "app.application.site.search_observation",
     "app.application.storage",
     "app.application.subscription",
     "app.application.subscription.candidates",
@@ -9119,6 +9137,7 @@
     "app.application.subscription.priority",
     "app.application.subscription.query",
     "app.application.subscription.search",
+    "app.application.subscription.sitebudget",
     "app.application.subscription.write",
     "app.application.system",
     "app.application.torrent",

tests/test_subscription_search_governance.py

@@ -73,6 +73,7 @@ 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)
+    monkeypatch.setattr(chain, "_search_batch_available_at", lambda _source: "1970-01-01T00:00:00+00:00")
     processed = []
     monkeypatch.setattr(
         chain,
@@ -94,6 +95,7 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke
     """单订阅异常不得中止批次后续任务，聚合终态必须暴露失败。"""
     subscribes = [_subscribe(3), _subscribe(4)]
     chain = _chain(tmp_path, subscribes)
+    monkeypatch.setattr(chain, "_search_batch_available_at", lambda _source: "1970-01-01T00:00:00+00:00")
     processed = []
 
     def process(subscribe, _searchchain):

tests/test_subscription_site_budget.py

@@ -0,0 +1,266 @@
+"""订阅兜底搜索的站点并发、冷却、恢复与取消测试。"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from app.application.site.search_observation import (
+    SiteSearchObservation,
+    report_site_search_outcome,
+)
+from app.application.subscription.sitebudget import (
+    SiteBudgetClaim,
+    SubscriptionSearchCancelled,
+    SubscriptionSiteBudget,
+)
+from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository
+from app.db.base import Base
+from app.db.models.subscriptionsearch import SubscriptionSiteBudget as SiteBudgetRecord
+from app.chain.search.facade import SearchChain
+from app.modules.indexer import _classify_search_failure
+from app.runtime.stop import ProcessStopState
+
+
+def _repository(tmp_path):
+    """构造使用独立 SQLite 文件的站点预算仓储。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'site-budget.db'}")
+    Base.metadata.create_all(engine)
+    return TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)), engine
+
+
+def test_site_budget_allows_one_inflight_per_site_and_independent_sites(tmp_path):
+    """同站点第二个调用必须等待，不同站点可立即并行。"""
+    repository, _engine = _repository(tmp_path)
+
+    first = repository.claim_site(site_id=1, owner="worker-a", lease_seconds=900)
+    same_site = repository.claim_site(site_id=1, owner="worker-b", lease_seconds=900)
+    other_site = repository.claim_site(site_id=2, owner="worker-b", lease_seconds=900)
+
+    assert first.acquired is True
+    assert same_site.acquired is False
+    assert same_site.lease_token is None
+    assert other_site.acquired is True
+
+
+def test_site_budget_applies_error_cooldown_and_gradual_success_recovery(tmp_path):
+    """失败增加冷却计数，后续成功每次只恢复一级而非直接清零。"""
+    repository, engine = _repository(tmp_path)
+    claim = repository.claim_site(site_id=3, owner="worker-a", lease_seconds=900)
+    now = datetime.now(timezone.utc)
+
+    assert repository.finish_site(
+        site_id=3,
+        lease_token=claim.lease_token,
+        outcome="timeout",
+        next_allowed_at=(now + timedelta(minutes=5)).isoformat(timespec="seconds"),
+        error="request timeout",
+    ) is True
+    with Session(engine) as session:
+        failed = session.execute(
+            select(SiteBudgetRecord).where(SiteBudgetRecord.site_id == 3)
+        ).scalar_one()
+        assert failed.consecutive_failures == 1
+        assert failed.success_streak == 0
+        failed.next_allowed_at = (now - timedelta(seconds=1)).isoformat(timespec="seconds")
+        session.commit()
+
+    recovered = repository.claim_site(site_id=3, owner="worker-b", lease_seconds=900)
+    assert recovered.acquired is True
+    assert recovered.consecutive_failures == 1
+    assert repository.finish_site(
+        site_id=3,
+        lease_token=recovered.lease_token,
+        outcome="success",
+        next_allowed_at=(now + timedelta(minutes=1)).isoformat(timespec="seconds"),
+    ) is True
+    with Session(engine) as session:
+        healthy = session.execute(
+            select(SiteBudgetRecord).where(SiteBudgetRecord.site_id == 3)
+        ).scalar_one()
+        assert healthy.consecutive_failures == 0
+        assert healthy.success_streak == 1
+        assert healthy.last_outcome == "success"
+
+
+@pytest.mark.parametrize(
+    ("error", "outcome"),
+    [
+        (RuntimeError("HTTP 429"), "rate_limited"),
+        (RuntimeError("403 forbidden"), "forbidden"),
+        (RuntimeError("Cookie 登录失效"), "login_invalid"),
+        (TimeoutError("request timed out"), "timeout"),
+    ],
+)
+def test_indexer_failures_map_to_site_cooldown_categories(error, outcome):
+    """外站典型失败必须进入对应的站点级冷却类别。"""
+    assert _classify_search_failure(error) == outcome
+
+
+class _WaitingRepository:
+    """始终返回未来重试时间的可取消预算仓储。"""
+
+    def claim_site(self, *, site_id: int, owner: str, lease_seconds: int) -> SiteBudgetClaim:
+        """返回未认领的未来预算。"""
+        del owner, lease_seconds
+        return SiteBudgetClaim(
+            site_id=site_id,
+            acquired=False,
+            retry_at=(datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat(timespec="seconds"),
+            consecutive_failures=0,
+        )
+
+    def finish_site(self, **_kwargs) -> bool:
+        """等待测试不会取得租约，因此不应调用收口。"""
+        raise AssertionError("未取得站点预算时不应收口")
+
+
+def test_site_budget_wait_is_cancellable_without_business_lock():
+    """批次取消应在一次短等待后终止预算获取。"""
+    cancelled = False
+
+    def sleeper(_seconds: float) -> None:
+        """模拟等待一次后收到取消请求。"""
+        nonlocal cancelled
+        cancelled = True
+
+    budget = SubscriptionSiteBudget(
+        repository=_WaitingRepository(),
+        owner="worker-a",
+        cancelled=lambda: cancelled,
+        stop_state=ProcessStopState(),
+        sleeper=sleeper,
+        max_wait_seconds=600,
+    )
+
+    with pytest.raises(SubscriptionSearchCancelled):
+        budget.acquire(9)
+
+
+def test_site_budget_delay_keeps_manual_requests_under_same_budget():
+    """来源优先级不参与站点冷却计算，手工任务不能获得旁路。"""
+    budget = SubscriptionSiteBudget(
+        repository=_WaitingRepository(),
+        owner="manual-worker",
+        cancelled=lambda: False,
+        stop_state=ProcessStopState(),
+        random_uniform=lambda _low, _high: 60.0,
+    )
+
+    assert budget._next_delay("success", consecutive_failures=0) == 60.0  # pylint: disable=protected-access
+    assert budget._next_delay("success", consecutive_failures=2) == 120.0  # pylint: disable=protected-access
+    assert budget._next_delay("rate_limited", consecutive_failures=0) == 900.0  # pylint: disable=protected-access
+
+
+def test_skipped_search_releases_budget_without_external_interval():
+    """本地限流等未发请求结果不应伪造成功或追加外站间隔。"""
+    captured = {}
+
+    class _Repository(_WaitingRepository):
+        """记录预算收口参数。"""
+
+        def finish_site(self, **kwargs) -> bool:
+            """保存收口事实供断言。"""
+            captured.update(kwargs)
+            return True
+
+    now = datetime(2026, 9, 1, tzinfo=timezone.utc)
+    budget = SubscriptionSiteBudget(
+        repository=_Repository(),
+        owner="worker-a",
+        cancelled=lambda: False,
+        stop_state=ProcessStopState(),
+        clock=lambda: now,
+    )
+    claim = SiteBudgetClaim(
+        site_id=10,
+        acquired=True,
+        retry_at=now.isoformat(timespec="seconds"),
+        consecutive_failures=0,
+        lease_token="lease-token",
+    )
+
+    assert budget.finish(claim, SiteSearchObservation()) is True
+    assert captured["outcome"] == "skipped"
+    assert captured["next_allowed_at"] == now.isoformat(timespec="seconds")
+
+
+def test_search_provider_reports_cooled_site_without_blocking_other_results():
+    """超出短等待窗口的站点返回空页并记录聚合失败，而非阻塞整个 provider。"""
+    repository = _WaitingRepository()
+    budget = SubscriptionSiteBudget(
+        repository=repository,
+        owner="fallback-worker",
+        cancelled=lambda: False,
+        stop_state=ProcessStopState(),
+        max_wait_seconds=0,
+    )
+    chain = object.__new__(SearchChain)
+    chain.configure_subscription_site_budget(budget)
+    chain.search_site_torrents = lambda **_kwargs: ["unexpected"]
+
+    result = chain._search_site_torrents_with_budget(  # pylint: disable=protected-access
+        site={"id": 11, "name": "Cooling"},
+        keyword="movie",
+        mtype=None,
+        page=0,
+    )
+
+    assert result == []
+    failures = chain.consume_subscription_site_budget_failures()
+    assert len(failures) == 1
+    assert "站点 11" in failures[0]
+
+
+def test_search_provider_releases_successful_site_budget():
+    """真实站点页完成后必须释放租约并写入成功间隔。"""
+    captured = {}
+
+    class _Repository(_WaitingRepository):
+        """提供立即可用租约并记录收口参数。"""
+
+        def claim_site(self, *, site_id: int, owner: str, lease_seconds: int) -> SiteBudgetClaim:
+            """返回当前调用独占的站点租约。"""
+            del owner, lease_seconds
+            return SiteBudgetClaim(
+                site_id=site_id,
+                acquired=True,
+                retry_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
+                consecutive_failures=0,
+                lease_token="lease-token",
+            )
+
+        def finish_site(self, **kwargs) -> bool:
+            """记录成功收口事实。"""
+            captured.update(kwargs)
+            return True
+
+    budget = SubscriptionSiteBudget(
+        repository=_Repository(),
+        owner="manual-worker",
+        cancelled=lambda: False,
+        stop_state=ProcessStopState(),
+        random_uniform=lambda _low, _high: 60.0,
+    )
+    chain = object.__new__(SearchChain)
+    chain.configure_subscription_site_budget(budget)
+
+    def search_site_torrents(**_kwargs):
+        """模拟索引器成功并发布观察结果。"""
+        report_site_search_outcome(attempted=True, outcome="success")
+        return ["torrent"]
+
+    chain.search_site_torrents = search_site_torrents
+
+    result = chain._search_site_torrents_with_budget(  # pylint: disable=protected-access
+        site={"id": 12, "name": "Healthy"},
+        keyword="movie",
+        mtype=None,
+        page=0,
+    )
+
+    assert result == ["torrent"]
+    assert captured["site_id"] == 12
+    assert captured["outcome"] == "success"
+    assert captured["lease_token"] == "lease-token"