fix(subscription): always close search log summaries

2026-10-30 20:05:42 +08:00 · 2026-09-03 12:12:54 +08:00
parent eefc34076e
commit e50bd06905
4 changed files with 146 additions and 84 deletions
--- 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} "
--- 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:
        """请求取消持久搜索批次；未注入队列时返回失败。"""
--- 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`，不能只留在轮次
--- 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)]