diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index d066c7462..87496012c 100644 --- a/app/application/subscription/sitebudget.py +++ b/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 diff --git a/app/chain/search/facade.py b/app/chain/search/facade.py index 891f5f02b..0f1f8a5ef 100644 --- a/app/chain/search/facade.py +++ b/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 diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index 6f27a652e..6b5b797a5 100644 --- a/app/chain/search/provider.py +++ b/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} " diff --git a/app/chain/search/scan.py b/app/chain/search/scan.py index 0b691fa26..abc74e171 100644 --- a/app/chain/search/scan.py +++ b/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: diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py index 5e78ccd1f..d5635bc2e 100644 --- a/app/db/oper/subscriptionsearch.py +++ b/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: diff --git a/tests/test_search_scan.py b/tests/test_search_scan.py index 63adfbee1..37a6c21ac 100644 --- a/tests/test_search_scan.py +++ b/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 diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index dce984e85..582f7a04f 100644 --- a/tests/test_subscription_site_budget.py +++ b/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)