From e50bd06905d9ef577751a499a7a268edb87184ea Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Thu, 3 Sep 2026 12:12:54 +0800 Subject: [PATCH] fix(subscription): always close search log summaries --- app/application/subscription/observability.py | 48 ++++++- app/chain/subscribe/search.py | 136 +++++++----------- ...iepilot-subscription-governance-roadmap.md | 5 +- tests/test_subscription_search_governance.py | 41 ++++++ 4 files changed, 146 insertions(+), 84 deletions(-) diff --git a/app/application/subscription/observability.py b/app/application/subscription/observability.py index 84b503400..f6da645e6 100644 --- a/app/application/subscription/observability.py +++ b/app/application/subscription/observability.py @@ -17,6 +17,48 @@ from app.application.subscription.sitebudget import SubscriptionSiteBudgetMetric SearchTaskOutcome = Literal["completed", "skipped", "failed", "cancelled", "requeued"] +def batch_progress_text(batch: Optional[SearchBatchSnapshot]) -> str: + """把批次聚合终态转为兼容进度文案。""" + if batch is None: + return "订阅搜索任务已提交" + if batch.state == "failed": + return "订阅搜索完成，部分任务失败" + if batch.state == "cancelled": + return "订阅搜索已取消" + if batch.state == "skipped": + return "订阅搜索完成，部分任务本轮已跳过" + if batch.state in {"queued", "running", "cancelling"}: + return "订阅搜索任务已排队" + if batch.skipped_count: + return "订阅搜索完成，部分任务本轮已跳过" + return "订阅搜索完成" + + +def inline_search_result(total: int, finished: int) -> tuple[str, dict[str, int]]: + """返回兼容搜索的真实终态文案与计数。""" + text = ( + "订阅搜索完成" + if finished == total + else "订阅搜索结束，部分订阅本轮未执行或未完成" + ) + return text, {"total": total, "finished": finished} + + +def batch_finished_count( + batch: Optional[SearchBatchSnapshot], + fallback: int, +) -> int: + """返回批次所有终态任务数；批次暂不可读时使用本轮实际完成数。""" + if batch is None: + return fallback + return ( + batch.finished_count + + batch.failed_count + + batch.cancelled_count + + batch.skipped_count + ) + + def finish_returned_search_task( *, queue: SubscriptionSearchRepository, @@ -186,6 +228,7 @@ class SearchExecutionSummary: release_failures: int = 0 consumer_conflicts: int = 0 stopped: bool = False + round_failed: bool = False site_metrics: SubscriptionSiteBudgetMetrics = field( default_factory=SubscriptionSiteBudgetMetrics ) @@ -223,7 +266,9 @@ class SearchExecutionSummary: def finish_log(self, batch: Optional[SearchBatchSnapshot] = None) -> str: """构造 Search 轮次结束日志，任务完成与订阅完成保持分离。""" sites = self.site_metrics.snapshot() - if batch is not None: + if self.round_failed: + state = "failed" + elif batch is not None: state = batch.state elif self.consumer_conflicts: state = "skipped" @@ -246,6 +291,7 @@ class SearchExecutionSummary: f"task_failed={self.failed} task_cancelled={self.cancelled} task_requeued={self.requeued} " f"admission_conflicts={self.admission_conflicts} cancelled={self.cancelled} " f"ttl_timeouts={self.ttl_timeouts} consumer_conflicts={self.consumer_conflicts} " + f"round_failed={int(self.round_failed)} " f"sites={sites.site_count} site_requests={sites.request_count} " f"site_failures={sites.failure_count} site_cooldown_skips={sites.cooldown_skip_count} " f"candidates={sites.candidate_count} cooldown_seconds={sites.cooldown_seconds:.1f} " diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index aa941d360..18ab1f546 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -23,7 +23,10 @@ from app.application.subscription.execution import ( from app.application.subscription.observability import ( SearchExecutionSummary, SearchTaskOutcome, + batch_finished_count, + batch_progress_text, finish_returned_search_task, + inline_search_result, ) from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.sitebudget import ( @@ -113,48 +116,6 @@ def _search_task_available_at( return available_at -def _batch_progress_text(batch: Optional[SearchBatchSnapshot]) -> str: - """把批次聚合终态转为兼容进度文案。""" - if batch is None: - return "订阅搜索任务已提交" - if batch.state == "failed": - return "订阅搜索完成，部分任务失败" - if batch.state == "cancelled": - return "订阅搜索已取消" - if batch.state == "skipped": - return "订阅搜索完成，部分任务本轮已跳过" - if batch.state in {"queued", "running", "cancelling"}: - return "订阅搜索任务已排队" - if batch.skipped_count: - return "订阅搜索完成，部分任务本轮已跳过" - return "订阅搜索完成" - - -def _inline_search_result(total: int, finished: int) -> tuple[str, dict[str, int]]: - """返回兼容搜索的真实终态文案与计数。""" - text = ( - "订阅搜索完成" - if finished == total - else "订阅搜索结束，部分订阅本轮未执行或未完成" - ) - return text, {"total": total, "finished": finished} - - -def _batch_finished_count( - batch: Optional[SearchBatchSnapshot], - fallback: int, -) -> int: - """返回批次所有终态任务数；批次暂不可读时使用本轮实际完成数。""" - if batch is None: - return fallback - return ( - batch.finished_count - + batch.failed_count - + batch.cancelled_count - + batch.skipped_count - ) - - def _skip_search_task( queue: SubscriptionSearchRepository, task: SearchTaskSnapshot, @@ -397,7 +358,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): ) self._notify_manual_search(manual, sid, sids, subscribes, processed) if progress_callback: - text, data = _inline_search_result(total, len(processed)) + text, data = inline_search_result(total, len(processed)) progress_callback(value=100, text=text, data=data) finally: subscribes.clear() @@ -442,39 +403,45 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): coalesced=enqueued.coalesced_count, ) logger.info(summary.start_log()) - if progress_callback: - progress_callback( - value=0, - text=f"开始订阅搜索，共 {total} 个订阅 ...", - data={ - "batch_id": enqueued.batch.batch_id, - "total": total, - "finished": 0, - "coalesced": enqueued.coalesced_count, - }, + batch: Optional[SearchBatchSnapshot] = None + try: + if progress_callback: + progress_callback( + value=0, + text=f"开始订阅搜索，共 {total} 个订阅 ...", + data={ + "batch_id": enqueued.batch.batch_id, + "total": total, + "finished": 0, + "coalesced": enqueued.coalesced_count, + }, + ) + processed = self._drain_search_queue( + queue=queue, + limit=max(1, enqueued.created_count + enqueued.coalesced_count), + progress_callback=progress_callback, + summary=summary, ) - processed = self._drain_search_queue( - queue=queue, - limit=max(1, enqueued.created_count + enqueued.coalesced_count), - progress_callback=progress_callback, - summary=summary, - ) - processed_subscribes = [item for item in subscribes if item.id in processed] - self._notify_manual_search(manual, sid, sids, subscribes, processed_subscribes) - batch = queue.get_batch(enqueued.batch.batch_id) - if progress_callback: - progress_callback( - value=100, - text=_batch_progress_text(batch), - data={ - "batch_id": enqueued.batch.batch_id, - "total": total, - "finished": _batch_finished_count(batch, len(processed_subscribes)), - "coalesced": enqueued.coalesced_count, - }, - ) - logger.info(summary.finish_log(batch)) - return str(enqueued.batch.batch_id) + processed_subscribes = [item for item in subscribes if item.id in processed] + self._notify_manual_search(manual, sid, sids, subscribes, processed_subscribes) + batch = queue.get_batch(enqueued.batch.batch_id) + if progress_callback: + progress_callback( + value=100, + text=batch_progress_text(batch), + data={ + "batch_id": enqueued.batch.batch_id, + "total": total, + "finished": batch_finished_count(batch, len(processed_subscribes)), + "coalesced": enqueued.coalesced_count, + }, + ) + return str(enqueued.batch.batch_id) + except Exception: + summary.round_failed = True + raise + finally: + logger.info(summary.finish_log(batch)) def _drain_search_queue( self, @@ -722,13 +689,18 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): return summary = SearchExecutionSummary(source="resume", requested=max(1, limit)) logger.info(summary.start_log()) - self._drain_search_queue( - queue=queue, - limit=max(1, limit), - progress_callback=progress_callback, - summary=summary, - ) - logger.info(summary.finish_log()) + try: + self._drain_search_queue( + queue=queue, + limit=max(1, limit), + progress_callback=progress_callback, + summary=summary, + ) + except Exception: + summary.round_failed = True + raise + finally: + logger.info(summary.finish_log()) def cancel_search_batch(self, batch_id: str) -> bool: """请求取消持久搜索批次；未注入队列时返回失败。""" diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index 64555b196..ef10ab0cb 100644 --- a/docs/refactor/moviepilot-subscription-governance-roadmap.md +++ b/docs/refactor/moviepilot-subscription-governance-roadmap.md @@ -494,7 +494,8 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 ## 10. 当前运行日志验收 当前日志用于上线后确认新治理链路是否健康，不承担旧新链路 A/B 对照，也不引入独立日志聚合器。日志内容遵循 -现有主日志输出方式，以稳定 `key=value` 字段提供机器可检索的轮次摘要。 +现有主日志输出方式，以稳定 `key=value` 字段提供机器可检索的轮次摘要。V3 生产组合根固定注入持久搜索队列， +站点请求聚合以该治理路径为验收对象；无队列的兼容执行路径保留任务摘要，但站点字段为零值。 ### 10.1 轮次摘要 @@ -503,6 +504,8 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 站点数和完整候选数； - 结束摘要使用 `task_completed`、`task_skipped`、`task_failed`、`task_cancelled` 等任务终态字段；其中 `task_completed` 只表示本次处理任务正常收口，不表示订阅已经下载完成或已经删除； +- Search 在队列消费、通知、批次查询或进度回调异常时仍输出配对的结束摘要，并以 `round_failed=1` 区分轮次 + 编排失败与单条任务失败； - `admission_conflicts`、`cancelled`、`ttl_timeouts` 分开计数，避免把准入跳过、用户取消和协作 TTL 超时混为 同一种“未执行”； - `release_failures` 汇总订阅准入与站点租约收口失败；任何单次释放失败同时立即输出 `ERROR`，不能只留在轮次 diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index df4419c26..ad7a42085 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -6,6 +6,7 @@ from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import Mock, patch +import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker @@ -311,6 +312,46 @@ def test_search_logs_subscription_admission_release_failure(tmp_path, monkeypatc assert "release_failures=1" in info_logs[-1] +def test_queued_search_logs_failed_finish_summary_when_callback_raises(tmp_path, monkeypatch): + """排队搜索开始后的外围异常仍须输出失败结束摘要。""" + chain = _chain(tmp_path, [_subscribe(53)]) + _make_tasks_ready(monkeypatch) + info_logs = [] + monkeypatch.setattr(subscribe_search.logger, "info", info_logs.append) + + with pytest.raises(RuntimeError, match="callback failed"): + chain.search( + state="R", + progress_callback=Mock(side_effect=RuntimeError("callback failed")), + ) + + assert len(info_logs) == 2 + assert info_logs[-1].startswith("订阅治理轮次结束: operation=search ") + assert "state=failed" in info_logs[-1] + assert "round_failed=1" in info_logs[-1] + + +def test_resume_search_logs_failed_finish_summary_when_drain_raises(tmp_path, monkeypatch): + """恢复消费开始后的外围异常仍须输出失败结束摘要。""" + chain = _chain(tmp_path, [_subscribe(54)]) + info_logs = [] + monkeypatch.setattr(subscribe_search.logger, "info", info_logs.append) + monkeypatch.setattr( + chain, + "_drain_search_queue", + Mock(side_effect=RuntimeError("claim failed")), + ) + + with pytest.raises(RuntimeError, match="claim failed"): + chain.resume_search_queue() + + assert len(info_logs) == 2 + assert info_logs[-1].startswith("订阅治理轮次结束: operation=search ") + assert "source=resume" in info_logs[-1] + assert "state=failed" in info_logs[-1] + assert "round_failed=1" in info_logs[-1] + + def test_swallowed_indexer_failure_marks_task_and_batch_failed_but_continues(tmp_path, monkeypatch): """索引器吞错后仍须失败收口当前任务，并继续处理后续订阅。""" subscribes = [_subscribe(13), _subscribe(14)]