fix: defer subscription search on site budget conflicts

app/application/subscription/execution.py

@@ -200,6 +200,16 @@ class SubscriptionSearchRepository(Protocol):
         """释放尚未完成的任务租约，供停止或取消后恢复。"""
         ...
 
+    def defer_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        available_at: str,
+    ) -> bool:
+        """把临时站点预算冲突任务退回队列，并设置下一次领取时间。"""
+        ...
+
     def is_cancel_requested(self, task_id: str) -> bool:
         """判断任务或所属批次是否已请求取消。"""
         ...

app/application/subscription/sitebudget.py

@@ -15,11 +15,29 @@ class SubscriptionSearchCancelled(RuntimeError):
     """表示订阅搜索在可取消预算等待点终止。"""
 
 
+@dataclass(frozen=True, slots=True)
+class SubscriptionSiteBudgetDeferral:
+    """记录一次站点预算冲突及该站点最早可再次尝试的时间。"""
+
+    site_id: int
+    retry_at: str
+
+
+class SubscriptionSearchDeferred(RuntimeError):
+    """表示订阅搜索未失败，而是应在站点预算可用后重新入队。"""
+
+    def __init__(self, *, retry_at: str, site_ids: tuple[int, ...]) -> None:
+        """保存队列恢复所需的时间和冲突站点，避免把临时冲突写成错误。"""
+        super().__init__(f"订阅搜索已延后，站点预算最早可重试：{retry_at}")
+        self.retry_at = retry_at
+        self.site_ids = site_ids
+
+
 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

app/application/subscription/status.py

@@ -153,6 +153,8 @@ class SubscriptionExecutionStatusService:
             state = phase = "cancelling"
         elif task.state == "running":
             state = phase = task.phase or "running"
