Merge pull request #7189 from jxxghp/codex/issue-7187

fix(search): 按实际流控原因计算站点剩余等待时间

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,

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(

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:

docs/testing.md

@@ -173,6 +173,14 @@ 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`）的主动搜索始终使用洗版规则组，
 包括尚无下载记录或仍有缺集的订阅；未通过洗版规则的整包不会回落到普通订阅规则组。
 订阅显式指定的规则组仍优先。分集洗版中从未下载的剧集继续使用普通订阅规则。

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):

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,55 @@ 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.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):
     """

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():
     """
     本地流控拒绝必须按站点窗口写入冷却，而不是立即恢复后再次触发限流。
@@ -525,10 +559,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 +611,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 +627,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() == ()