fix(subscribe): 站点本地流控按限流窗口退避 (#6996) (#7001)

订阅分页搜索遇到站点本地流控（limit_interval/limit_count/limit_seconds）
拒绝时只按秒级 cooldown 反复重试，在整个限流窗口内持续触发风控。现在：

- 本地流控被拒时按站点配置窗口计算退避（访问间隔与计数窗口取最大值，
  并多留 1 秒越过窗口的严格比较边界）；
- 同一退避写入站点预算冷却，窗口内其它订阅任务不会继续访问该站点；
- 本地流控没有发出请求，不计入连续失败与错误退避类别。

app/application/subscription/sitebudget.py

@@ -250,13 +250,19 @@ class SubscriptionSiteBudget:
         if self._metrics:
             self._metrics.record_release_failure()
 
-    def finish(self, claim: SiteBudgetClaim, observation: SiteSearchObservation) -> bool:
-        """正常结果立即释放站点，错误结果按类别写入冷却。"""
+    def finish(self, claim: SiteBudgetClaim, observation: SiteSearchObservation,
+               *, deferral_seconds: float = 0) -> bool:
+        """正常结果立即释放站点，错误结果按类别写入冷却，本地流控按站点窗口退避。"""
         if not claim.lease_token:
             self.record_release_failure()
             return False
-        outcome = observation.outcome if observation.attempted else "skipped"
-        delay = self._next_delay(outcome, claim.consecutive_failures)
+        if observation.outcome == "deferred":
+            # 本地限流没有发出请求，不累计失败；只按站点窗口写入冷却供其它订阅共享退避。
+            outcome = "deferred"
+            delay = max(0.0, deferral_seconds)
+        else:
+            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")
         with self._outcome_lock:
             paced = claim.lease_token in self._paced_leases

app/chain/search/facade.py

@@ -520,6 +520,7 @@ class SearchChain(ChainBase):
     _torrent_keyword = staticmethod(SearchProviderOwner._torrent_keyword)
     _torrent_type = staticmethod(SearchProviderOwner._torrent_type)
     _search_site_torrents_with_budget = SearchProviderOwner._search_site_torrents_with_budget
+    _settle_site_search_result = SearchProviderOwner._settle_site_search_result
     _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

@@ -11,6 +11,7 @@ from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Iterator, Lis
 
 from app.application.configuration import get_configured_system_config
 from app.application.site.observation import (
+    SiteSearchObservation,
     capture_site_search_observation,
     report_site_search_outcome,
 )
@@ -74,6 +75,20 @@ def _site_request_interval(site: SiteIndexer) -> float:
         return 0.0
 
 
+def _site_rate_limit_backoff(site: SiteIndexer) -> float:
+    """本地流控拒绝后的退避秒数：覆盖访问间隔与计数窗口，并留 1 秒越过窗口的严格比较边界。"""
+    interval = _site_request_interval(site)
+    window = 0.0
+    if site.get("limit_interval") and site.get("limit_count"):
+        try:
+            window = max(0.0, float(site.get("limit_interval") or 0))
+        except (TypeError, ValueError):
+            window = 0.0
+    wait = max(interval, window)
+    # 计数窗口按 `current - last > interval` 清零，等于窗口时长时仍可能被拒，所以多等 1 秒。
+    return wait + 1.0 if wait > 0 else 0.0
+
+
 def _reserve_site_request(site: SiteIndexer) -> float:
     """按站点原子预约请求起始时间，避免并发搜索绕过同一流控窗口。"""
     interval = _site_request_interval(site)
@@ -247,6 +262,34 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
         )
         pending[future] = (site, page_index, page_number)
 