+        elif task.state == "queued" and task.phase == "waiting_site_budget":
+            state = phase = "waiting_site_budget"
         else:
             state = phase = task.state
         return SubscriptionExecutionStatus(

app/chain/search/contract.py

@@ -99,6 +99,8 @@ if TYPE_CHECKING:
         process: Callable[..., Any]
         record_subscription_site_budget_failure: Callable[..., Any]
         consume_subscription_site_budget_failures: Callable[..., Any]
+        record_subscription_site_budget_deferred: Callable[..., Any]
+        consume_subscription_site_budget_deferrals: Callable[..., Any]
         save_last_search_params: Callable[..., Any]
         search_by_id: Callable[..., Any]
         search_by_title: Callable[..., Any]

app/chain/search/facade.py

@@ -4,7 +4,10 @@ import threading
 from collections.abc import AsyncIterator, Callable
 from typing import Any, Optional, TypeVar, cast
 
-from app.application.subscription.sitebudget import SubscriptionSiteBudget
+from app.application.subscription.sitebudget import (
+    SubscriptionSiteBudget,
+    SubscriptionSiteBudgetDeferral,
+)
 from app.chain.base import ChainBase
 from app.chain.search.cache import SearchCacheOwner
 from app.chain.search.media import SearchMediaOwner
@@ -49,6 +52,7 @@ class SearchChain(ChainBase):
         """仅为订阅搜索启用或清除站点预算，不影响其它搜索入口。"""
         self._subscription_site_budget = budget
         self._subscription_site_budget_failures: list[str] = []
+        self._subscription_site_budget_deferrals: list[SubscriptionSiteBudgetDeferral] = []
         self._subscription_site_budget_failure_lock = threading.Lock()
 
     def record_subscription_site_budget_failure(self, error: str) -> None:
@@ -69,6 +73,27 @@ class SearchChain(ChainBase):
             self._subscription_site_budget_failures.clear()
         return failures
 
+    def record_subscription_site_budget_deferred(
+        self,
+        deferral: SubscriptionSiteBudgetDeferral,
+    ) -> None:
+        """线程安全地记录临时站点冲突，供订阅任务重新入队。"""
+        lock = getattr(self, "_subscription_site_budget_failure_lock", None)
+        if lock is None:
+            return
+        with lock:
+            self._subscription_site_budget_deferrals.append(deferral)
+
+    def consume_subscription_site_budget_deferrals(self) -> tuple[SubscriptionSiteBudgetDeferral, ...]:
+        """读取并清空当前订阅搜索积累的站点预算延后事实。"""
+        lock = getattr(self, "_subscription_site_budget_failure_lock", None)
+        if lock is None:
+            return ()
+        with lock:
+            deferrals = tuple(self._subscription_site_budget_deferrals)
+            self._subscription_site_budget_deferrals.clear()
+        return deferrals
+
     # owner descriptor 经类访问后被 mypy 视为普通 Callable；运行时仍需取回原始
     # classmethod 函数，才能保持 SearchChain 的直接 MRO 与既有绑定语义。
     music_site_keywords = classmethod(  # type: ignore[var-annotated]

app/chain/search/provider.py

@@ -17,6 +17,7 @@ from app.application.site.observation import (
 from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
 from app.application.subscription.sitebudget import (
     SubscriptionSiteBudget,
+    SubscriptionSiteBudgetDeferral,
     SubscriptionSiteBudgetUnavailable,
 )
 from app.chain.search.contract import _SearchOwnerBase
@@ -215,7 +216,12 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
         try:
             claim = budget.acquire(site_id)
         except SubscriptionSiteBudgetUnavailable as error:
-            self.record_subscription_site_budget_failure(str(error))
+            self.record_subscription_site_budget_deferred(
+                SubscriptionSiteBudgetDeferral(
+                    site_id=error.site_id,
+                    retry_at=error.retry_at,
+                )
+            )
             logger.debug(str(error))
             return []
         with capture_site_search_observation() as observation:

app/chain/subscribe/search.py

@@ -31,7 +31,9 @@ from app.application.subscription.observability import (
 from app.application.subscription.query import SubscriptionQueryService
 from app.application.subscription.sitebudget import (
     SubscriptionSearchCancelled,
+    SubscriptionSearchDeferred,
     SubscriptionSiteBudget,
+    SubscriptionSiteBudgetDeferral,
 )
 from app.chain.media import MediaChain
 from app.chain.search.facade import SearchChain
@@ -323,6 +325,12 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
                     reason = "ttl_timeout" if execution_context.is_expired() else "cancelled"
                     outcome = "failed" if reason == "ttl_timeout" else "cancelled"
                     logger.debug(f"订阅 {subscribe.name} 搜索已在安全边界取消")
+                except SubscriptionSearchDeferred as deferred:
+                    outcome = "skipped"
+                    reason = "site_budget_deferred"
+                    logger.debug(
+                        f"订阅 {subscribe.name} 站点预算冲突，兼容搜索将在 {deferred.retry_at} 后重试"
+                    )
                 except Exception as err:
                     outcome = "failed"
                     reason = "error"
@@ -610,6 +618,20 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator):
                     "requeued" if system_stopped else "cancelled",
                     "system_stop" if system_stopped else "cancelled",
                 )
+        except SubscriptionSearchDeferred as deferred:
+            requeued = queue.defer_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                available_at=deferred.retry_at,
+            )
+            if requeued:
+                logger.debug(
+                    f"订阅 {subscribe.name} 站点预算冲突，已排队至 {deferred.retry_at} 后重试，"
+                    f"sites={','.join(str(site_id) for site_id in deferred.site_ids)}"
+                )
+                summary.record("requeued", "site_budget_deferred")
+            else:
+                logger.debug(f"订阅搜索任务 {task_id} 租约已变化，跳过重复站点预算重排队")
         except Exception as err:
             logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True)
             queue.finish_task(
@@ -821,9 +843,12 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
             filter_params=self.get_params(subscribe),
         )
         site_budget_failures = searchchain.consume_subscription_site_budget_failures()
+        site_budget_deferrals = searchchain.consume_subscription_site_budget_deferrals()
         _ensure_execution_active(execution_context)
         if not contexts:
             logger.debug(f"订阅 {subscribe.keyword or subscribe.name} 未搜索到资源")
+            if not site_budget_failures:
+                self._raise_site_budget_deferral(site_budget_deferrals, execution_context)
             self.finish_subscribe_or_not(
                 subscribe=subscribe,
                 meta=meta,
@@ -835,6 +860,8 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
         matched = self._filter_search_contexts(subscribe, contexts)
         if not matched:
             logger.debug(f"订阅 {subscribe.name} 没有符合过滤条件的资源")
+            if not site_budget_failures:
+                self._raise_site_budget_deferral(site_budget_deferrals, execution_context)
             self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo, lefts=no_exists)
             self._raise_site_budget_failures(site_budget_failures)
             return subscribe
@@ -862,6 +889,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
                 lefts=lefts,
             )
         self._raise_site_budget_failures(site_budget_failures)
+        self._raise_site_budget_deferral(site_budget_deferrals, execution_context)
         return cast(Optional[SubscriptionSnapshot], current)
 
     @staticmethod
