From b4be185e445337900976d4dcd1a1b26a8818c595 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 5 Sep 2026 20:03:09 +0800 Subject: [PATCH] fix(subscribe): tolerate partial site search failures --- app/application/subscription/execution.py | 8 +- app/chain/search/contract.py | 1 + app/chain/search/facade.py | 23 ++++- app/chain/search/provider.py | 13 +-- app/chain/subscribe/search.py | 10 ++- app/chain/subscribe/searchtask.py | 6 +- app/modules/indexer/__init__.py | 15 +++- ...iepilot-subscription-governance-roadmap.md | 8 +- tests/test_subscription_search_governance.py | 72 +++++++++++++++- tests/test_subscription_site_budget.py | 85 +++++++++++++++++++ 10 files changed, 220 insertions(+), 21 deletions(-) diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py index e35d43284..780718200 100644 --- a/app/application/subscription/execution.py +++ b/app/application/subscription/execution.py @@ -66,6 +66,10 @@ class SubscriptionExecutionAdmission: return self._clock() >= lease.expires_at +class SubscriptionSiteSearchFailed(RuntimeError): + """表示本轮没有搜索源成功完成，无需输出内部异常堆栈。""" + + @dataclass(slots=True) class SubscriptionExecutionContext: """一次订阅执行的显式取消、阶段和副作用边界。""" @@ -101,9 +105,9 @@ class SubscriptionExecutionContext: def raise_subscription_site_budget_failures(failures: tuple[str, ...]) -> None: - """在成功站点结果完成处理后暴露其余站点的聚合失败。""" + """在所有实际搜索源均未成功时暴露站点聚合失败。""" if failures: - raise RuntimeError("；".join(failures)) + raise SubscriptionSiteSearchFailed("；".join(failures)) def raise_subscription_site_budget_deferral( diff --git a/app/chain/search/contract.py b/app/chain/search/contract.py index ef3bd7600..cf56ac060 100644 --- a/app/chain/search/contract.py +++ b/app/chain/search/contract.py @@ -97,6 +97,7 @@ if TYPE_CHECKING: matches_music_resource: Callable[..., Any] music_site_keywords: Callable[..., Any] process: Callable[..., Any] + record_subscription_site_budget_success: Callable[..., Any] record_subscription_site_budget_failure: Callable[..., Any] consume_subscription_site_budget_failures: Callable[..., Any] record_subscription_site_budget_deferred: Callable[..., Any] diff --git a/app/chain/search/facade.py b/app/chain/search/facade.py index 4c718c364..21acb55cd 100644 --- a/app/chain/search/facade.py +++ b/app/chain/search/facade.py @@ -52,26 +52,41 @@ class SearchChain(ChainBase): """仅为订阅搜索启用或清除站点预算，不影响其它搜索入口。""" self._subscription_site_budget = budget self._subscription_site_budget_failures: list[str] = [] + self._subscription_site_budget_successes: set[int] = set() self._subscription_site_budget_deferrals: list[SubscriptionSiteBudgetDeferral] = [] self._subscription_site_budget_failure_lock = threading.Lock() + def record_subscription_site_budget_success(self, site_id: int) -> None: + """线程安全地记录一个已正常完成真实请求的站点。""" + lock = getattr(self, "_subscription_site_budget_failure_lock", None) + if lock is None: + return + with lock: + self._subscription_site_budget_successes.add(site_id) + 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, ...]: - """读取并清空当前订阅搜索积累的站点预算失败。""" + def consume_subscription_site_budget_failures( + self, + *, + has_results: bool = False, + ) -> 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) + has_successful_site = bool(self._subscription_site_budget_successes) self._subscription_site_budget_failures.clear() - return failures + self._subscription_site_budget_successes.clear() + return () if has_results or has_successful_site else failures def record_subscription_site_budget_deferred( self, diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index 18106aad2..67d75fbf2 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -243,11 +243,14 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): raise else: budget.record_request(site_id, len(result or [])) - if observation.attempted and observation.outcome not in {"success", "skipped"}: - failure = observation.error or observation.outcome - self.record_subscription_site_budget_failure( - f"站点 {site.get('name') or site_id} 搜索失败：{failure}" - ) + if observation.attempted: + if observation.outcome == "success": + self.record_subscription_site_budget_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) return result finally: try: diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index 0946c1e5c..b9ebde639 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -19,6 +19,7 @@ from app.application.subscription.execution import ( SearchTaskSnapshot, SubscriptionExecutionContext, SubscriptionSearchRepository, + SubscriptionSiteSearchFailed, raise_subscription_site_budget_deferral, raise_subscription_site_budget_failures, ) @@ -300,7 +301,10 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): except Exception as err: outcome = "failed" reason = "error" - logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True) + logger.error( + f"订阅 {subscribe.name} 搜索失败：{str(err)}", + exc_info=not isinstance(err, SubscriptionSiteSearchFailed), + ) finally: try: if current and current.state == "N": @@ -730,7 +734,9 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): 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() + site_budget_failures = searchchain.consume_subscription_site_budget_failures( + has_results=bool(contexts), + ) site_budget_deferrals = searchchain.consume_subscription_site_budget_deferrals() _ensure_execution_active(execution_context) if not contexts: diff --git a/app/chain/subscribe/searchtask.py b/app/chain/subscribe/searchtask.py index ca783efee..3a440208a 100644 --- a/app/chain/subscribe/searchtask.py +++ b/app/chain/subscribe/searchtask.py @@ -15,6 +15,7 @@ from app.application.subscription.execution import ( SubscriptionExecutionContext, SubscriptionExecutionLease, SubscriptionSearchRepository, + SubscriptionSiteSearchFailed, handle_subscription_search_deferred, ) from app.application.subscription.observability import ( @@ -239,7 +240,10 @@ class SubscriptionSearchTaskRunner: self.summary.record, ) except Exception as err: - logger.error(f"订阅《{subscribe.name}》搜索失败：{str(err)}", exc_info=True) + logger.error( + f"订阅《{subscribe.name}》搜索失败：{str(err)}", + exc_info=not isinstance(err, SubscriptionSiteSearchFailed), + ) self.queue.finish_task( task_id=task_id, lease_token=lease_token, diff --git a/app/modules/indexer/__init__.py b/app/modules/indexer/__init__.py index f7c320c81..f64d1953b 100644 --- a/app/modules/indexer/__init__.py +++ b/app/modules/indexer/__init__.py @@ -49,6 +49,8 @@ _SPECIALIZED_SEARCH_ARGUMENTS = { "RousiPro": ("keyword", "mtype", "cat", "page"), } +_UNKNOWN_SEARCH_FAILURE_MESSAGE = "站点请求或页面解析失败" + @dataclass(frozen=True) class _IndexerSearchRequest: @@ -85,6 +87,15 @@ def _classify_search_failure(error: Optional[Exception]) -> str: return "error" +def _search_failure_message(outcome: _IndexerSearchOutcome) -> Optional[str]: + """返回供站点预算持久化和任务聚合展示的失败原因。""" + if outcome.error is not None: + return str(outcome.error) + if outcome.error_flag: + return _UNKNOWN_SEARCH_FAILURE_MESSAGE + return None + + class IndexerModule(_ModuleBase): """ 索引模块 @@ -450,7 +461,7 @@ class IndexerModule(_ModuleBase): if outcome.error_flag or outcome.error is not None else "success" ), - error=str(outcome.error) if outcome.error is not None else None, + error=_search_failure_message(outcome), ) # 返回结果 @@ -557,7 +568,7 @@ class IndexerModule(_ModuleBase): if outcome.error_flag or outcome.error is not None else "success" ), - error=str(outcome.error) if outcome.error is not None else None, + error=_search_failure_message(outcome), ) # 返回结果 diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index ef10ab0cb..08886aa1f 100644 --- a/docs/refactor/moviepilot-subscription-governance-roadmap.md +++ b/docs/refactor/moviepilot-subscription-governance-roadmap.md @@ -329,8 +329,8 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 但进入 Search provider 后仍使用完全相同的站点租约、间隔和冷却，不能旁路； - 站点预算只包装订阅队列的同步搜索实例，不改变 RSS/Spider 刷新入口和普通手工资源搜索；索引器列表返回 合同、同步/异步结果及既有站点健康统计语义保持不变； -- 等待每秒检查批次取消，超过 5 秒短窗口的冷却或在途站点不阻塞整批：该站点本轮返回空页，其他站点继续， - 订阅在处理成功站点结果后以聚合失败收口，避免慢站点拖死后续订阅； +- 等待每秒检查批次取消，超过 5 秒短窗口的冷却或在途站点不阻塞整批：该站点本轮返回空页，其他站点继续； + 部分站点失败只记录告警与冷却，仅在没有站点或其他搜索源成功完成时以聚合失败收口； - 验证：站点预算/队列/编排/provider 专项 `38 passed`，搜索和索引器兼容 `91 passed`，迁移与声明式模型 `37 passed, 1 skipped`，架构合同组合 `149 passed`，错误级 Pylint 为 0，Alembic 唯一 head 为 `d2a7c5e9f1b4`，Host 架构基线与 `git diff --check` 通过。 @@ -339,8 +339,8 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 - 队列认领增加 15 分钟持久老化窗口：新手工任务继续优先，但等待超过窗口的 fallback 任务先于后续高优先级 新任务执行，防止持续手工流量造成后台饥饿； -- 正式搜索队列继续完全不访问 Match 类级锁；联合测试在同一 provider fan-out 中证明冷却站点返回可聚合失败， - 健康站点仍独立完成并释放预算，慢站点不拖住整轮； +- 正式搜索队列继续完全不访问 Match 类级锁；联合测试在同一 provider fan-out 中证明冷却或失败站点独立记录， + 健康站点仍独立完成并释放预算，部分站点失败不改变订阅任务终态； - 同站点第二个 worker 无法认领，另一站点可同时认领；任务租约与站点租约均验证过期后以新 token 恢复， 单订阅失败继续后续任务且批次保持聚合 failed； - 手工入口仅改变队列优先级和启动时机，不改变站点预算、失败冷却或 single-flight；RSS/Spider 仍未接入该预算， diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index 55124df18..525fb7737 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -12,7 +12,10 @@ from sqlalchemy.orm import sessionmaker from app.application.site.observation import report_site_search_outcome from app.application.subscription.contract import SubscriptionSnapshot -from app.application.subscription.execution import SubscriptionExecutionAdmission +from app.application.subscription.execution import ( + SubscriptionExecutionAdmission, + raise_subscription_site_budget_failures, +) from app.application.subscription.sitebudget import ( SubscriptionSearchCancelled, SubscriptionSearchDeferred, @@ -252,6 +255,73 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke assert chain._subscription_execution_admission.release(lease) is True +def test_site_search_failure_logs_without_internal_traceback(tmp_path, monkeypatch): + """可预期的站点聚合失败仍标记任务失败，但不输出误导性的内部堆栈。""" + chain = _chain(tmp_path, [_subscribe(15)]) + _make_tasks_ready(monkeypatch) + + def process(*_args, **_kwargs): + """通过正式聚合助手制造站点失败。""" + raise_subscription_site_budget_failures( + ("站点 Generic 搜索失败：站点请求或页面解析失败",) + ) + + monkeypatch.setattr( + chain, + "_process_search_subscription", + process, + ) + error_logs = [] + monkeypatch.setattr( + "app.chain.subscribe.searchtask.logger.error", + lambda message, **kwargs: error_logs.append((message, kwargs)), + ) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(state="R") + + batch = chain.get_search_batch(batch_id) + assert batch.state == "failed" + assert batch.failed_count == 1 + assert error_logs == [ + ( + "订阅《治理电影 15》搜索失败：站点 Generic 搜索失败：站点请求或页面解析失败", + {"exc_info": False}, + ) + ] + + +def test_inline_site_search_failure_logs_without_internal_traceback(tmp_path, monkeypatch): + """兼容内联搜索遇到站点聚合失败时也不得输出内部堆栈。""" + chain = _chain(tmp_path, [_subscribe(16)]) + del chain.subscription_search_repository + monkeypatch.setattr(chain, "_wait_before_scheduled_search", lambda *_args: None) + + def process(*_args, **_kwargs): + """通过正式聚合助手制造内联站点失败。""" + raise_subscription_site_budget_failures( + ("站点 Generic 搜索失败：站点请求或页面解析失败",) + ) + + monkeypatch.setattr(chain, "_process_search_subscription", process) + error_logs = [] + monkeypatch.setattr( + subscribe_search.logger, + "error", + lambda message, **kwargs: error_logs.append((message, kwargs)), + ) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + chain.search(state="R") + + assert error_logs == [ + ( + "订阅 治理电影 16 搜索失败：站点 Generic 搜索失败：站点请求或页面解析失败", + {"exc_info": False}, + ) + ] + + def test_site_budget_conflict_requeues_task_without_batch_failure(tmp_path, monkeypatch): """站点预算冲突应自动排队重试，不能把订阅批次置为失败。""" subscribe = _subscribe(5) diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index 49d79e731..4a09db9a1 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -337,6 +337,17 @@ def test_search_provider_releases_successful_site_budget(): assert snapshot.candidate_count == 1 assert snapshot.failure_count == 0 assert snapshot.release_failure_count == 0 + chain.record_subscription_site_budget_failure("站点 Flaky 搜索失败：HTTP 429") + assert not chain.consume_subscription_site_budget_failures() + + +def test_other_search_results_keep_site_failures_non_terminal(): + """插件等其他搜索源已有候选时，站点失败不得覆盖可用结果。""" + chain = object.__new__(SearchChain) + chain.configure_subscription_site_budget(None) + chain.record_subscription_site_budget_failure("站点 Flaky 搜索失败：HTTP 429") + + assert not chain.consume_subscription_site_budget_failures(has_results=True) def test_search_provider_aggregates_swallowed_indexer_failure(monkeypatch): @@ -420,6 +431,80 @@ def test_search_provider_aggregates_swallowed_indexer_failure(monkeypatch): assert snapshot.cooldown_seconds == 900.0 +def test_search_provider_explains_error_flag_without_exception(monkeypatch): + """索引器仅返回错误标志时也必须持久化可读原因，不能向任务暴露裸 `error`。""" + 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 + + monkeypatch.setattr( + IndexerModule, + "_IndexerModule__search_check", + staticmethod(lambda _site, _keyword=None: True), + ) + monkeypatch.setattr( + IndexerModule, + "_IndexerModule__execute_search", + staticmethod(lambda _site, _request: (True, [])), + ) + monkeypatch.setattr( + IndexerModule, + "_IndexerModule__indexer_statistic", + staticmethod(lambda **_kwargs: None), + ) + monkeypatch.setattr( + IndexerModule, + "_IndexerModule__parse_result", + staticmethod(lambda **_kwargs: []), + ) + budget = SubscriptionSiteBudget( + repository=_Repository(), + owner="fallback-worker", + cancelled=lambda: False, + stop_state=ProcessStopState(), + ) + chain = object.__new__(SearchChain) + chain.configure_subscription_site_budget(budget) + chain.search_site_torrents = object.__new__(IndexerModule).search_torrents + warnings = [] + monkeypatch.setattr( + "app.chain.search.provider.logger.warning", + warnings.append, + ) + + result = chain._search_site_torrents_with_budget( # pylint: disable=protected-access + site={"id": 15, "name": "Generic"}, + keyword="movie", + mtype=None, + page=0, + ) + + assert result == [] + assert captured["outcome"] == "error" + assert captured["error"] == "站点请求或页面解析失败" + assert chain.consume_subscription_site_budget_failures() == ( + "站点 Generic 搜索失败：站点请求或页面解析失败", + ) + assert warnings == ["站点 Generic 搜索失败：站点请求或页面解析失败"] + + def test_search_provider_logs_site_budget_release_failure(monkeypatch): """站点租约收口返回失败时必须进入摘要并即时记录错误。""" class _Repository(_WaitingRepository):