From 251513f580f226f4a322bbc312dbb6222c3fc228 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 26 Oct 2026 09:02:20 +0800 Subject: [PATCH 1/3] fix(search): honor the actual site rate limit when deferring Refs #7187 --- app/chain/search/provider.py | 37 +++++++++++++++++++------ app/chain/search/scan.py | 4 +-- docs/testing.md | 6 ++++ tests/test_search_scan.py | 16 ++++++++--- tests/test_search_site_throttle.py | 38 ++++++++++++++++++++++++++ tests/test_subscription_site_budget.py | 19 +++++++++---- 6 files changed, 101 insertions(+), 19 deletions(-) diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index 2ea82eff8..60f97d4ed 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -89,9 +89,16 @@ def _site_request_interval(site: SiteIndexer) -> float: return 0.0 -def _site_rate_limit_backoff(site: SiteIndexer) -> float: +def _site_rate_limit_backoff(site: SiteIndexer, error: Optional[str] = None) -> float: """ - 本地流控拒绝后的退避秒数：覆盖访问间隔与计数窗口，并留 1 秒越过窗口的严格比较边界。 + 按实际流控规则计算剩余等待时间，未知原因沿用完整配置窗口。 + + 资源包现有接口只返回状态和说明；使用说明中的拒绝原因与本机上次访问时间， + 不读取限流器私有状态。保留 1 秒余量，跨过秒精度时间和计数窗口的严格比较边界。 + + :param site: 本次搜索使用的站点流控配置 + :param error: 资源包返回的本地流控说明；缺失或无法识别时保守等待完整窗口 + :return: 允许重试前的秒数；没有有效流控配置时返回 0 """ interval = _site_request_interval(site) window = 0.0 @@ -100,9 +107,23 @@ def _site_rate_limit_backoff(site: SiteIndexer) -> float: 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 + message = error or "" + if "访问间隔不得小于" in message: + wait = interval + elif "访问次数不得超过" in message: + wait = window + else: + wait = max(interval, window) + return wait + 1.0 if wait > 0 else 0.0 + if wait <= 0: + return 0.0 + try: + # 资源包通过 localtime 输出该时间，按同一进程的本地时区解析，避免重复等待已经过去的时间。 + last_visit = datetime.strptime(message.partition("上次访问时间：")[2].strip(), "%Y-%m-%d %H:%M:%S") + elapsed = max(0.0, time.time() - last_visit.timestamp()) + except (ValueError, OverflowError, OSError): + elapsed = 0.0 + return max(0.0, wait - elapsed) + 1.0 def _reserve_site_request(site: SiteIndexer) -> float: @@ -325,7 +346,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): observation: SiteSearchObservation, ) -> float: """ - 记录一次已返回站点页的候选与限流事实，返回本地流控需要的窗口退避秒数。 + 记录一次已返回站点页的候选与限流事实，返回实际触发规则的剩余退避秒数。 """ activity = current_activity_run() if activity: @@ -341,8 +362,8 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): self.record_subscription_site_budget_failure(message) logger.warning(message) elif observation.outcome == "deferred": - # 本地流控未发出请求，按站点窗口写入冷却，避免其它订阅继续触发限流。 - return _site_rate_limit_backoff(site) + # 本地流控未发出请求，只等待实际受限规则，避免短间隔误占整个计数窗口。 + return _site_rate_limit_backoff(site, observation.error) return 0.0 def _search_site_torrents_with_budget( diff --git a/app/chain/search/scan.py b/app/chain/search/scan.py index c6f75d0ca..e9bd1a221 100644 --- a/app/chain/search/scan.py +++ b/app/chain/search/scan.py @@ -180,8 +180,8 @@ class SearchSources: 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": - # 本地流控未发出请求，按站点窗口退避，避免在限流窗口内以秒级间隔反复重试。 - deferral_seconds = _site_rate_limit_backoff(site) or float(AUTOMATIC_REQUEST_INTERVAL) + # 本地流控未发出请求，只按实际受限规则退避，不能把短间隔扩大为小时计数窗口。 + deferral_seconds = _site_rate_limit_backoff(site, observation.error) 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: diff --git a/docs/testing.md b/docs/testing.md index 6de7cad14..66190cf7f 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -173,6 +173,12 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)): 和虚拟时钟覆盖秒级冷却下的连续恢复，验证剩余来源继续翻页、末页候选处理后正常收尾， 且未收录缺集仍保留。新一轮搜索或缺集范围扩大时重新评估候选；旧检查点缺少该记录时兼容恢复。 +本地流控等待由实际拒绝原因决定：访问间隔不足只等待间隔，计数配额耗尽才等待计数窗口。 +资源包说明包含上次访问时间时扣除已过去的时间，并保留一秒边界余量；原因未知时沿用保守窗口， +时间缺失或无效时等待该规则全长。`tests/test_search_site_throttle.py`、`tests/test_search_scan.py` +和 `tests/test_subscription_site_budget.py` 覆盖两类限制同时配置、剩余时间、时间异常与共享预算， +验证短冷却不累计长冷却止损，也不会提前关闭优先站点。 + 全集洗版（`best_version=1`、`best_version_full=1`）的主动搜索始终使用洗版规则组， 包括尚无下载记录或仍有缺集的订阅；未通过洗版规则的整包不会回落到普通订阅规则组。 订阅显式指定的规则组仍优先。分集洗版中从未下载的剧集继续使用普通订阅规则。 diff --git a/tests/test_search_scan.py b/tests/test_search_scan.py index aa1c3bf88..ce937a53a 100644 --- a/tests/test_search_scan.py +++ b/tests/test_search_scan.py @@ -619,10 +619,16 @@ 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): +@pytest.mark.parametrize("reason, seconds", [ + ("访问间隔不得小于 20 秒", 21), + ("3600 秒内访问次数不得超过 25 次", 3601), + ("未知限制", 3601), +]) +def test_site_rate_limit_window_defers_retry_until_window_refresh(owner, monkeypatch, reason, seconds): """ - 站点触发计数窗口限流时按窗口长度退避，不能继续用秒级冷却反复重试。 + 分页搜索按实际限流原因等待，短间隔不扩大为计数窗口，也不提前结束来源。 """ + monkeypatch.setattr(scan_module, "_wait_for_site_request", lambda _site: None) monkeypatch.setattr(owner, "_sync_indexers", lambda _sites: [ {"id": 1, "name": "test", "parser": "mTorrent", "limit_interval": 3600, "limit_count": 25, "limit_seconds": 20} @@ -630,7 +636,7 @@ def test_site_rate_limit_window_defers_retry_until_window_refresh(owner, monkeyp def request(**_params): report_site_search_outcome( attempted=False, outcome="deferred", - error="触发流控规则，3600 秒内访问次数不得超过 25 次", + error=f"触发流控规则，{reason}", ) return [] monkeypatch.setattr(owner, "search_site_torrents", request) @@ -640,10 +646,12 @@ def test_site_rate_limit_window_defers_retry_until_window_refresh(owner, monkeyp assert not scan.step(key) - assert datetime.fromisoformat(scan.retry[key]) >= before + timedelta(seconds=3600) + delay = (datetime.fromisoformat(scan.retry[key]) - before).total_seconds() + assert seconds - 1 < delay <= seconds assert scan.retry_reasons[key] == "cooldown" assert not scan.collection.sources[key].failed assert not scan.collection.sources[key].exhausted + assert scan.collection.sources[key].budget_deferrals == (1 if seconds >= 600 else 0) def test_scan_persists_local_rate_limit_window_on_site_budget(owner, monkeypatch): diff --git a/tests/test_search_site_throttle.py b/tests/test_search_site_throttle.py index e8e456bee..6082dfa0a 100644 --- a/tests/test_search_site_throttle.py +++ b/tests/test_search_site_throttle.py @@ -2,6 +2,8 @@ 站点搜索请求间隔调度测试。 """ +from datetime import datetime + import pytest from app.chain.search import provider @@ -48,6 +50,42 @@ def test_invalid_interval_does_not_throttle(monkeypatch): assert provider._reserve_site_request(site) == 0 +@pytest.mark.parametrize("interval, reason, elapsed, expected", [ + (20, "访问间隔不得小于 20 秒", 7, 14), + (30, "访问间隔不得小于 30 秒", 7, 24), + (20, "3600 秒内访问次数不得超过 25 次", 7, 3594), + (20, "3600 秒内访问次数不得超过 25 次", 3600, 1), + (20, "访问间隔不得小于 20 秒", 25, 1), + (20, "访问间隔不得小于 20 秒", -5, 21), + (20, "未知限制", 7, 3601), +]) +def test_local_limit_backoff_uses_actual_rule_and_remaining_time(monkeypatch, interval, reason, elapsed, expected): + """ + 访问间隔和计数窗口分别扣除已等待时间，保留边界余量，未知原因沿用保守窗口。 + """ + now = datetime(2026, 10, 26, 12, 0, 0).timestamp() + monkeypatch.setattr(provider.time, "time", lambda: now) + last_visit = datetime.fromtimestamp(now - elapsed).strftime("%Y-%m-%d %H:%M:%S") + message = f"触发流控规则，{reason}，上次访问时间：{last_visit}" + site = {"limit_seconds": interval, "limit_interval": 3600, "limit_count": 25} + + assert provider._site_rate_limit_backoff(site, message) == expected + + +@pytest.mark.parametrize("message, expected", [ + ("触发流控规则，访问间隔不得小于 20 秒", 21), + ("触发流控规则，访问间隔不得小于 20 秒，上次访问时间：invalid", 21), + ("触发流控规则，3600 秒内访问次数不得超过 25 次，上次访问时间：invalid", 3601), + (None, 3601), +]) +def test_local_limit_without_valid_timestamp_keeps_rule_duration(message, expected): + """ + 旧资源缺少有效时间时等待对应规则全长，缺少原因时保留既有窗口回退。 + """ + site = {"limit_seconds": 20, "limit_interval": 3600, "limit_count": 25} + assert provider._site_rate_limit_backoff(site, message) == expected + + @pytest.mark.asyncio async def test_async_wait_only_sleeps_for_throttled_site(monkeypatch): """ diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index 82a291405..c2046fdcc 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -525,10 +525,18 @@ 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(): +@pytest.mark.parametrize("reason, seconds", [ + ("访问间隔不得小于 20 秒", 21), + ("3600 秒内访问次数不得超过 25 次", 3601), + ("未知限制", 3601), +]) +def test_search_provider_defers_budget_for_local_rate_limit_window(monkeypatch, reason, seconds): """ - 同步 provider 遇到本地流控拒绝时同样按站点窗口写入冷却，且不算站点失败。 + 同步 provider 按实际限流原因持久化冷却，短间隔不扩大为小时窗口，且不算站点失败。 """ + from app.chain.search import provider + + monkeypatch.setattr(provider, "_wait_for_site_request", lambda _site: None) captured = {} class _Repository(_WaitingRepository): @@ -569,7 +577,7 @@ def test_search_provider_defers_budget_for_local_rate_limit_window(): """ 模拟索引器被本地流控拒绝，未发出任何请求。 """ - report_site_search_outcome(attempted=False, outcome="deferred", error="触发流控规则") + report_site_search_outcome(attempted=False, outcome="deferred", error=f"触发流控规则，{reason}") return [] chain.search_site_torrents = search_site_torrents @@ -585,8 +593,9 @@ def test_search_provider_defers_budget_for_local_rate_limit_window(): assert result == [] assert captured["outcome"] == "deferred" - assert captured["error"] == "触发流控规则" - assert datetime.fromisoformat(captured["next_allowed_at"]) >= before + timedelta(seconds=3600) + assert captured["error"] == f"触发流控规则，{reason}" + delay = (datetime.fromisoformat(captured["next_allowed_at"]) - before).total_seconds() + assert seconds - 1 < delay <= seconds assert chain.consume_subscription_site_budget_failures() == () From 109a6d283404d99dd6edb70196133cfc3743b882 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 26 Oct 2026 09:11:45 +0800 Subject: [PATCH 2/3] test(search): cover disabled local rate limit rules --- tests/test_search_site_throttle.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/tests/test_search_site_throttle.py b/tests/test_search_site_throttle.py index 6082dfa0a..8f4c5a697 100644 --- a/tests/test_search_site_throttle.py +++ b/tests/test_search_site_throttle.py @@ -86,6 +86,19 @@ def test_local_limit_without_valid_timestamp_keeps_rule_duration(message, expect assert provider._site_rate_limit_backoff(site, message) == expected +@pytest.mark.parametrize("site, reason", [ + ({"limit_seconds": 0, "limit_interval": 3600, "limit_count": 25}, "访问间隔不得小于 20 秒"), + ({"limit_seconds": "invalid", "limit_interval": 3600, "limit_count": 25}, "访问间隔不得小于 20 秒"), + ({"limit_seconds": 20, "limit_interval": "invalid", "limit_count": 25}, "3600 秒内访问次数不得超过 25 次"), + ({"limit_seconds": 20, "limit_interval": 3600, "limit_count": 0}, "3600 秒内访问次数不得超过 25 次"), +]) +def test_inactive_or_invalid_reported_rule_does_not_invent_cooldown(site, reason): + """ + 资源包说明与当前配置不一致时，已停用或不可读的规则不产生额外冷却，沿用调用方的默认间隔。 + """ + assert provider._site_rate_limit_backoff(site, f"触发流控规则，{reason}") == 0 + + @pytest.mark.asyncio async def test_async_wait_only_sleeps_for_throttled_site(monkeypatch): """ From 8f21e12d755c6b1b7859c2a1ce839c51593b799c Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 26 Oct 2026 09:23:29 +0800 Subject: [PATCH 3/3] fix(search): release unused site request reservations --- app/application/subscription/sitebudget.py | 8 ++--- docs/testing.md | 2 ++ tests/test_subscription_site_budget.py | 34 ++++++++++++++++++++++ 3 files changed, 40 insertions(+), 4 deletions(-) diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index 0dc533da4..a42ac1319 100644 --- a/app/application/subscription/sitebudget.py +++ b/app/application/subscription/sitebudget.py @@ -272,7 +272,7 @@ class SubscriptionSiteBudget: self._metrics = metrics self._outcome_lock = threading.Lock() self._successful_site_ids: set[int] = set() - # 带最小请求间隔认领的租约；释放时下次允许时间不早于认领时预留的开始间隔。 + # 带最小请求间隔认领的租约；只有实际发出请求，释放后才保留预留的开始间隔。 self._paced_leases: set[str] = set() def acquire(self, site_id: int, *, minimum_interval: float = 0) -> SiteBudgetClaim: @@ -341,13 +341,13 @@ class SubscriptionSiteBudget: def finish(self, claim: SiteBudgetClaim, observation: SiteSearchObservation, *, deferral_seconds: float = 0) -> bool: """ - 正常结果立即释放站点，错误结果按类别写入冷却，本地流控按站点窗口退避。 + 释放站点租约，真实请求保留最小间隔或错误退避，本地拒绝只等待实际剩余流控时间。 """ if not claim.lease_token: self.record_release_failure() return False if observation.outcome == "deferred": - # 本地限流没有发出请求，不累计失败；只按站点窗口写入冷却供其它订阅共享退避。 + # 本地限流没有发出请求，不累计失败；只写入实际剩余冷却供其它订阅共享退避。 outcome = "deferred" delay = max(0.0, deferral_seconds) else: @@ -357,7 +357,7 @@ class SubscriptionSiteBudget: with self._outcome_lock: paced = claim.lease_token in self._paced_leases self._paced_leases.discard(claim.lease_token) - if paced and claim.retry_at: + if paced and claim.retry_at and observation.attempted: next_allowed_at = max(next_allowed_at, claim.retry_at) released = self._repository.finish_site( site_id=claim.site_id, diff --git a/docs/testing.md b/docs/testing.md index 66190cf7f..d5e366bad 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -178,6 +178,8 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)): 时间缺失或无效时等待该规则全长。`tests/test_search_site_throttle.py`、`tests/test_search_scan.py` 和 `tests/test_subscription_site_budget.py` 覆盖两类限制同时配置、剩余时间、时间异常与共享预算， 验证短冷却不累计长冷却止损，也不会提前关闭优先站点。 +自动分页未实际发出请求时，共享预算取消完整请求间隔的预约，只保留实际剩余冷却； +真实成功请求仍保留最小间隔。该收口使用临时 SQLite 验证跨订阅可见的预算时间和租约释放。 全集洗版（`best_version=1`、`best_version_full=1`）的主动搜索始终使用洗版规则组， 包括尚无下载记录或仍有缺集的订阅；未通过洗版规则的整包不会回落到普通订阅规则组。 diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index c2046fdcc..9490075a4 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -335,6 +335,40 @@ def test_skipped_search_releases_budget_without_external_interval(): assert captured["next_allowed_at"] == now.isoformat(timespec="seconds") +@pytest.mark.parametrize("outcome, attempted, delay", [ + ("deferred", False, 6), + ("deferred", False, 3601), + ("skipped", False, 0), + ("success", True, 0), +]) +def test_request_reservation_preserves_only_actual_request_interval(tmp_path, outcome, attempted, delay): + """ + 未发请求时按实际剩余时间释放预约，真实成功请求仍受完整访问间隔约束。 + """ + repository, engine = _repository(tmp_path) + now = datetime.now(timezone.utc) + budget = SubscriptionSiteBudget( + repository=repository, owner="scan", cancelled=lambda: False, + stop_state=ProcessStopState(), clock=lambda: now, + ) + try: + claim = budget.acquire(15, minimum_interval=20) + assert claim.acquired + assert budget.finish( + claim, SiteSearchObservation(attempted=attempted, outcome=outcome), + deferral_seconds=delay, + ) + with Session(engine) as session: + record = session.execute(select(SiteBudgetRecord).where(SiteBudgetRecord.site_id == 15)).scalar_one() + expected = claim.retry_at if attempted else (now + timedelta(seconds=delay)).isoformat(timespec="seconds") + assert record.next_allowed_at == expected + assert record.last_outcome == outcome + assert record.consecutive_failures == 0 + assert record.lease_token is None + finally: + engine.dispose() + + def test_local_rate_limit_defers_site_budget_for_full_window(): """ 本地流控拒绝必须按站点窗口写入冷却，而不是立即恢复后再次触发限流。