@@ -870,6 +898,18 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
         if failures:
             raise RuntimeError("；".join(failures))
 
+    @staticmethod
+    def _raise_site_budget_deferral(
+        deferrals: tuple[SubscriptionSiteBudgetDeferral, ...],
+        execution_context: Optional[SubscriptionExecutionContext],
+    ) -> None:
+        """在没有下载副作用时，将临时站点冲突转换为持久队列延后。"""
+        if not deferrals or (execution_context and execution_context.download_started):
+            return
+        retry_at = min(deferrals, key=lambda item: item.retry_at).retry_at
+        site_ids = tuple(dict.fromkeys(item.site_id for item in deferrals))
+        raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=site_ids)
+
     def _filter_search_contexts(
         self,
         subscribe: SubscriptionSnapshot,

app/db/adapters/subscriptionsearch.py

@@ -177,6 +177,22 @@ class TransactionalSubscriptionSearchRepository:
             )
         )
 
+    def defer_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        available_at: str,
+    ) -> bool:
+        """以站点预算的下一次可用时间重新排队任务。"""
+        return self._write(
+            lambda repository: repository.defer_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                available_at=available_at,
+            )
+        )
+
     def is_cancel_requested(self, task_id: str) -> bool:
         """查询任务或批次的取消请求。"""
         return self._read(lambda repository: repository.is_cancel_requested(task_id))
@@ -208,11 +224,9 @@ class TransactionalSubscriptionSearchRepository:
                 owner=owner,
                 lease_seconds=lease_seconds,
             )
-            retry_at = (
+            retry_at = record.next_allowed_at
-                record.lease_expires_at
+            if not acquired and record.lease_token and record.lease_expires_at:
-                if record.lease_token and not acquired
+                retry_at = max(retry_at, record.lease_expires_at)
-                else record.next_allowed_at
-            ) or record.next_allowed_at
             return SiteBudgetClaim(
                 site_id=record.site_id,
                 acquired=acquired,

app/db/oper/subscriptionsearch.py

@@ -363,6 +363,60 @@ class SubscriptionSearchOper(DbOper):
         self._refresh_batch(task.batch_id, now=now, error=None)
         return True
 
+    def defer_task(
+        self,
+        *,
+        task_id: str,
+        lease_token: str,
+        available_at: str,
+    ) -> 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
+        if bool(task.cancel_requested) or self._batch_cancel_requested(task.batch_id):
+            return self.finish_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                state="cancelled",
+                error=None,
+            )
+        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="queued",
+                phase="waiting_site_budget",
+                current_site_id=None,
+                lease_owner=None,
+                lease_token=None,
+                lease_expires_at=None,
+                available_at=available_at,
+                updated_at=now,
+                finished_at=None,
+                last_error=None,
+            ),
+            execution_options={"synchronize_session": False},
+        )
+        if not updated:
+            return False
+        self._refresh_batch(task.batch_id, now=now, error=None)
+        return True
+
     def is_cancel_requested(self, task_id: str) -> bool:
         """读取任务和批次取消标记。"""
         if not isinstance(self._db, Session):

scripts/validation/subscription_governance_scale.py