+    def _settle_site_search_result(
+        self,
+        *,
+        budget: SubscriptionSiteBudget,
+        site: SiteIndexer,
+        site_id: int,
+        result: List[TorrentInfo],
+        observation: SiteSearchObservation,
+    ) -> float:
+        """记录一次已返回站点页的候选与限流事实，返回本地流控需要的窗口退避秒数。"""
+        activity = current_activity_run()
+        if activity:
+            activity.site(site_id, name=str(site.get("name") or ""), candidates=len(result or []),
+                          status=observation.outcome if observation.attempted else "received", attempted=observation.attempted)
+        budget.record_request(site_id, len(result or []))
+        if observation.attempted:
+            if observation.outcome == "success":
+                budget.record_success(site_id)
+            elif observation.outcome != "skipped":
+                failure = observation.error or observation.outcome
+                message = f"站点 {site.get('name') or site_id} 搜索失败：{failure}"
+                self.record_subscription_site_budget_failure(message)
+                logger.warning(message)
+        elif observation.outcome == "deferred":
+            # 本地流控未发出请求，按站点窗口写入冷却，避免其它订阅继续触发限流。
+            return _site_rate_limit_backoff(site)
+        return 0.0
+
     def _search_site_torrents_with_budget(
         self,
         *,
@@ -283,6 +326,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
             logger.debug(str(error))
             return []
         with capture_site_search_observation() as observation:
+            deferral_seconds = 0.0
             try:
                 result = _search_site_page(
                     self,
@@ -302,22 +346,17 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
                 )
                 raise
             else:
-                if activity:
-                    activity.site(site_id, name=str(site.get("name") or ""), candidates=len(result or []),
-                                  status=observation.outcome if observation.attempted else "received", attempted=observation.attempted)
-                budget.record_request(site_id, len(result or []))
-                if observation.attempted:
-                    if observation.outcome == "success":
-                        budget.record_success(site_id)
-                    elif observation.outcome != "skipped":
-                        failure = observation.error or observation.outcome
-                        message = f"站点 {site.get('name') or site_id} 搜索失败：{failure}"
-                        self.record_subscription_site_budget_failure(message)
-                        logger.warning(message)
+                deferral_seconds = self._settle_site_search_result(
+                    budget=budget,
+                    site=site,
+                    site_id=site_id,
+                    result=result,
+                    observation=observation,
+                )
                 return result
             finally:
                 try:
-                    released = budget.finish(claim, observation)
+                    released = budget.finish(claim, observation, deferral_seconds=deferral_seconds)
                     if not released:
                         logger.error(
                             f"订阅站点预算释放失败: site_id={site_id} "

app/chain/search/scan.py

@@ -28,7 +28,12 @@ from app.application.subscription.sitebudget import (
 from app.chain.base import ChainBase
 from app.chain.search.contract import _SearchOwnerBase
 from app.chain.search.execution import MediaSearchPlan
-from app.chain.search.provider import _site_keyword, _site_request_interval, _wait_for_site_request
+from app.chain.search.provider import (
+    _site_keyword,
+    _site_rate_limit_backoff,
+    _site_request_interval,
+    _wait_for_site_request,
+)
 from app.chain.search.result import DisambiguationCache, SearchResultOwner, _resource_targets, stored_meta
 from app.domain.context import Context, MediaInfo, TorrentInfo
 from app.domain.episode import format_ranges
@@ -142,13 +147,16 @@ class SearchSources:
         else:
             _wait_for_site_request(site)
         with capture_site_search_observation() as observation:
+            deferral_seconds = 0.0
             try:
                 items = self.owner.search_site_torrents(site=site, keyword=query["keyword"],
                                                        mtype=media_type, page=page) or []
                 if self.activity:
                     self.activity.site(int(query["site"]), name=str(site.get("name") or ""), candidates=len(items), status=observation.outcome, attempted=observation.attempted)
                 if observation.outcome == "deferred":
-                    retry_at = (datetime.now(timezone.utc) + timedelta(seconds=AUTOMATIC_REQUEST_INTERVAL)).isoformat(timespec="seconds")
+                    # 本地流控未发出请求，按站点窗口退避，避免在限流窗口内以秒级间隔反复重试。
+                    deferral_seconds = _site_rate_limit_backoff(site) or float(AUTOMATIC_REQUEST_INTERVAL)
+                    retry_at = (datetime.now(timezone.utc) + timedelta(seconds=deferral_seconds)).isoformat(timespec="seconds")
                     return ScanPage(items, observation, retry_at, "cooldown")
                 if claim and budget:
                     budget.record_request(claim.site_id, len(items))
@@ -164,7 +172,7 @@ class SearchSources:
                 return ScanPage([], observation)
             finally:
                 if claim and budget:
-                    budget.finish(claim, observation)
+                    budget.finish(claim, observation, deferral_seconds=deferral_seconds)
 
 
 class SearchScan:

app/db/oper/subscriptionsearch.py

@@ -644,7 +644,8 @@ class SubscriptionSearchOper(DbOper):
         if outcome == "success":
             failures = max(0, record.consecutive_failures - 1)
             success_streak = record.success_streak + 1
-        elif outcome == "skipped":
+        elif outcome in {"skipped", "deferred"}:
+            # 本地流控拒绝没有发出请求，不累计失败，但保留未来冷却时间。
             failures = record.consecutive_failures
             success_streak = record.success_streak
         else:

tests/test_search_scan.py

@@ -443,6 +443,78 @@ def test_site_window_wait_keeps_cursor(owner, monkeypatch):
     assert scan.collection.sources[key].error == "站点窗口限流"
 
 
+def test_site_rate_limit_window_defers_retry_until_window_refresh(owner, monkeypatch):
+    """站点触发计数窗口限流时按窗口长度退避，不能继续用秒级冷却反复重试。"""
+    monkeypatch.setattr(owner, "_sync_indexers", lambda _sites: [
+        {"id": 1, "name": "test", "parser": "mTorrent",
+         "limit_interval": 3600, "limit_count": 25, "limit_seconds": 20}
+    ])
+    def request(**_params):
+        report_site_search_outcome(
+            attempted=False, outcome="deferred",
+            error="触发流控规则，3600 秒内访问次数不得超过 25 次",
+        )
+        return []
+    monkeypatch.setattr(owner, "search_site_torrents", request)
+    scan = SearchScan(owner=owner, plan=plan([1]))
+    key = next(key for key in scan.queries if not key.startswith("plugin:"))
+    before = datetime.now(timezone.utc)
+
+    assert not scan.step(key)
+
+    assert datetime.fromisoformat(scan.retry[key]) >= before + timedelta(seconds=3600)
+    assert scan.retry_reasons[key] == "cooldown"
+    assert not scan.collection.sources[key].failed
+    assert not scan.collection.sources[key].exhausted
+
+
+def test_scan_persists_local_rate_limit_window_on_site_budget(owner, monkeypatch):
+    """本地流控退避必须写入站点预算，让其它订阅在窗口内共享同一冷却。"""
+    from app.application.subscription.sitebudget import SiteBudgetClaim, SubscriptionSiteBudget
+    from app.runtime.stop import ProcessStopState
+
+    captured = {}
+
+    class _Repository:
+        """发放独占租约并记录收口参数。"""
+
+        def claim_site(self, *, site_id, owner, lease_seconds, minimum_interval=0):
+            del owner, lease_seconds, minimum_interval
+            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):
+            captured.update(kwargs)
+            return True
+
+    monkeypatch.setattr(owner, "_sync_indexers", lambda _sites: [
+        {"id": 1, "name": "test", "parser": "mTorrent",
+         "limit_interval": 3600, "limit_count": 25, "limit_seconds": 20}
+    ])
+    def request(**_params):
+        report_site_search_outcome(attempted=False, outcome="deferred", error="触发流控规则")
+        return []
+    monkeypatch.setattr(owner, "search_site_torrents", request)
+    owner.configure_subscription_site_budget(SubscriptionSiteBudget(
+        repository=_Repository(), owner="scan-worker", cancelled=lambda: False,
+        stop_state=ProcessStopState(),
+    ))
+    scan = SearchScan(owner=owner, plan=plan([1]))
+    key = next(key for key in scan.queries if not key.startswith("plugin:"))
+    before = datetime.now(timezone.utc)
+
+    assert not scan.step(key)
+
+    assert captured["outcome"] == "deferred"
+    assert captured["error"] == "触发流控规则"
+    assert datetime.fromisoformat(captured["next_allowed_at"]) >= before + timedelta(seconds=3600)
+
+
 def test_defer_ignores_expired_retries_and_terminal_sources(owner, monkeypatch):
     """失败或耗尽来源的检查点残留不能覆盖仍活跃来源的冷却时间。"""
     from app.application.subscription.sitebudget import SubscriptionSearchDeferred

tests/test_subscription_site_budget.py

@@ -291,6 +291,78 @@ def test_skipped_search_releases_budget_without_external_interval():
     assert captured["next_allowed_at"] == now.isoformat(timespec="seconds")
 
 
+def test_local_rate_limit_defers_site_budget_for_full_window():
+    """本地流控拒绝必须按站点窗口写入冷却，而不是立即恢复后再次触发限流。"""
+    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=2,
+        lease_token="lease-token",
+    )
+
+    assert budget.finish(
+        claim,
+        SiteSearchObservation(attempted=False, outcome="deferred", error="触发流控规则"),
+        deferral_seconds=3601,
+    ) is True
+    assert captured["outcome"] == "deferred"
+    assert captured["error"] == "触发流控规则"
+    assert captured["next_allowed_at"] == (now + timedelta(seconds=3601)).isoformat(timespec="seconds")
+
+
+def test_deferred_site_cooldown_blocks_other_subscriptions_without_failure_growth(tmp_path):
+    """窗口冷却中的站点对其它订阅同样不可认领，且本地限流不计入失败次数。"""
+    repository, engine = _repository(tmp_path)
+    claim = repository.claim_site(site_id=30, owner="worker-a", lease_seconds=900)
+    future = (datetime.now(timezone.utc) + timedelta(seconds=3601)).isoformat(timespec="seconds")
+
+    assert repository.finish_site(
+        site_id=30,
+        lease_token=claim.lease_token,
+        outcome="deferred",
+        next_allowed_at=future,
+        error="触发流控规则，3600 秒内访问次数不得超过 25 次",
+    ) is True
+
+    waiting = repository.claim_site(site_id=30, owner="worker-b", lease_seconds=900)
+    assert waiting.acquired is False
+    assert waiting.wait_reason == "cooldown"
+    assert waiting.retry_at == future
+    with Session(engine) as session:
+        record = session.execute(
+            select(SiteBudgetRecord).where(SiteBudgetRecord.site_id == 30)
+        ).scalar_one()
+        assert record.consecutive_failures == 0
+        assert record.success_streak == 0
+        assert record.last_outcome == "deferred"
+        record.next_allowed_at = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat(timespec="seconds")
+        session.commit()
+
+    recovered = repository.claim_site(site_id=30, owner="worker-b", lease_seconds=900)
+    assert recovered.acquired is True
+    assert recovered.consecutive_failures == 0
+    engine.dispose()
+
+
 def test_search_provider_reports_cooled_site_without_blocking_other_results():
     """错误冷却中的站点返回空页并记录延后，而非阻塞 provider 或制造失败。"""
     repository = _WaitingRepository()
@@ -387,6 +459,60 @@ def test_search_provider_releases_successful_site_budget():
     assert not chain.consume_subscription_site_budget_failures()
 
 
+def test_search_provider_defers_budget_for_local_rate_limit_window():
+    """同步 provider 遇到本地流控拒绝时同样按站点窗口写入冷却，且不算站点失败。"""
+    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(),
+    )
+    chain = object.__new__(SearchChain)
+    chain.configure_subscription_site_budget(budget)
+
+    def search_site_torrents(**_kwargs):
+        """模拟索引器被本地流控拒绝，未发出任何请求。"""
+        report_site_search_outcome(attempted=False, outcome="deferred", error="触发流控规则")
+        return []
+
+    chain.search_site_torrents = search_site_torrents
+    before = datetime.now(timezone.utc)
+
+    result = chain._search_site_torrents_with_budget(  # pylint: disable=protected-access
+        site={"id": 16, "name": "Limited", "limit_interval": 3600, "limit_count": 25, "limit_seconds": 20},
+        keyword="movie",
+        mtype=None,
+        page=0,
+    )
+
+    assert result == []
+    assert captured["outcome"] == "deferred"
+    assert captured["error"] == "触发流控规则"
+    assert datetime.fromisoformat(captured["next_allowed_at"]) >= before + timedelta(seconds=3600)
+    assert chain.consume_subscription_site_budget_failures() == ()
+
+
 def test_other_search_results_keep_site_failures_non_terminal():
     """插件等其他搜索源已有候选时，站点失败不得覆盖可用结果。"""
     chain = object.__new__(SearchChain)