From bab1d2862ed855344469b469aacc54fc149b7232 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 1 Sep 2026 13:57:07 +0800 Subject: [PATCH] refactor(subscribe): govern fallback site pressure --- app/application/site/search_observation.py | 50 ++++ app/application/subscription/execution.py | 28 +- app/application/subscription/sitebudget.py | 156 ++++++++++ app/chain/search/contract.py | 3 + app/chain/search/facade.py | 30 ++ app/chain/search/provider.py | 67 ++++- app/chain/subscribe/search.py | 42 ++- app/db/adapters/subscriptionsearch.py | 55 ++++ app/db/models/__init__.py | 4 + app/db/models/subscriptionsearch.py | 28 ++ app/db/oper/subscriptionsearch.py | 159 +++++++++++ app/modules/indexer/__init__.py | 49 +++- database/versions/d2a7c5e9f1b4_3_0_20.py | 95 +++++++ docs/architecture-overview.md | 4 +- docs/architecture/optimization-checklist.md | 2 +- ...iepilot-subscription-governance-roadmap.md | 24 +- .../architecture/dependency-baseline.json | 25 +- tests/test_subscription_search_governance.py | 2 + tests/test_subscription_site_budget.py | 266 ++++++++++++++++++ 19 files changed, 1071 insertions(+), 18 deletions(-) create mode 100644 app/application/site/search_observation.py create mode 100644 app/application/subscription/sitebudget.py create mode 100644 database/versions/d2a7c5e9f1b4_3_0_20.py create mode 100644 tests/test_subscription_site_budget.py diff --git a/app/application/site/search_observation.py b/app/application/site/search_observation.py new file mode 100644 index 000000000..32b6c66a6 --- /dev/null +++ b/app/application/site/search_observation.py @@ -0,0 +1,50 @@ +"""站点搜索调用的轻量结果观察上下文。""" + +from __future__ import annotations + +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from typing import Iterator, Optional + + +@dataclass(slots=True) +class SiteSearchObservation: + """记录一次站点搜索是否发出请求及其可治理结果。""" + + attempted: bool = False + outcome: str = "skipped" + error: Optional[str] = None + + +_current_observation: ContextVar[Optional[SiteSearchObservation]] = ContextVar( + "site_search_observation", + default=None, +) + + +@contextmanager +def capture_site_search_observation() -> Iterator[SiteSearchObservation]: + """为当前同步 worker 捕获一次索引器搜索结果。""" + observation = SiteSearchObservation() + token = _current_observation.set(observation) + try: + yield observation + finally: + _current_observation.reset(token) + + +def report_site_search_outcome( + *, + attempted: bool, + outcome: str, + error: Optional[str] = None, +) -> None: + """由索引器在不改变公开返回合同的前提下发布调用结果。""" + observation = _current_observation.get() + if observation is None: + return + observation.attempted = attempted + observation.outcome = outcome + observation.error = error + diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py index 2cff3c333..18d9044e1 100644 --- a/app/application/subscription/execution.py +++ b/app/application/subscription/execution.py @@ -3,6 +3,8 @@ from dataclasses import dataclass from typing import Optional, Protocol +from app.application.subscription.sitebudget import SiteBudgetClaim + @dataclass(frozen=True, slots=True) class SearchBatchSnapshot: @@ -40,6 +42,7 @@ class SearchTaskSnapshot: lease_token: Optional[str] created_at: str updated_at: str + available_at: Optional[str] = None started_at: Optional[str] = None finished_at: Optional[str] = None last_error: Optional[str] = None @@ -63,8 +66,9 @@ class SubscriptionSearchRepository(Protocol): subscription_ids: tuple[int, ...], source: str, priority: int, + available_at: Optional[str] = None, ) -> SearchEnqueueResult: - """按订阅 ID 建立批次，并合并已存在的活动任务。""" + """按订阅 ID 建立批次，在启动抖动后合并活动任务。""" ... def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]: @@ -103,3 +107,25 @@ class SubscriptionSearchRepository(Protocol): def get_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]: """按稳定批次 ID 返回当前聚合状态。""" ... + + def claim_site( + self, + *, + site_id: int, + owner: str, + lease_seconds: int, + ) -> SiteBudgetClaim: + """认领一个站点的唯一在途搜索预算。""" + ... + + def finish_site( + self, + *, + site_id: int, + lease_token: str, + outcome: str, + next_allowed_at: str, + error: Optional[str] = None, + ) -> bool: + """释放站点预算并写入间隔、冷却和恢复状态。""" + ... diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py new file mode 100644 index 000000000..e89f4a5b7 --- /dev/null +++ b/app/application/subscription/sitebudget.py @@ -0,0 +1,156 @@ +"""订阅兜底搜索的站点级容量、间隔与冷却治理。""" + +from __future__ import annotations + +import random +import time +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Callable, Optional, Protocol + +from app.application.site.search_observation import SiteSearchObservation +from app.runtime.stop import StopState + + +class SubscriptionSearchCancelled(RuntimeError): + """表示订阅搜索在可取消预算等待点终止。""" + + +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 + + +@dataclass(frozen=True, slots=True) +class SiteBudgetClaim: + """一次站点预算认领结果或下一次可尝试时间。""" + + site_id: int + acquired: bool + retry_at: str + consecutive_failures: int + lease_token: Optional[str] = None + + +class SubscriptionSiteBudgetRepository(Protocol): + """站点预算租约和冷却状态的持久化端口。""" + + def claim_site( + self, + *, + site_id: int, + owner: str, + lease_seconds: int, + ) -> SiteBudgetClaim: + """认领站点唯一在途租约，未就绪时返回重试时间。""" + ... + + def finish_site( + self, + *, + site_id: int, + lease_token: str, + outcome: str, + next_allowed_at: str, + error: Optional[str] = None, + ) -> bool: + """收口当前站点租约并持久化下次允许时间。""" + ... + + +def _utc_now() -> datetime: + """返回带时区的 UTC 当前时间。""" + return datetime.now(timezone.utc) + + +class SubscriptionSiteBudget: + """在同步搜索 worker 内执行可取消的站点预算等待与反馈。""" + + def __init__( + self, + *, + repository: SubscriptionSiteBudgetRepository, + owner: str, + cancelled: Callable[[], bool], + stop_state: StopState, + interval_range: tuple[float, float] = (60.0, 300.0), + lease_seconds: int = 900, + max_wait_seconds: float = 5.0, + random_uniform: Callable[[float, float], float] = random.uniform, + sleeper: Callable[[float], None] = time.sleep, + clock: Callable[[], datetime] = _utc_now, + ) -> None: + """保存持久化端口及可注入的时钟、随机数和等待实现。""" + self._repository = repository + self._owner = owner + self._cancelled = cancelled + self._stop_state = stop_state + self._interval_range = interval_range + self._lease_seconds = max(1, lease_seconds) + self._max_wait_seconds = max(0.0, max_wait_seconds) + self._random_uniform = random_uniform + self._sleeper = sleeper + self._clock = clock + + def acquire(self, site_id: int) -> SiteBudgetClaim: + """循环认领指定站点，并在每秒边界检查取消与停机。""" + deadline = time.monotonic() + self._max_wait_seconds + while True: + self._raise_if_cancelled() + claim = self._repository.claim_site( + site_id=site_id, + owner=self._owner, + lease_seconds=self._lease_seconds, + ) + if claim.acquired: + return claim + retry_at = datetime.fromisoformat(claim.retry_at) + remaining = max(0.0, (retry_at - self._clock()).total_seconds()) + if remaining > max(0.0, deadline - time.monotonic()): + raise SubscriptionSiteBudgetUnavailable( + site_id=site_id, + retry_at=claim.retry_at, + ) + self._sleeper(min(max(remaining, 0.05), 1.0)) + + def finish(self, claim: SiteBudgetClaim, observation: SiteSearchObservation) -> bool: + """依据调用结果计算随机间隔或错误冷却并释放租约。""" + if not claim.lease_token: + return False + 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") + return self._repository.finish_site( + site_id=claim.site_id, + lease_token=claim.lease_token, + outcome=outcome, + next_allowed_at=next_allowed_at, + error=observation.error, + ) + + def _next_delay(self, outcome: str, consecutive_failures: int) -> float: + """为成功渐进恢复、错误退避和本地跳过计算下一次等待。""" + if outcome == "skipped": + return 0.0 + if outcome == "success": + low, high = self._interval_range + recovery_factor = 1.0 + min(max(consecutive_failures, 0), 3) * 0.5 + return self._random_uniform(low, high) * recovery_factor + exponent = min(max(consecutive_failures, 0), 5) + base, ceiling = { + "rate_limited": (900.0, 21600.0), + "forbidden": (900.0, 21600.0), + "login_invalid": (900.0, 21600.0), + "timeout": (300.0, 7200.0), + }.get(outcome, (180.0, 3600.0)) + return min(base * (2**exponent), ceiling) + + def _raise_if_cancelled(self) -> None: + """在不持有业务锁的等待边界传播取消或停机。""" + if self._stop_state.is_system_stopped or self._cancelled(): + raise SubscriptionSearchCancelled("订阅搜索已取消") diff --git a/app/chain/search/contract.py b/app/chain/search/contract.py index 6e95e8bfd..daaaa56de 100644 --- a/app/chain/search/contract.py +++ b/app/chain/search/contract.py @@ -63,6 +63,7 @@ if TYPE_CHECKING: _restore_original_indices: Callable[..., Any] _save_results: Callable[..., Any] _search_all_sites: Callable[..., Any] + _search_site_torrents_with_budget: Callable[..., Any] _search_state: Callable[..., Any] _select_indexers: Callable[..., Any] _selected_site_ids: Callable[..., Any] @@ -96,6 +97,8 @@ if TYPE_CHECKING: matches_music_resource: Callable[..., Any] music_site_keywords: Callable[..., Any] process: Callable[..., Any] + record_subscription_site_budget_failure: Callable[..., Any] + consume_subscription_site_budget_failures: Callable[..., Any] save_last_search_params: Callable[..., Any] search_by_id: Callable[..., Any] search_by_title: Callable[..., Any] diff --git a/app/chain/search/facade.py b/app/chain/search/facade.py index 76587b466..3ad76144d 100644 --- a/app/chain/search/facade.py +++ b/app/chain/search/facade.py @@ -1,5 +1,6 @@ """搜索处理链稳定 Facade。""" +import threading from collections.abc import AsyncIterator, Callable from typing import Any, Optional, TypeVar, cast @@ -15,6 +16,7 @@ from app.chain.search.result import SearchResultOwner from app.chain.search.site import SearchSiteOwner from app.chain.search.subtitle import SearchSubtitleOwner from app.chain.search.title import SearchTitleOwner +from app.application.subscription.sitebudget import SubscriptionSiteBudget from app.domain.context import Context, MediaInfo, SubtitleInfo from app.runtime.events import Event, eventmanager from app.schemas.mediaserver import NotExistMediaInfo @@ -40,6 +42,33 @@ class SearchChain(ChainBase): _SEARCH_PARAMS_CACHE_KEY = "__search_params__" _AI_INDICES_CACHE_KEY = "__ai_recommend_indices__" + def configure_subscription_site_budget( + self, + budget: Optional[SubscriptionSiteBudget], + ) -> None: + """仅为订阅搜索启用或清除站点预算，不影响其它搜索入口。""" + self._subscription_site_budget = budget + self._subscription_site_budget_failures: list[str] = [] + self._subscription_site_budget_failure_lock = threading.Lock() + + def record_subscription_site_budget_failure(self, error: str) -> None: + """线程安全地记录一个未执行站点，供订阅任务暴露聚合失败。""" + lock = getattr(self, "_subscription_site_budget_failure_lock", None) + if lock is None: + return + with lock: + self._subscription_site_budget_failures.append(error) + + def consume_subscription_site_budget_failures(self) -> tuple[str, ...]: + """读取并清空当前订阅搜索积累的站点预算失败。""" + lock = getattr(self, "_subscription_site_budget_failure_lock", None) + if lock is None: + return () + with lock: + failures = tuple(self._subscription_site_budget_failures) + self._subscription_site_budget_failures.clear() + return failures + # owner descriptor 经类访问后被 mypy 视为普通 Callable；运行时仍需取回原始 # classmethod 函数，才能保持 SearchChain 的直接 MRO 与既有绑定语义。 music_site_keywords = classmethod( # type: ignore[var-annotated] @@ -456,6 +485,7 @@ class SearchChain(ChainBase): _async_indexers = SearchProviderOwner._async_indexers _torrent_keyword = staticmethod(SearchProviderOwner._torrent_keyword) _torrent_type = staticmethod(SearchProviderOwner._torrent_type) + _search_site_torrents_with_budget = SearchProviderOwner._search_site_torrents_with_budget _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 63703f98a..deb60cf3a 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -10,7 +10,15 @@ from datetime import datetime from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Iterator, List, Optional from app.application.configuration import get_configured_system_config +from app.application.site.search_observation import ( + capture_site_search_observation, + report_site_search_outcome, +) from app.application.site.sites import SitesHelper # pylint: disable=import-error,no-name-in-module +from app.application.subscription.sitebudget import ( + SubscriptionSiteBudget, + SubscriptionSiteBudgetUnavailable, +) from app.chain.search.contract import _SearchOwnerBase from app.domain.context import MediaInfo, SubtitleInfo, TorrentInfo from app.runtime.log import logger @@ -182,13 +190,64 @@ class SearchProviderOwner(_SearchOwnerBase): """向进程共享线程 owner 提交一页，并登记该站点的续页位置。""" page_number = search_pages[page_index] future = ThreadHelper().submit( - _search_site_page, - self, - site=site, keyword=search_keyword, - media_type=media_type, page=page_number, + self._search_site_torrents_with_budget, + site=site, + keyword=search_keyword, + mtype=media_type, + page=page_number, ) pending[future] = (site, page_index, page_number) + def _search_site_torrents_with_budget( + self, + *, + site: SiteIndexer, + keyword: str, + mtype: Optional[MediaType], + page: int, + ) -> List[TorrentInfo]: + """在订阅专属站点预算内执行一页同步搜索。""" + budget = getattr(self, "_subscription_site_budget", None) + site_id = site.get("id") + if not isinstance(budget, SubscriptionSiteBudget) or not isinstance(site_id, int): + return _search_site_page( + self, + site=site, + keyword=keyword, + media_type=mtype, + page=page, + ) + try: + claim = budget.acquire(site_id) + except SubscriptionSiteBudgetUnavailable as error: + self.record_subscription_site_budget_failure(str(error)) + logger.info(str(error)) + return [] + with capture_site_search_observation() as observation: + try: + return _search_site_page( + self, + site=site, + keyword=keyword, + media_type=mtype, + page=page, + ) + except Exception as error: + report_site_search_outcome( + attempted=True, + outcome="error", + error=str(error), + ) + raise + finally: + try: + budget.finish(claim, observation) + except Exception as error: # noqa: BLE001 + logger.error( + f"站点 {site.get('name') or site_id} 搜索预算收口失败：{str(error)}", + exc_info=True, + ) + def _submit_next_sync_site( self, *, diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index 77927769b..8edd9e03c 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -2,7 +2,7 @@ import random import time -from datetime import datetime +from datetime import datetime, timedelta, timezone from typing import Any, Callable, Optional, cast from uuid import uuid4 @@ -15,6 +15,10 @@ from app.application.subscription.contract import ( ) from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.execution import SearchBatchSnapshot, SubscriptionSearchRepository +from app.application.subscription.sitebudget import ( + SubscriptionSearchCancelled, + SubscriptionSiteBudget, +) from app.chain.media import MediaChain from app.chain.search.facade import SearchChain from app.chain.subscribe.contract import _SubscribeOwnerBase @@ -209,6 +213,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): subscription_ids=tuple(subscribe.id for subscribe in subscribes), source=source, priority=priority, + available_at=self._search_batch_available_at(source), ) total = len(subscribes) if progress_callback: @@ -293,6 +298,14 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): ) continue current = subscribe + searchchain.configure_subscription_site_budget( + SubscriptionSiteBudget( + repository=queue, + owner=f"{owner}:{task.task_id}", + cancelled=lambda task_id=task.task_id: queue.is_cancel_requested(task_id), + stop_state=getattr(self, "stop_state", runtime_stop_state), + ) + ) try: current = self._process_search_subscription(subscribe, searchchain) if queue.is_cancel_requested(task.task_id): @@ -308,6 +321,12 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): state="completed", ) processed.add(subscribe.id) + except SubscriptionSearchCancelled: + queue.release_task( + task_id=task.task_id, + lease_token=task.lease_token, + cancelled=True, + ) except Exception as err: logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True) queue.finish_task( @@ -317,6 +336,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): error=str(err), ) finally: + searchchain.configure_subscription_site_budget(None) if current and current.state == "N": try: self._SubscribeChain__apply_subscribe_update( @@ -379,8 +399,18 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): return "订阅搜索完成，部分任务失败" if batch.state == "cancelled": return "订阅搜索已取消" + if batch.state in {"queued", "running", "cancelling"}: + return "订阅搜索任务已排队" return "订阅搜索完成" + @staticmethod + def _search_batch_available_at(source: str) -> str: + """为自动兜底批次持久化一次全局启动抖动。""" + delay = random.randint(0, 60) if source == "fallback" else 0 + return ( + datetime.now(timezone.utc) + timedelta(seconds=delay) + ).isoformat(timespec="seconds") + def cancel_search_batch(self, batch_id: str) -> bool: """请求取消持久搜索批次；未注入队列时返回失败。""" queue = cast( @@ -491,6 +521,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): custom_words=subscribe.custom_words.split("\n") if subscribe.custom_words else None, filter_params=self.get_params(subscribe), ) + site_budget_failures = searchchain.consume_subscription_site_budget_failures() if not contexts: logger.warning(f"订阅 {subscribe.keyword or subscribe.name} 未搜索到资源") self.finish_subscribe_or_not( @@ -499,11 +530,13 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): mediainfo=mediainfo, lefts=no_exists, ) + self._raise_site_budget_failures(site_budget_failures) return subscribe matched = self._filter_search_contexts(subscribe, contexts) if not matched: logger.warning(f"订阅 {subscribe.name} 没有符合过滤条件的资源") self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo, lefts=no_exists) + self._raise_site_budget_failures(site_budget_failures) return subscribe downloads, lefts = self._SubscribeChain__download_best_version_with_full_pack_first( contexts=matched, @@ -524,8 +557,15 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): downloads=downloads, lefts=lefts, ) + self._raise_site_budget_failures(site_budget_failures) return cast(Optional[SubscriptionSnapshot], current) + @staticmethod + def _raise_site_budget_failures(failures: tuple[str, ...]) -> None: + """在成功站点结果完成处理后暴露未执行站点的聚合失败。""" + if failures: + raise RuntimeError("；".join(failures)) + def _filter_search_contexts( self, subscribe: SubscriptionSnapshot, diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py index 0a9daf9f9..468135c05 100644 --- a/app/db/adapters/subscriptionsearch.py +++ b/app/db/adapters/subscriptionsearch.py @@ -10,7 +10,9 @@ from app.application.subscription.execution import ( SearchEnqueueResult, SearchTaskSnapshot, ) +from app.application.subscription.sitebudget import SiteBudgetClaim from app.db.models.subscriptionsearch import ( + SubscriptionSiteBudget, SubscriptionSearchBatch, SubscriptionSearchTask, ) @@ -55,6 +57,7 @@ def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot: lease_token=record.lease_token, created_at=record.created_at, updated_at=record.updated_at, + available_at=record.available_at, started_at=record.started_at, finished_at=record.finished_at, last_error=record.last_error, @@ -91,6 +94,7 @@ class TransactionalSubscriptionSearchRepository: subscription_ids: tuple[int, ...], source: str, priority: int, + available_at: Optional[str] = None, ) -> SearchEnqueueResult: """创建批次并返回 single-flight 合并计数。""" def operation(repository: SubscriptionSearchOper) -> SearchEnqueueResult: @@ -99,6 +103,7 @@ class TransactionalSubscriptionSearchRepository: subscription_ids=subscription_ids, source=source, priority=priority, + available_at=available_at, ) return SearchEnqueueResult( batch=_batch(record), @@ -167,3 +172,53 @@ class TransactionalSubscriptionSearchRepository: _batch(record) if (record := repository.get_batch(batch_id)) is not None else None ) ) + + def claim_site( + self, + *, + site_id: int, + owner: str, + lease_seconds: int, + ) -> SiteBudgetClaim: + """认领单站点预算并投影等待或租约事实。""" + def operation(repository: SubscriptionSearchOper) -> SiteBudgetClaim: + """在短事务中认领并复制站点预算状态。""" + record, acquired = repository.claim_site( + site_id=site_id, + owner=owner, + lease_seconds=lease_seconds, + ) + retry_at = ( + record.lease_expires_at + if record.lease_token and not acquired + else record.next_allowed_at + ) or record.next_allowed_at + return SiteBudgetClaim( + site_id=record.site_id, + acquired=acquired, + retry_at=retry_at, + consecutive_failures=record.consecutive_failures, + lease_token=record.lease_token if acquired else None, + ) + + return self._write(operation) + + def finish_site( + self, + *, + site_id: int, + lease_token: str, + outcome: str, + next_allowed_at: str, + error: Optional[str] = None, + ) -> bool: + """释放站点租约并持久化间隔或冷却。""" + return self._write( + lambda repository: repository.finish_site( + site_id=site_id, + lease_token=lease_token, + outcome=outcome, + next_allowed_at=next_allowed_at, + error=error, + ) + ) diff --git a/app/db/models/__init__.py b/app/db/models/__init__.py index 0d19a2c2f..6287ec479 100644 --- a/app/db/models/__init__.py +++ b/app/db/models/__init__.py @@ -43,6 +43,10 @@ _MODEL_EXPORTS = { "app.db.models.subscriptionsearch", "SubscriptionSearchTask", ), + "SubscriptionSiteBudget": ( + "app.db.models.subscriptionsearch", + "SubscriptionSiteBudget", + ), "SystemConfig": ("app.db.models.systemconfig", "SystemConfig"), "TransferHistory": ("app.db.models.transferhistory", "TransferHistory"), "TransferExecutionStep": ( diff --git a/app/db/models/subscriptionsearch.py b/app/db/models/subscriptionsearch.py index f51dc00bc..2dcc711be 100644 --- a/app/db/models/subscriptionsearch.py +++ b/app/db/models/subscriptionsearch.py @@ -50,6 +50,7 @@ class SubscriptionSearchTask(Base): lease_owner: Mapped[Optional[str]] = mapped_column(String(128)) lease_token: Mapped[Optional[str]] = mapped_column(String(64)) lease_expires_at: Mapped[Optional[str]] = mapped_column(String(40)) + available_at: Mapped[Optional[str]] = mapped_column(String(40)) created_at: Mapped[str] = mapped_column(String(40), nullable=False) updated_at: Mapped[str] = mapped_column(String(40), nullable=False) started_at: Mapped[Optional[str]] = mapped_column(String(40)) @@ -63,6 +64,7 @@ class SubscriptionSearchTask(Base): "ix_subscriptionsearchtask_claim", "state", "priority", + "available_at", "lease_expires_at", "created_at", "id", @@ -70,3 +72,29 @@ class SubscriptionSearchTask(Base): Index("ix_subscriptionsearchtask_batch_position", "batch_id", "position", "id"), Index("ix_subscriptionsearchtask_subscription", "subscription_id", "created_at", "id"), ) + + +class SubscriptionSiteBudget(Base): + """记录兜底搜索对单个站点的唯一租约、间隔与错误冷却。""" + + id = get_id_column() + site_id: Mapped[int] = mapped_column(Integer, nullable=False) + lease_owner: Mapped[Optional[str]] = mapped_column(String(128)) + lease_token: Mapped[Optional[str]] = mapped_column(String(64)) + lease_expires_at: Mapped[Optional[str]] = mapped_column(String(40)) + next_allowed_at: Mapped[str] = mapped_column(String(40), nullable=False) + consecutive_failures: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + success_streak: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + last_outcome: Mapped[Optional[str]] = mapped_column(String(32)) + last_error: Mapped[Optional[str]] = mapped_column(Text) + updated_at: Mapped[str] = mapped_column(String(40), nullable=False) + + __table_args__ = ( + UniqueConstraint("site_id", name="uq_subscriptionsitebudget_site_id"), + Index( + "ix_subscriptionsitebudget_ready", + "next_allowed_at", + "lease_expires_at", + "site_id", + ), + ) diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py index d1a660516..2f3a37d91 100644 --- a/app/db/oper/subscriptionsearch.py +++ b/app/db/oper/subscriptionsearch.py @@ -10,6 +10,7 @@ from sqlalchemy.orm import Session from app.db.base import DbOper, execute_dml from app.db.models.subscriptionsearch import ( + SubscriptionSiteBudget, SubscriptionSearchBatch, SubscriptionSearchTask, ) @@ -29,6 +30,7 @@ class SubscriptionSearchOper(DbOper): subscription_ids: tuple[int, ...], source: str, priority: int, + available_at: Optional[str], ) -> tuple[SubscriptionSearchBatch, int, int]: """创建批次，并以活动键合并同一订阅的重叠搜索入口。""" if not isinstance(self._db, Session): @@ -58,6 +60,7 @@ class SubscriptionSearchOper(DbOper): priority=priority, position=position, state="queued", + available_at=available_at or now, created_at=now, updated_at=now, ) @@ -77,6 +80,16 @@ class SubscriptionSearchOper(DbOper): (SubscriptionSearchTask.priority < priority, priority), else_=SubscriptionSearchTask.priority, ), + available_at=case( + ( + or_( + SubscriptionSearchTask.available_at.is_(None), + SubscriptionSearchTask.available_at > (available_at or now), + ), + available_at or now, + ), + else_=SubscriptionSearchTask.available_at, + ), updated_at=now, ), execution_options={"synchronize_session": False}, @@ -100,6 +113,10 @@ class SubscriptionSearchOper(DbOper): select(SubscriptionSearchTask) .where( SubscriptionSearchTask.cancel_requested == 0, + or_( + SubscriptionSearchTask.available_at.is_(None), + SubscriptionSearchTask.available_at <= now, + ), or_( SubscriptionSearchTask.state == "queued", and_( @@ -113,6 +130,7 @@ class SubscriptionSearchOper(DbOper): ) .order_by( SubscriptionSearchTask.priority.desc(), + SubscriptionSearchTask.available_at.asc(), SubscriptionSearchTask.created_at.asc(), SubscriptionSearchTask.position.asc(), SubscriptionSearchTask.id.asc(), @@ -128,6 +146,10 @@ class SubscriptionSearchOper(DbOper): .where( SubscriptionSearchTask.id == candidate.id, SubscriptionSearchTask.cancel_requested == 0, + or_( + SubscriptionSearchTask.available_at.is_(None), + SubscriptionSearchTask.available_at <= now, + ), or_( SubscriptionSearchTask.state == "queued", and_( @@ -334,6 +356,143 @@ class SubscriptionSearchOper(DbOper): select(SubscriptionSearchBatch).where(SubscriptionSearchBatch.batch_id == batch_id) ).scalars().first() + def claim_site( + self, + *, + site_id: int, + owner: str, + lease_seconds: int, + ) -> tuple[SubscriptionSiteBudget, bool]: + """以 CAS 认领单站点租约，返回当前或已认领预算记录。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅站点预算认领需要调用方提供同步 Session") + now = utc_now_text() + record = self._ensure_site_budget(site_id=site_id, now=now) + lease_busy = bool( + record.lease_token + and record.lease_expires_at + and record.lease_expires_at > now + ) + if lease_busy or record.next_allowed_at > now: + return record, False + lease_token = uuid4().hex + lease_expires_at = ( + datetime.now(timezone.utc) + timedelta(seconds=max(1, lease_seconds)) + ).isoformat(timespec="seconds") + updated = execute_dml( + self._db, + update(SubscriptionSiteBudget) + .where( + SubscriptionSiteBudget.id == record.id, + or_( + SubscriptionSiteBudget.lease_token.is_(None), + SubscriptionSiteBudget.lease_expires_at.is_(None), + SubscriptionSiteBudget.lease_expires_at <= now, + ), + SubscriptionSiteBudget.next_allowed_at <= now, + ) + .values( + lease_owner=owner, + lease_token=lease_token, + lease_expires_at=lease_expires_at, + updated_at=now, + ), + execution_options={"synchronize_session": False}, + ) + if not updated: + self._db.expire_all() + current = self._db.execute( + select(SubscriptionSiteBudget).where( + SubscriptionSiteBudget.site_id == site_id + ) + ).scalar_one() + return current, False + self._db.flush() + self._db.expire_all() + claimed = self._db.execute( + select(SubscriptionSiteBudget).where( + SubscriptionSiteBudget.site_id == site_id, + SubscriptionSiteBudget.lease_token == lease_token, + ) + ).scalar_one() + return claimed, True + + def finish_site( + self, + *, + site_id: int, + lease_token: str, + outcome: str, + next_allowed_at: str, + error: Optional[str], + ) -> bool: + """释放当前站点租约，并按结果推进失败或恢复计数。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅站点预算收口需要调用方提供同步 Session") + record = self._db.execute( + select(SubscriptionSiteBudget).where( + SubscriptionSiteBudget.site_id == site_id, + SubscriptionSiteBudget.lease_token == lease_token, + ) + ).scalars().first() + if record is None: + return False + if outcome == "success": + failures = max(0, record.consecutive_failures - 1) + success_streak = record.success_streak + 1 + elif outcome == "skipped": + failures = record.consecutive_failures + success_streak = record.success_streak + else: + failures = record.consecutive_failures + 1 + success_streak = 0 + now = utc_now_text() + return bool(execute_dml( + self._db, + update(SubscriptionSiteBudget) + .where( + SubscriptionSiteBudget.id == record.id, + SubscriptionSiteBudget.lease_token == lease_token, + ) + .values( + lease_owner=None, + lease_token=None, + lease_expires_at=None, + next_allowed_at=next_allowed_at, + consecutive_failures=failures, + success_streak=success_streak, + last_outcome=outcome, + last_error=error, + updated_at=now, + ), + execution_options={"synchronize_session": False}, + )) + + def _ensure_site_budget(self, *, site_id: int, now: str) -> SubscriptionSiteBudget: + """并发安全地创建站点预算初始记录。""" + record = self._db.execute( + select(SubscriptionSiteBudget).where( + SubscriptionSiteBudget.site_id == site_id + ) + ).scalars().first() + if record is not None: + return record + try: + with self._db.begin_nested(): + self._db.add(SubscriptionSiteBudget( + site_id=site_id, + next_allowed_at=now, + updated_at=now, + )) + self._db.flush() + except IntegrityError: + self._db.expire_all() + return self._db.execute( + select(SubscriptionSiteBudget).where( + SubscriptionSiteBudget.site_id == site_id + ) + ).scalar_one() + def _batch_cancel_requested(self, batch_id: str) -> bool: """在当前事务中读取批次取消标记。""" return bool(self._db.execute( diff --git a/app/modules/indexer/__init__.py b/app/modules/indexer/__init__.py index ee96741cb..9a0b29660 100644 --- a/app/modules/indexer/__init__.py +++ b/app/modules/indexer/__init__.py @@ -5,6 +5,7 @@ from typing import Any, Callable, List, Mapping, Optional, Tuple, Union, cast from app.application.site.health import get_configured_site_health_service from app.application.site.query import get_configured_site_query_service +from app.application.site.search_observation import report_site_search_outcome from app.application.site.sites import SitesHelper # pylint: disable=import-error,no-name-in-module from app.domain import site as site_rules from app.domain.context import Context, SubtitleInfo, TorrentInfo @@ -64,6 +65,24 @@ class _IndexerSearchOutcome: error_flag: bool result: List[dict[str, Any]] seconds: int + error: Optional[Exception] = None + + +def _classify_search_failure(error: Optional[Exception]) -> str: + """把索引器异常归一为站点预算可执行的冷却类别。""" + if error is None: + return "error" + status_code = getattr(error, "status_code", None) + text = f"{type(error).__name__}: {error}".lower() + if status_code == 429 or "429" in text or "rate limit" in text: + return "rate_limited" + if status_code == 403 or "403" in text or "forbidden" in text: + return "forbidden" + if any(token in text for token in ("未登录", "登录失效", "login", "cookie")): + return "login_invalid" + if isinstance(error, TimeoutError) or "timeout" in text or "timed out" in text or "超时" in text: + return "timeout" + return "error" class IndexerModule(_ModuleBase): @@ -359,12 +378,14 @@ class IndexerModule(_ModuleBase): start_time: datetime, error_flag: bool, result: List[dict[str, Any]], + error: Optional[Exception] = None, ) -> _IndexerSearchOutcome: """把同步、异步 I/O 结果整理为共用的搜索完成状态""" return _IndexerSearchOutcome( error_flag=error_flag, result=result, seconds=(datetime.now() - start_time).seconds, + error=error, ) @staticmethod @@ -403,15 +424,18 @@ class IndexerModule(_ModuleBase): page=page, ) if not request: + report_site_search_outcome(attempted=False, outcome="skipped") return [] # 开始搜索 + error: Optional[Exception] = None try: error_flag, result = self.__execute_search(site, request) except Exception as err: + error = err self.__log_search_error(site, "torrents", err) - outcome = self.__create_search_outcome(start_time, error_flag, result) + outcome = self.__create_search_outcome(start_time, error_flag, result, error) # 统计索引情况 self.__indexer_statistic( @@ -419,6 +443,15 @@ class IndexerModule(_ModuleBase): error_flag=outcome.error_flag, seconds=outcome.seconds, ) + report_site_search_outcome( + attempted=True, + outcome=( + _classify_search_failure(outcome.error) + if outcome.error_flag or outcome.error is not None + else "success" + ), + error=str(outcome.error) if outcome.error is not None else None, + ) # 返回结果 return self.__parse_result( @@ -498,15 +531,18 @@ class IndexerModule(_ModuleBase): page=page, ) if not request: + report_site_search_outcome(attempted=False, outcome="skipped") return [] # 开始搜索 + error: Optional[Exception] = None try: error_flag, result = await self.__async_execute_search(site, request) except Exception as err: + error = err self.__log_search_error(site, "torrents", err) - outcome = self.__create_search_outcome(start_time, error_flag, result) + outcome = self.__create_search_outcome(start_time, error_flag, result, error) # 统计索引情况 await self.__async_indexer_statistic( @@ -514,6 +550,15 @@ class IndexerModule(_ModuleBase): error_flag=outcome.error_flag, seconds=outcome.seconds, ) + report_site_search_outcome( + attempted=True, + outcome=( + _classify_search_failure(outcome.error) + if outcome.error_flag or outcome.error is not None + else "success" + ), + error=str(outcome.error) if outcome.error is not None else None, + ) # 返回结果 return self.__parse_result( diff --git a/database/versions/d2a7c5e9f1b4_3_0_20.py b/database/versions/d2a7c5e9f1b4_3_0_20.py new file mode 100644 index 000000000..48b32361d --- /dev/null +++ b/database/versions/d2a7c5e9f1b4_3_0_20.py @@ -0,0 +1,95 @@ +"""3.0.20 增加订阅搜索站点预算与批次启动抖动。 + +Revision ID: d2a7c5e9f1b4 +Revises: c1f4a8d2e6b9 +Create Date: 2026-09-01 +""" + +# Alembic 的 op 是运行期代理，静态分析无法看到实际操作方法。 +# pylint: disable=no-member + +import sqlalchemy as sa +from alembic import op + +revision = "d2a7c5e9f1b4" +down_revision = "c1f4a8d2e6b9" +branch_labels = None +depends_on = None + +_TASK_TABLE = "subscriptionsearchtask" +_BUDGET_TABLE = "subscriptionsitebudget" + + +def _table_names() -> set[str]: + """返回当前数据库表名集合。""" + return set(sa.inspect(op.get_bind()).get_table_names()) + + +def _column_names(table_name: str) -> set[str]: + """返回指定表的当前字段名集合。""" + return { + column["name"] + for column in sa.inspect(op.get_bind()).get_columns(table_name) + } + + +def upgrade() -> None: + """创建单站点预算，并让搜索任务支持持久启动抖动。""" + tables = _table_names() + if _TASK_TABLE in tables and "available_at" not in _column_names(_TASK_TABLE): + with op.batch_alter_table(_TASK_TABLE) as batch_op: + batch_op.add_column(sa.Column("available_at", sa.String(length=40), nullable=True)) + batch_op.drop_index("ix_subscriptionsearchtask_claim") + batch_op.create_index( + "ix_subscriptionsearchtask_claim", + [ + "state", + "priority", + "available_at", + "lease_expires_at", + "created_at", + "id", + ], + ) + op.execute( + sa.text( + "UPDATE subscriptionsearchtask " + "SET available_at = created_at WHERE available_at IS NULL" + ) + ) + if _BUDGET_TABLE not in tables: + op.create_table( + _BUDGET_TABLE, + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("site_id", sa.Integer(), nullable=False), + sa.Column("lease_owner", sa.String(length=128), nullable=True), + sa.Column("lease_token", sa.String(length=64), nullable=True), + sa.Column("lease_expires_at", sa.String(length=40), nullable=True), + sa.Column("next_allowed_at", sa.String(length=40), nullable=False), + sa.Column("consecutive_failures", sa.Integer(), nullable=False), + sa.Column("success_streak", sa.Integer(), nullable=False), + sa.Column("last_outcome", sa.String(length=32), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("updated_at", sa.String(length=40), nullable=False), + sa.UniqueConstraint("site_id", name="uq_subscriptionsitebudget_site_id"), + ) + op.create_index( + "ix_subscriptionsitebudget_ready", + _BUDGET_TABLE, + ["next_allowed_at", "lease_expires_at", "site_id"], + ) + + +def downgrade() -> None: + """移除站点预算与任务启动抖动。""" + tables = _table_names() + if _BUDGET_TABLE in tables: + op.drop_table(_BUDGET_TABLE) + if _TASK_TABLE in tables and "available_at" in _column_names(_TASK_TABLE): + with op.batch_alter_table(_TASK_TABLE) as batch_op: + batch_op.drop_index("ix_subscriptionsearchtask_claim") + batch_op.drop_column("available_at") + batch_op.create_index( + "ix_subscriptionsearchtask_claim", + ["state", "priority", "lease_expires_at", "created_at", "id"], + ) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index f350f3a73..a4683ffd5 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -754,8 +754,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 925 | -| 内部导入边 | 7,728 | +| Python 模块 | 927 | +| 内部导入边 | 7,745 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 2f4470d99..c72a46493 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 925 / 7,728 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 927 / 7,745 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index a28f3fd92..71f475d71 100644 --- a/docs/refactor/moviepilot-subscription-governance-roadmap.md +++ b/docs/refactor/moviepilot-subscription-governance-roadmap.md @@ -1,7 +1,7 @@ # MoviePilot 订阅执行治理 > 状态：`active（2026-09-01 已由 MoviePilot v3 接管）` -> 当前叶：`SUB-GOV-002B` +> 当前叶：`SUB-GOV-002C` > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、 > `V3-RDY-009A1C`、`V3-RDY-009A1D` > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环 @@ -233,13 +233,13 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 | `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `completed（2026-09-01）` | | `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `not_activated（2026-09-01）` | | `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `completed（2026-09-01）` | -| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `in_progress` | -| `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `pending` | +| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `completed（2026-09-01）` | +| `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `in_progress` | | `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `pending` | | `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` | | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` | -当前只激活 `SUB-GOV-002B`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B +当前只激活 `SUB-GOV-002C`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。 ### 6.1 SUB-GOV-001A 验收证据 @@ -314,6 +314,22 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 - 验证：队列/编排/调度专项 `43 passed`，迁移与声明式模型 `35 passed`，复杂度/并发/架构/迁移组合 `119 passed`，错误级 Pylint 为 0，Alembic 唯一 head 为 `c1f4a8d2e6b9`。 +### 6.7 SUB-GOV-002B 验收证据 + +- 新增持久 `SubscriptionSiteBudget`，以 `site_id` 唯一租约保证每站点最多一个在途兜底搜索；不同站点使用 + 独立记录，可由 Search provider 的现有 worker 并行推进； +- 每次真实站点页请求后写入 `next_allowed_at`：成功保持 `60–300s` 随机间隔，历史失败按一级一级递减实现 + 渐进恢复；`429/403`、登录失效、超时和通用错误进入有上限的指数冷却； +- 自动 fallback 批次在任务 `available_at` 中持久化 `0–60s` 全局启动抖动；手工任务优先级更高且立即入队， + 但进入 Search provider 后仍使用完全相同的站点租约、间隔和冷却，不能旁路； +- 站点预算只包装订阅队列的同步搜索实例，不改变 RSS/Spider 刷新入口和普通手工资源搜索；索引器列表返回 + 合同、同步/异步结果及既有站点健康统计语义保持不变； +- 等待每秒检查批次取消，超过 5 秒短窗口的冷却或在途站点不阻塞整批：该站点本轮返回空页，其他站点继续， + 订阅在处理成功站点结果后以聚合失败收口，避免慢站点拖死后续订阅； +- 验证：站点预算/队列/编排/provider 专项 `38 passed`，搜索和索引器兼容 `91 passed`，迁移与声明式模型 + `37 passed, 1 skipped`，架构合同组合 `149 passed`，错误级 Pylint 为 0，Alembic 唯一 head 为 + `d2a7c5e9f1b4`，Host 架构基线与 `git diff --check` 通过。 + ## 7. 上线前验证与验收 ### 7.1 场景 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index b1b952d46..e21957238 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 7728, - "edge_sha256": "33cdb555e19d7a5d15dbed065c98908269c3528ec4d61c2ac5515420c0b4cfa4", + "edge_count": 7745, + "edge_sha256": "4c722884c3a228e2fc51d5389d4fec3e10436e797c4ae168b9306154a9179475", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3442,6 +3442,9 @@ "app.application.subscription.delete -> app.schemas", "app.application.subscription.delete -> app.schemas.common", "app.application.subscription.delete -> app.schemas.event", + "app.application.subscription.execution -> app.application", + "app.application.subscription.execution -> app.application.subscription", + "app.application.subscription.execution -> app.application.subscription.sitebudget", "app.application.subscription.facts -> app.application", "app.application.subscription.facts -> app.application.subscription", "app.application.subscription.facts -> app.application.subscription.contract", @@ -3492,6 +3495,11 @@ "app.application.subscription.search -> app.application", "app.application.subscription.search -> app.application.subscription", "app.application.subscription.search -> app.application.subscription.contract", + "app.application.subscription.sitebudget -> app.application", + "app.application.subscription.sitebudget -> app.application.site", + "app.application.subscription.sitebudget -> app.application.site.search_observation", + "app.application.subscription.sitebudget -> app.runtime", + "app.application.subscription.sitebudget -> app.runtime.stop", "app.application.subscription.write -> app.application", "app.application.subscription.write -> app.application.outbox", "app.application.subscription.write -> app.application.subscription", @@ -4172,6 +4180,9 @@ "app.chain.search.cache -> app.schemas.types", "app.chain.search.contract -> app.chain", "app.chain.search.contract -> app.chain.base", + "app.chain.search.facade -> app.application", + "app.chain.search.facade -> app.application.subscription", + "app.chain.search.facade -> app.application.subscription.sitebudget", "app.chain.search.facade -> app.chain", "app.chain.search.facade -> app.chain.base", "app.chain.search.facade -> app.chain.search", @@ -4246,6 +4257,9 @@ "app.chain.search.provider -> app.application", "app.chain.search.provider -> app.application.configuration", "app.chain.search.provider -> app.application.site", + "app.chain.search.provider -> app.application.site.search_observation", + "app.chain.search.provider -> app.application.subscription", + "app.chain.search.provider -> app.application.subscription.sitebudget", "app.chain.search.provider -> app.chain", "app.chain.search.provider -> app.chain.search", "app.chain.search.provider -> app.chain.search.contract", @@ -4588,6 +4602,7 @@ "app.chain.subscribe.search -> app.application.subscription.contract", "app.chain.subscribe.search -> app.application.subscription.execution", "app.chain.subscribe.search -> app.application.subscription.query", + "app.chain.subscribe.search -> app.application.subscription.sitebudget", "app.chain.subscribe.search -> app.chain", "app.chain.subscribe.search -> app.chain.media", "app.chain.subscribe.search -> app.chain.search", @@ -5134,6 +5149,7 @@ "app.db.adapters.subscriptionsearch -> app.application", "app.db.adapters.subscriptionsearch -> app.application.subscription", "app.db.adapters.subscriptionsearch -> app.application.subscription.execution", + "app.db.adapters.subscriptionsearch -> app.application.subscription.sitebudget", "app.db.adapters.subscriptionsearch -> app.db", "app.db.adapters.subscriptionsearch -> app.db.models", "app.db.adapters.subscriptionsearch -> app.db.models.subscriptionsearch", @@ -6133,6 +6149,7 @@ "app.modules.indexer -> app.application.site", "app.modules.indexer -> app.application.site.health", "app.modules.indexer -> app.application.site.query", + "app.modules.indexer -> app.application.site.search_observation", "app.modules.indexer -> app.domain", "app.modules.indexer -> app.domain.context", "app.modules.indexer -> app.domain.site", @@ -8821,7 +8838,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 925, + "module_count": 927, "modules": [ "app", "app.adapters", @@ -9106,6 +9123,7 @@ "app.application.site.health", "app.application.site.mutation", "app.application.site.query", + "app.application.site.search_observation", "app.application.storage", "app.application.subscription", "app.application.subscription.candidates", @@ -9119,6 +9137,7 @@ "app.application.subscription.priority", "app.application.subscription.query", "app.application.subscription.search", + "app.application.subscription.sitebudget", "app.application.subscription.write", "app.application.system", "app.application.torrent", diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index 6b052e7aa..1826f90a3 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -73,6 +73,7 @@ def test_fallback_queue_executes_without_match_global_lock(tmp_path, monkeypatch """R/P 兜底搜索在持久队列中执行，不受日常 Match 长锁阻塞。""" subscribes = [_subscribe(1), _subscribe(2)] chain = _chain(tmp_path, subscribes) + monkeypatch.setattr(chain, "_search_batch_available_at", lambda _source: "1970-01-01T00:00:00+00:00") processed = [] monkeypatch.setattr( chain, @@ -94,6 +95,7 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke """单订阅异常不得中止批次后续任务，聚合终态必须暴露失败。""" subscribes = [_subscribe(3), _subscribe(4)] chain = _chain(tmp_path, subscribes) + monkeypatch.setattr(chain, "_search_batch_available_at", lambda _source: "1970-01-01T00:00:00+00:00") processed = [] def process(subscribe, _searchchain): diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py new file mode 100644 index 000000000..14510a048 --- /dev/null +++ b/tests/test_subscription_site_budget.py @@ -0,0 +1,266 @@ +"""订阅兜底搜索的站点并发、冷却、恢复与取消测试。""" + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session, sessionmaker + +from app.application.site.search_observation import ( + SiteSearchObservation, + report_site_search_outcome, +) +from app.application.subscription.sitebudget import ( + SiteBudgetClaim, + SubscriptionSearchCancelled, + SubscriptionSiteBudget, +) +from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository +from app.db.base import Base +from app.db.models.subscriptionsearch import SubscriptionSiteBudget as SiteBudgetRecord +from app.chain.search.facade import SearchChain +from app.modules.indexer import _classify_search_failure +from app.runtime.stop import ProcessStopState + + +def _repository(tmp_path): + """构造使用独立 SQLite 文件的站点预算仓储。""" + engine = create_engine(f"sqlite:///{tmp_path / 'site-budget.db'}") + Base.metadata.create_all(engine) + return TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)), engine + + +def test_site_budget_allows_one_inflight_per_site_and_independent_sites(tmp_path): + """同站点第二个调用必须等待，不同站点可立即并行。""" + repository, _engine = _repository(tmp_path) + + first = repository.claim_site(site_id=1, owner="worker-a", lease_seconds=900) + same_site = repository.claim_site(site_id=1, owner="worker-b", lease_seconds=900) + other_site = repository.claim_site(site_id=2, owner="worker-b", lease_seconds=900) + + assert first.acquired is True + assert same_site.acquired is False + assert same_site.lease_token is None + assert other_site.acquired is True + + +def test_site_budget_applies_error_cooldown_and_gradual_success_recovery(tmp_path): + """失败增加冷却计数，后续成功每次只恢复一级而非直接清零。""" + repository, engine = _repository(tmp_path) + claim = repository.claim_site(site_id=3, owner="worker-a", lease_seconds=900) + now = datetime.now(timezone.utc) + + assert repository.finish_site( + site_id=3, + lease_token=claim.lease_token, + outcome="timeout", + next_allowed_at=(now + timedelta(minutes=5)).isoformat(timespec="seconds"), + error="request timeout", + ) is True + with Session(engine) as session: + failed = session.execute( + select(SiteBudgetRecord).where(SiteBudgetRecord.site_id == 3) + ).scalar_one() + assert failed.consecutive_failures == 1 + assert failed.success_streak == 0 + failed.next_allowed_at = (now - timedelta(seconds=1)).isoformat(timespec="seconds") + session.commit() + + recovered = repository.claim_site(site_id=3, owner="worker-b", lease_seconds=900) + assert recovered.acquired is True + assert recovered.consecutive_failures == 1 + assert repository.finish_site( + site_id=3, + lease_token=recovered.lease_token, + outcome="success", + next_allowed_at=(now + timedelta(minutes=1)).isoformat(timespec="seconds"), + ) is True + with Session(engine) as session: + healthy = session.execute( + select(SiteBudgetRecord).where(SiteBudgetRecord.site_id == 3) + ).scalar_one() + assert healthy.consecutive_failures == 0 + assert healthy.success_streak == 1 + assert healthy.last_outcome == "success" + + +@pytest.mark.parametrize( + ("error", "outcome"), + [ + (RuntimeError("HTTP 429"), "rate_limited"), + (RuntimeError("403 forbidden"), "forbidden"), + (RuntimeError("Cookie 登录失效"), "login_invalid"), + (TimeoutError("request timed out"), "timeout"), + ], +) +def test_indexer_failures_map_to_site_cooldown_categories(error, outcome): + """外站典型失败必须进入对应的站点级冷却类别。""" + assert _classify_search_failure(error) == outcome + + +class _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=False, + retry_at=(datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat(timespec="seconds"), + consecutive_failures=0, + ) + + def finish_site(self, **_kwargs) -> bool: + """等待测试不会取得租约，因此不应调用收口。""" + raise AssertionError("未取得站点预算时不应收口") + + +def test_site_budget_wait_is_cancellable_without_business_lock(): + """批次取消应在一次短等待后终止预算获取。""" + cancelled = False + + def sleeper(_seconds: float) -> None: + """模拟等待一次后收到取消请求。""" + nonlocal cancelled + cancelled = True + + budget = SubscriptionSiteBudget( + repository=_WaitingRepository(), + owner="worker-a", + cancelled=lambda: cancelled, + stop_state=ProcessStopState(), + sleeper=sleeper, + max_wait_seconds=600, + ) + + with pytest.raises(SubscriptionSearchCancelled): + budget.acquire(9) + + +def test_site_budget_delay_keeps_manual_requests_under_same_budget(): + """来源优先级不参与站点冷却计算，手工任务不能获得旁路。""" + budget = SubscriptionSiteBudget( + repository=_WaitingRepository(), + owner="manual-worker", + cancelled=lambda: False, + stop_state=ProcessStopState(), + random_uniform=lambda _low, _high: 60.0, + ) + + assert budget._next_delay("success", consecutive_failures=0) == 60.0 # pylint: disable=protected-access + assert budget._next_delay("success", consecutive_failures=2) == 120.0 # pylint: disable=protected-access + assert budget._next_delay("rate_limited", consecutive_failures=0) == 900.0 # pylint: disable=protected-access + + +def test_skipped_search_releases_budget_without_external_interval(): + """本地限流等未发请求结果不应伪造成功或追加外站间隔。""" + 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=0, + lease_token="lease-token", + ) + + assert budget.finish(claim, SiteSearchObservation()) is True + assert captured["outcome"] == "skipped" + assert captured["next_allowed_at"] == now.isoformat(timespec="seconds") + + +def test_search_provider_reports_cooled_site_without_blocking_other_results(): + """超出短等待窗口的站点返回空页并记录聚合失败，而非阻塞整个 provider。""" + repository = _WaitingRepository() + budget = SubscriptionSiteBudget( + repository=repository, + owner="fallback-worker", + cancelled=lambda: False, + stop_state=ProcessStopState(), + max_wait_seconds=0, + ) + chain = object.__new__(SearchChain) + chain.configure_subscription_site_budget(budget) + chain.search_site_torrents = lambda **_kwargs: ["unexpected"] + + result = chain._search_site_torrents_with_budget( # pylint: disable=protected-access + site={"id": 11, "name": "Cooling"}, + keyword="movie", + mtype=None, + page=0, + ) + + assert result == [] + failures = chain.consume_subscription_site_budget_failures() + assert len(failures) == 1 + assert "站点 11" in failures[0] + + +def test_search_provider_releases_successful_site_budget(): + """真实站点页完成后必须释放租约并写入成功间隔。""" + 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(), + random_uniform=lambda _low, _high: 60.0, + ) + chain = object.__new__(SearchChain) + chain.configure_subscription_site_budget(budget) + + def search_site_torrents(**_kwargs): + """模拟索引器成功并发布观察结果。""" + report_site_search_outcome(attempted=True, outcome="success") + return ["torrent"] + + chain.search_site_torrents = search_site_torrents + + result = chain._search_site_torrents_with_budget( # pylint: disable=protected-access + site={"id": 12, "name": "Healthy"}, + keyword="movie", + mtype=None, + page=0, + ) + + assert result == ["torrent"] + assert captured["site_id"] == 12 + assert captured["outcome"] == "success" + assert captured["lease_token"] == "lease-token"