@@ -379,7 +379,7 @@ def _run_site_pressure_case(
         start_barrier = threading.Barrier(2)
         owners = ("scale-owner-a", "scale-owner-b")
         chains: dict[str, SearchChain] = {}
-        owner_failures: dict[str, list[str]] = {owner: [] for owner in owners}
+        owner_deferrals: dict[str, list[Any]] = {owner: [] for owner in owners}
 
         for owner in owners:
             controller = SubscriptionSiteBudgetController(
@@ -391,22 +391,21 @@ def _run_site_pressure_case(
             )
             chain = object.__new__(SearchChain)
             chain.configure_subscription_site_budget(controller)
-            original_record_failure = chain.record_subscription_site_budget_failure
+            original_record_deferred = chain.record_subscription_site_budget_deferred
 
-            def record_failure(
+            def record_deferred(
-                error: str,
+                deferral: Any,
                 *,
-                original=original_record_failure,
+                original=original_record_deferred,
                 owner_name=owner,
             ) -> None:
-                """保留 wrapper 聚合失败并同步压力轮次观测。"""
+                """保留 wrapper 延后事实并同步压力轮次观测。"""
-                original(error)
+                original(deferral)
-                owner_failures[owner_name].append(error)
+                owner_deferrals[owner_name].append(deferral)
-                if "冷却或已有在途搜索" in error:
                 boundary.record_budget_rejection()
                 pressure_event.set()
 
-            chain.record_subscription_site_budget_failure = record_failure
+            chain.record_subscription_site_budget_deferred = record_deferred
 
             def search_site_torrents(*, _owner=owner, **_kwargs: Any) -> list[str]:
                 """将 SearchChain 的真实站点请求委托给固定边界。"""
@@ -453,11 +452,10 @@ def _run_site_pressure_case(
             thread.join(timeout=_SITE_PRESSURE_SYNC_TIMEOUT)
         owners_finished = all(not thread.is_alive() for thread in threads)
 
-        initial_failures = [
+        initial_deferrals = [
-            error
+            deferral
             for owner in owners
-            for error in owner_failures[owner]
+            for deferral in owner_deferrals[owner]
-            if "冷却或已有在途搜索" in error
         ]
         successful_owners = [
             owner
@@ -469,10 +467,7 @@ def _run_site_pressure_case(
             for owner in owners
             if owner in owner_results
             and not owner_results[owner]
-            and any(
+            and owner_deferrals[owner]
-                "冷却或已有在途搜索" in error
-                for error in owner_failures[owner]
-            )
         ]
         successful_owner = successful_owners[0] if len(successful_owners) == 1 else None
         blocked_owner = blocked_owners[0] if len(blocked_owners) == 1 else None
@@ -482,7 +477,7 @@ def _run_site_pressure_case(
             and boundary.active_at_rejection == 1
             and boundary.peak == 1
             and boundary.active == 0
-            and len(initial_failures) == 1
+            and len(initial_deferrals) == 1
             and len(successful_owners) == 1
             and len(blocked_owners) == 1
             and all(count == 1 for count in owner_invocations.values())
@@ -501,8 +496,8 @@ def _run_site_pressure_case(
         if successful_owner and blocked_owner:
             winner = chains[successful_owner]
             loser = chains[blocked_owner]
-            winner.consume_subscription_site_budget_failures()
+            winner.consume_subscription_site_budget_deferrals()
-            loser.consume_subscription_site_budget_failures()
+            loser.consume_subscription_site_budget_deferrals()
 
             if site_id != case.site_count:
                 boundary.set_outcome("success")
@@ -552,11 +547,11 @@ def _run_site_pressure_case(
                     page=0,
                 )
                 cooldown_calls = boundary.calls - calls_before_cooldown
-                cooldown_failures = loser.consume_subscription_site_budget_failures()
+                cooldown_deferrals = loser.consume_subscription_site_budget_deferrals()
                 error_cooldown_blocked = bool(
                     not cooldown_result
                     and cooldown_calls == 0
-                    and cooldown_failures
+                    and cooldown_deferrals
                 )
                 error_cooldown_persisted = bool(
                     error_record
@@ -576,7 +571,7 @@ def _run_site_pressure_case(
                 and error_cooldown_persisted
             )
         )
-        duplicate_site_claims_blocked += len(initial_failures)
+        duplicate_site_claims_blocked += len(initial_deferrals)
         successful_site_claims_reused += int(success_reused)
         error_cooldown_claims_blocked += int(error_cooldown_blocked)
         site_observations.append(
@@ -589,7 +584,7 @@ def _run_site_pressure_case(
                 "request_peak": boundary.peak,
                 "request_calls": boundary.calls,
                 "request_active_at_rejection": boundary.active_at_rejection,
-                "budget_rejections": len(initial_failures),
+                "budget_rejections": len(initial_deferrals),
                 "success_reused": success_reused,
                 "error_observed": error_observed,
                 "error_cooldown_blocked": error_cooldown_blocked,

tests/test_subscription_execution_status.py

@@ -84,6 +84,19 @@ def test_execution_status_exposes_site_wait_and_cancel_capability():
     assert statuses[1].can_cancel is True
 
 
+def test_execution_status_exposes_queued_site_wait_without_error():
+    """重新入队的站点预算冲突应显示等待状态而不是失败。"""
+    repository = _Repository()
+    repository.tasks[2] = _task(2, state="queued", phase="waiting_site_budget")
+
+    statuses = asyncio.run(SubscriptionExecutionStatusService(repository).for_subscriptions((2,)))
+
+    assert statuses[2].state == "waiting_site_budget"
+    assert statuses[2].phase == "waiting_site_budget"
+    assert statuses[2].error is None
+    assert statuses[2].can_cancel is True
+
+
 def test_failed_search_exposes_safe_error():
     """搜索失败文本必须压平且不暴露内部错误细节。"""
     repository = _Repository()

tests/test_subscription_governance_joint.py

@@ -52,8 +52,8 @@ class _MixedBudgetRepository:
         return True
 
 
-def test_cooled_site_does_not_block_independent_site_or_hide_batch_failure():
+def test_cooled_site_does_not_block_independent_site_or_mark_batch_failed():
-    """慢站点跳过后其它站点仍完成，调用方同时收到可聚合失败。"""
+    """冷却站点延后后其它站点仍完成，调用方只收到可重新入队的事实。"""
     repository = _MixedBudgetRepository()
     budget = SubscriptionSiteBudget(
         repository=repository,
@@ -91,7 +91,9 @@ def test_cooled_site_does_not_block_independent_site_or_hide_batch_failure():
 
     assert results == [2]
     assert repository.finished_sites == [2]
+    deferrals = chain.consume_subscription_site_budget_deferrals()
+    assert len(deferrals) == 1
+    assert deferrals[0].site_id == 1
     failures = chain.consume_subscription_site_budget_failures()
-    assert len(failures) == 1
+    assert failures == ()
-    assert "站点 1" in failures[0]
     assert progress.values[-1] == 100

tests/test_subscription_governance_scale.py

@@ -179,22 +179,21 @@ def test_scale_validator_rejects_fake_site_pressure(monkeypatch, tmp_path, mutat
 
 
 def test_scale_validator_rejects_unfinished_site_wrapper(monkeypatch, tmp_path):
-    """失败已登记但 wrapper 尚未返回时，压力门禁必须拒绝并暴露未收口 owner。"""
+    """延后已登记但 wrapper 尚未返回时，压力门禁必须拒绝并暴露未收口 owner。"""
     original_wrapper = SearchChain._search_site_torrents_with_budget
     release_stalled = threading.Event()
     threads_before = set(threading.enumerate())
 
     def stall_after_rejection(self, *, site, keyword, mtype, page):
         """模拟预算拒绝已记录、调用方却未取得返回值的挂起路径。"""
-        original_record_failure = self.record_subscription_site_budget_failure
+        original_record_deferred = self.record_subscription_site_budget_deferred
 
-        def record_failure_and_stall(error: str) -> None:
+        def record_deferred_and_stall(deferral) -> None:
             """在拒绝已登记后阻塞原始 wrapper 的返回。"""
-            original_record_failure(error)
+            original_record_deferred(deferral)
-            if "冷却或已有在途搜索" in error:
             release_stalled.wait()
 
-        self.record_subscription_site_budget_failure = record_failure_and_stall
+        self.record_subscription_site_budget_deferred = record_deferred_and_stall
         try:
             return original_wrapper(
                 self,
@@ -204,7 +203,7 @@ def test_scale_validator_rejects_unfinished_site_wrapper(monkeypatch, tmp_path):
                 page=page,
             )
         finally:
-            self.record_subscription_site_budget_failure = original_record_failure
+            self.record_subscription_site_budget_deferred = original_record_deferred
 
     monkeypatch.setattr(
         SearchChain,

tests/test_subscription_search_governance.py

@@ -13,13 +13,17 @@ from sqlalchemy.orm import sessionmaker
 from app.application.site.observation import report_site_search_outcome
 from app.application.subscription.contract import SubscriptionSnapshot
 from app.application.subscription.execution import SubscriptionExecutionAdmission
-from app.application.subscription.sitebudget import SubscriptionSearchCancelled
+from app.application.subscription.sitebudget import (
+    SubscriptionSearchCancelled,
+    SubscriptionSearchDeferred,
+)
 from app.chain.search.facade import SearchChain
 from app.chain.subscribe import search as subscribe_search
 from app.chain.subscribe.facade import SubscribeChain
 from app.chain.subscribe.search import _search_task_available_at
 from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository
 from app.db.base import Base
+from app.db.models.subscriptionsearch import SubscriptionSearchTask
 from app.modules.indexer import IndexerModule
 from app.schemas.types import MediaType
 
@@ -249,6 +253,37 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke
         assert chain._subscription_execution_admission.release(lease) is True
 
 
+def test_site_budget_conflict_requeues_task_without_batch_failure(tmp_path, monkeypatch):
+    """站点预算冲突应自动排队重试，不能把订阅批次置为失败。"""
+    subscribe = _subscribe(5)
+    chain = _chain(tmp_path, [subscribe])
+    _make_tasks_ready(monkeypatch)
+    retry_at = (datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat(timespec="seconds")
+    monkeypatch.setattr(
+        chain,
+        "_process_search_subscription",
+        Mock(side_effect=SubscriptionSearchDeferred(retry_at=retry_at, site_ids=(31,))),
+    )
+
+    with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()):
+        batch_id = chain.search(state="R")
+
+    batch = chain.get_search_batch(batch_id)
+    assert batch.state == "queued"
+    assert batch.finished_count == 0
+    assert batch.failed_count == 0
+    assert batch.last_error is None
+    with chain.subscription_search_repository._session_factory() as session:
+        task = session.query(SubscriptionSearchTask).filter_by(
+            subscription_id=subscribe.id,
+        ).one()
+        assert task.state == "queued"
+        assert task.phase == "waiting_site_budget"
+        assert task.available_at == retry_at
+        assert task.last_error is None
+    assert chain.subscription_search_repository.claim_next(owner="worker-after-retry") is None
+
+
 def test_search_logs_one_bounded_start_and_finish_summary(tmp_path, monkeypatch):
     """Search INFO 只保留轮次摘要，并携带任务终态与耗时字段。"""
     subscribes = [_subscribe(50), _subscribe(51)]

tests/test_subscription_search_queue.py

@@ -132,6 +132,47 @@ def test_search_queue_phase_update_requires_current_lease(tmp_path):
     ) is True
 
 
+def test_search_queue_defers_site_budget_conflict_until_retry_time(tmp_path):
+    """站点预算冲突应释放任务租约并保留同一任务等待后续恢复。"""
+    repository, engine = _repository(tmp_path)
+    enqueued = repository.enqueue(subscription_ids=(31,), source="fallback", priority=10)
+    running = repository.claim_next(owner="worker-a")
+    retry_at = (datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat(timespec="seconds")
+
+    assert repository.defer_task(
+        task_id=running.task_id,
+        lease_token=running.lease_token,
+        available_at=retry_at,
+    ) is True
+
+    batch = repository.get_batch(enqueued.batch.batch_id)
+    assert batch.state == "queued"
+    assert batch.finished_count == 0
+    assert batch.failed_count == 0
+    assert repository.claim_next(owner="worker-b") is None
+
+    with Session(engine) as session:
+        task = session.execute(
+            select(SubscriptionSearchTask).where(
+                SubscriptionSearchTask.task_id == running.task_id
+            )
+        ).scalar_one()
+        assert task.state == "queued"
+        assert task.phase == "waiting_site_budget"
+        assert task.available_at == retry_at
+        assert task.last_error is None
+        session.execute(
+            update(SubscriptionSearchTask)
+            .where(SubscriptionSearchTask.task_id == running.task_id)
+            .values(available_at="1970-01-01T00:00:00+00:00")
+        )
+        session.commit()
+
+    recovered = repository.claim_next(owner="worker-c")
+    assert recovered.task_id == running.task_id
+    assert recovered.attempt_count == 2
+
+
 def test_search_queue_cancel_finishes_queued_and_running_tasks(tmp_path):
     """取消立即终止未发请求任务，运行中任务在租约边界收口。"""
     repository, engine = _repository(tmp_path)

tests/test_subscription_site_budget.py

@@ -243,7 +243,7 @@ def test_skipped_search_releases_budget_without_external_interval():
 
 
 def test_search_provider_reports_cooled_site_without_blocking_other_results():
-    """错误冷却中的站点返回空页并记录聚合失败，而非阻塞 provider。"""
+    """错误冷却中的站点返回空页并记录延后，而非阻塞 provider 或制造失败。"""
     repository = _WaitingRepository()
     metrics = SubscriptionSiteBudgetMetrics()
     budget = SubscriptionSiteBudget(
@@ -265,9 +265,12 @@ def test_search_provider_reports_cooled_site_without_blocking_other_results():
     )
 
     assert result == []
+    deferrals = chain.consume_subscription_site_budget_deferrals()
+    assert len(deferrals) == 1
+    assert deferrals[0].site_id == 11
+    assert deferrals[0].retry_at
     failures = chain.consume_subscription_site_budget_failures()
-    assert len(failures) == 1
+    assert failures == ()
-    assert "站点 11" in failures[0]
     snapshot = metrics.snapshot()
     assert snapshot.request_count == 0
     assert snapshot.cooldown_skip_count == 1