From af47c1abeefc83de28f942e0b011b54850d563bf Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Thu, 3 Sep 2026 11:55:51 +0800 Subject: [PATCH] feat(subscription): add governance logging summaries --- app/application/subscription/sitebudget.py | 93 +++++++- app/chain/search/provider.py | 27 ++- app/chain/subscribe/match.py | 126 +++++++++-- app/chain/subscribe/search.py | 211 +++++++++++++++--- ...iepilot-subscription-governance-roadmap.md | 45 +++- .../subscription_governance_scale.py | 36 ++- tests/test_subscription_governance_scale.py | 5 +- tests/test_subscription_match_admission.py | 96 ++++++++ tests/test_subscription_search_governance.py | 86 ++++++- tests/test_subscription_site_budget.py | 66 ++++++ 10 files changed, 729 insertions(+), 62 deletions(-) diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index 97f7948d4..a134c8b45 100644 --- a/app/application/subscription/sitebudget.py +++ b/app/application/subscription/sitebudget.py @@ -2,6 +2,7 @@ from __future__ import annotations +import threading from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Callable, Optional, Protocol @@ -35,6 +36,74 @@ class SiteBudgetClaim: lease_token: Optional[str] = None +@dataclass(frozen=True, slots=True) +class SubscriptionSiteBudgetMetricsSnapshot: + """一次订阅搜索轮次内的站点预算观测快照。""" + + site_count: int + request_count: int + candidate_count: int + failure_count: int + cooldown_skip_count: int + release_failure_count: int + cooldown_seconds: float + + +class SubscriptionSiteBudgetMetrics: + """线程安全聚合站点请求、失败、冷却与租约收口结果。""" + + def __init__(self) -> None: + """初始化一次搜索轮次共享的计数器。""" + self._lock = threading.Lock() + self._site_ids: set[int] = set() + self._request_count = 0 + self._candidate_count = 0 + self._failure_count = 0 + self._cooldown_skip_count = 0 + self._release_failure_count = 0 + self._cooldown_seconds = 0.0 + + def record_unavailable(self, site_id: int) -> None: + """记录因在途租约或错误冷却而未发出的站点请求。""" + with self._lock: + self._site_ids.add(site_id) + self._cooldown_skip_count += 1 + + def record_request(self, site_id: int, candidate_count: int) -> None: + """记录一个真实站点请求及其返回候选数。""" + with self._lock: + self._site_ids.add(site_id) + self._request_count += 1 + self._candidate_count += max(0, candidate_count) + + def record_finish(self, *, outcome: str, delay: float, released: bool) -> None: + """记录外站结果、错误冷却时间和租约释放结果。""" + with self._lock: + if outcome not in {"success", "skipped"}: + self._failure_count += 1 + self._cooldown_seconds += max(0.0, delay) + if not released: + self._release_failure_count += 1 + + def record_release_failure(self) -> None: + """记录未进入正常结果收口的站点租约释放失败。""" + with self._lock: + self._release_failure_count += 1 + + def snapshot(self) -> SubscriptionSiteBudgetMetricsSnapshot: + """返回可安全跨线程读取的不可变观测结果。""" + with self._lock: + return SubscriptionSiteBudgetMetricsSnapshot( + site_count=len(self._site_ids), + request_count=self._request_count, + candidate_count=self._candidate_count, + failure_count=self._failure_count, + cooldown_skip_count=self._cooldown_skip_count, + release_failure_count=self._release_failure_count, + cooldown_seconds=round(self._cooldown_seconds, 3), + ) + + class SubscriptionSiteBudgetRepository(Protocol): """站点预算租约和冷却状态的持久化端口。""" @@ -79,6 +148,7 @@ class SubscriptionSiteBudget: lease_seconds: int = 900, clock: Callable[[], datetime] = _utc_now, phase_changed: Optional[Callable[[str, Optional[int]], None]] = None, + metrics: Optional[SubscriptionSiteBudgetMetrics] = None, ) -> None: """保存持久化端口及可注入的时钟和阶段回调。""" self._repository = repository @@ -88,6 +158,7 @@ class SubscriptionSiteBudget: self._lease_seconds = max(1, lease_seconds) self._clock = clock self._phase_changed = phase_changed + self._metrics = metrics def acquire(self, site_id: int) -> SiteBudgetClaim: """只认领一次指定站点，未就绪时留待下一次正常调度。""" @@ -100,6 +171,8 @@ class SubscriptionSiteBudget: if claim.acquired: self._report_phase("searching", site_id) return claim + if self._metrics: + self._metrics.record_unavailable(site_id) self._report_phase("waiting_site_budget", site_id) raise SubscriptionSiteBudgetUnavailable( site_id=site_id, @@ -111,20 +184,38 @@ class SubscriptionSiteBudget: if self._phase_changed: self._phase_changed(phase, site_id) + def record_request(self, site_id: int, candidate_count: int) -> None: + """把真实请求及返回候选数写入轮次聚合器。""" + if self._metrics: + self._metrics.record_request(site_id, candidate_count) + + def record_release_failure(self) -> None: + """把站点租约释放异常写入轮次聚合器。""" + if self._metrics: + self._metrics.record_release_failure() + def finish(self, claim: SiteBudgetClaim, observation: SiteSearchObservation) -> bool: """正常结果立即释放站点，错误结果按类别写入冷却。""" if not claim.lease_token: + self.record_release_failure() 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( + released = 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, ) + if self._metrics: + self._metrics.record_finish( + outcome=outcome, + delay=delay, + released=released, + ) + return released def _next_delay(self, outcome: str, consecutive_failures: int) -> float: """正常或本地跳过立即恢复，错误按连续失败次数退避。""" diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index c0f4b9f61..dd564f041 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -59,7 +59,7 @@ def _wait_for_site_request(site: SiteIndexer) -> None: """同步等待当前站点预约时间，不阻塞其他站点的线程任务。""" delay = _reserve_site_request(site) if delay > 0: - logger.info(f"{site.get('name')} 站点流控等待 {delay:.1f} 秒 ...") + logger.debug(f"{site.get('name')} 站点流控等待 {delay:.1f} 秒 ...") time.sleep(delay) @@ -67,7 +67,7 @@ async def _async_wait_for_site_request(site: SiteIndexer) -> None: """异步等待当前站点预约时间，让其他站点继续并发搜索。""" delay = _reserve_site_request(site) if delay > 0: - logger.info(f"{site.get('name')} 站点流控等待 {delay:.1f} 秒 ...") + logger.debug(f"{site.get('name')} 站点流控等待 {delay:.1f} 秒 ...") await asyncio.sleep(delay) @@ -216,7 +216,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): claim = budget.acquire(site_id) except SubscriptionSiteBudgetUnavailable as error: self.record_subscription_site_budget_failure(str(error)) - logger.info(str(error)) + logger.debug(str(error)) return [] with capture_site_search_observation() as observation: try: @@ -228,6 +228,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): page=page, ) except Exception as error: + budget.record_request(site_id, 0) report_site_search_outcome( attempted=True, outcome="error", @@ -235,6 +236,7 @@ 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( @@ -243,8 +245,14 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): return result finally: try: - budget.finish(claim, observation) + released = budget.finish(claim, observation) + if not released: + logger.error( + f"订阅站点预算释放失败: site_id={site_id} " + f"site={site.get('name') or '-'}" + ) except Exception as error: # noqa: BLE001 + budget.record_release_failure() logger.error( f"站点 {site.get('name') or site_id} 搜索预算收口失败：{str(error)}", exc_info=True, @@ -343,7 +351,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): search_keyword=search_keyword, media_type=media_type, ) - logger.info(f"站点搜索进度：{finish_count} / {total_num}") + logger.debug(f"站点搜索进度：{finish_count} / {total_num}") progress.update( value=finish_count / total_num * 100, text=(f"正在搜索{keyword or ''}，已完成 {finish_count} / {total_num} 个请求 ..."), @@ -401,7 +409,12 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): value=100, text=(f"站点搜索完成，有效资源数：{len(results)}，总耗时 {elapsed} 秒"), ) - logger.info(f"站点搜索完成，有效资源数：{len(results)}，总耗时 {elapsed} 秒") + log = ( + logger.debug + if isinstance(getattr(self, "_subscription_site_budget", None), SubscriptionSiteBudget) + else logger.info + ) + log(f"站点搜索完成，有效资源数：{len(results)}，总耗时 {elapsed} 秒") return results finally: progress.end() @@ -550,7 +563,7 @@ class SearchProviderOwner(_SearchProviderSyncOwner): logger.debug(f"{batch.site.get('name')} {label}第 {batch.page} 页返回 {len(batch.items)} 条，停止继续翻页") progress_value = batch.finished / batch.total * 100 progress_text = f"正在搜索{label}{keyword or ''}，已完成 {batch.finished} / {batch.total} 个请求 ..." - logger.info(f"站点{label}搜索进度：{batch.finished} / {batch.total}") + logger.debug(f"站点{label}搜索进度：{batch.finished} / {batch.total}") await progress.update(value=progress_value, text=progress_text) return { "type": "append", diff --git a/app/chain/subscribe/match.py b/app/chain/subscribe/match.py index f3e8658f1..52826a04c 100644 --- a/app/chain/subscribe/match.py +++ b/app/chain/subscribe/match.py @@ -1,9 +1,11 @@ """订阅候选准备、身份复核与资源匹配""" import copy +import time from dataclasses import dataclass from datetime import datetime from typing import Any, Callable, Dict, List, Literal, Optional +from uuid import uuid4 from app.application.configuration import get_configured_system_config from app.application.subscription.candidates import CandidateIndex @@ -43,16 +45,24 @@ class _MatchExecutionSummary: """保存一次 Match 批次的实际订阅执行结果。""" total: int + candidate_count: int + site_count: int + run_id: str + started_at: float completed: int = 0 skipped: int = 0 failed: int = 0 + admission_conflicts: int = 0 + cancelled: int = 0 + ttl_timeouts: int = 0 + release_failures: int = 0 @property def finished(self) -> int: """返回已经收口的订阅数量。""" return self.completed + self.skipped + self.failed - def record(self, outcome: Optional[str]) -> None: + def record(self, outcome: Optional[str], reason: Optional[str] = None) -> None: """将单订阅结果归入稳定的完成、跳过或失败计数。""" if outcome == "completed": self.completed += 1 @@ -60,6 +70,45 @@ class _MatchExecutionSummary: self.skipped += 1 else: self.failed += 1 + if reason == "admission_conflict": + self.admission_conflicts += 1 + elif reason == "cancelled": + self.cancelled += 1 + elif reason == "ttl_timeout": + self.ttl_timeouts += 1 + + @property + def state(self) -> str: + """按已处理数量和任务终态生成轮次状态。""" + if self.finished < self.total: + return "stopped" + if self.failed: + return "failed" + if self.skipped: + return "skipped" + return "completed" + + def start_log(self) -> str: + """构造 Match 轮次开始日志。""" + return ( + "订阅治理轮次开始: operation=match " + f"run_id={self.run_id} subscriptions={self.total} " + f"sites={self.site_count} candidates={self.candidate_count}" + ) + + def finish_log(self) -> str: + """构造 Match 轮次结束日志，completed 表示任务处理完成而非订阅成功。""" + elapsed_ms = (time.perf_counter() - self.started_at) * 1000 + return ( + "订阅治理轮次结束: operation=match " + f"run_id={self.run_id} state={self.state} subscriptions={self.total} " + f"processed={self.finished} task_completed={self.completed} " + f"task_skipped={self.skipped} task_failed={self.failed} " + f"admission_conflicts={self.admission_conflicts} cancelled={self.cancelled} " + f"ttl_timeouts={self.ttl_timeouts} release_failures={self.release_failures} " + f"sites={self.site_count} candidates={self.candidate_count} " + f"duration_ms={elapsed_ms:.1f}" + ) def as_data(self, current: Optional[int] = None) -> dict[str, int]: """构造供进度消费者读取的实际计数。""" @@ -92,6 +141,15 @@ def _report_match_progress( ) +def _match_stop_reason(context: SubscriptionExecutionContext) -> Optional[str]: + """区分 Match 安全停止来自 TTL 还是取消请求。""" + if context.is_expired(): + return "ttl_timeout" + if context.is_cancel_requested(): + return "cancelled" + return None + + def _is_title_match_allowed( mediainfo: MediaInfo, torrent_meta: MetaBase, @@ -109,7 +167,7 @@ def _is_title_match_allowed( torrent_meta=torrent_meta, ): return True - logger.info( + logger.debug( f"{torrent_info.site_name} - {torrent_info.title} 仅通过无年份别名命中且候选媒体身份无法确认，已跳过" ) return False @@ -224,7 +282,7 @@ def _prepare_subscription_match( site_ids=set(sub_sites) if sub_sites else None, ) if not routed_torrents: - logger.info(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备") + logger.debug(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备") return None mediainfo = fresh_fact_lease.get_or_load( subscribe, @@ -341,6 +399,14 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): progress_callback(value=0, text="正在预处理订阅资源 ...") lock_acquired = False + summary = _MatchExecutionSummary( + total=0, + candidate_count=sum(len(contexts) for contexts in torrents.values()), + site_count=len(torrents), + run_id=uuid4().hex, + started_at=time.perf_counter(), + ) + summary_started = False try: lock_acquired = self._acquire_run_lock("match", progress_callback) if not lock_acquired: @@ -352,7 +418,9 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): # 所有订阅 subscribes = self.subscription_repository.list(self.get_states_for_search("R")) total_num = len(subscribes) - summary = _MatchExecutionSummary(total=total_num) + summary.total = total_num + logger.info(summary.start_log()) + summary_started = True if progress_callback: _report_match_progress( progress_callback, @@ -373,13 +441,15 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): current=listed_subscribe.id, ) outcome: Optional[str] = "skipped" + reason: Optional[str] = "not_eligible" lease = self._subscription_execution_admission.try_acquire( subscription_id=listed_subscribe.id, operation="match", ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL, ) if lease is None: - logger.info(f"订阅 {listed_subscribe.name} 正在由其他通道处理，本轮匹配已跳过") + reason = "admission_conflict" + logger.debug(f"订阅 {listed_subscribe.name} 正在由其他通道处理，本轮匹配已跳过") else: current_subscribe = None execution_context = SubscriptionExecutionContext( @@ -390,13 +460,14 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): try: current_subscribe = self.subscription_repository.get(listed_subscribe.id) if current_subscribe is None: - logger.info(f"订阅 {listed_subscribe.id} 已不存在，本轮匹配跳过") + logger.debug(f"订阅 {listed_subscribe.id} 已不存在，本轮匹配跳过") elif current_subscribe.state not in {"R", "P"}: - logger.info( + logger.debug( f"订阅 {current_subscribe.name} 当前状态为 {current_subscribe.state}，本轮匹配跳过" ) elif execution_context.should_stop(): - logger.info(f"订阅 {current_subscribe.name} 已取消或超时，本轮匹配跳过") + reason = _match_stop_reason(execution_context) + logger.debug(f"订阅 {current_subscribe.name} 已取消或超时，本轮匹配跳过") else: outcome = self._match_subscription( subscribe=current_subscribe, @@ -405,18 +476,31 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): fresh_fact_lease=fresh_fact_lease, execution_context=execution_context, ) + reason = ( + _match_stop_reason(execution_context) + if outcome == "skipped" + else None + ) or ("business_skip" if outcome == "skipped" else None) except SubscriptionSearchCancelled as err: outcome = "skipped" - logger.info(f"订阅 {listed_subscribe.name} 匹配已取消：{str(err)}") + reason = _match_stop_reason(execution_context) or "cancelled" + logger.debug(f"订阅 {listed_subscribe.name} 匹配已取消：{str(err)}") except Exception as err: outcome = "failed" + reason = "error" subscribe_name = ( current_subscribe.name if current_subscribe is not None else listed_subscribe.name ) logger.error(f"订阅 {subscribe_name} 匹配失败：{str(err)}", exc_info=True) finally: - self._subscription_execution_admission.release(lease) - summary.record(outcome) + released = self._subscription_execution_admission.release(lease) + if not released: + summary.release_failures += 1 + logger.error( + "订阅准入释放失败: operation=match " + f"subscription_id={listed_subscribe.id} run_id={summary.run_id}" + ) + summary.record(outcome, reason) _report_match_progress( progress_callback, summary, @@ -443,6 +527,8 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): text=final_text, ) finally: + if summary_started: + logger.info(summary.finish_log()) if lock_acquired: self._match_lock.release() logger.debug(f"match Lock released at {datetime.now()}") @@ -477,7 +563,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): """在单条订阅准入期间完成匹配、下载和完成判定。""" if execution_context.should_stop(): return "skipped" - logger.info(f"开始匹配订阅，标题：{subscribe.name} ...") + logger.debug(f"开始匹配订阅，标题：{subscribe.name} ...") if subscribe.type == MediaType.MUSIC.value: music_contexts = [context for contexts in processed_torrents.values() for context in contexts] execution_context.report_phase("matching") @@ -551,7 +637,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): torrent_meta.org_string, custom_words=custom_words_list ) if apply_words: - logger.info( + logger.debug( f"{torrent_info.site_name} - {torrent_info.title} 因订阅存在自定义识别词，重新识别元数据..." ) # 重新识别元数据 @@ -593,7 +679,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): ): continue # 匹配成功 - logger.info( + logger.debug( f"{mediainfo.title_year} 通过标题匹配到可选资源：{torrent_info.site_name} - {torrent_info.title}" ) torrent_mediainfo = mediainfo @@ -633,7 +719,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): _context.candidate_recognized = True _context.media_info_is_target = False match_label = "媒体ID" - logger.info( + logger.debug( f"{mediainfo.title_year} 通过{match_label}匹配到可选资源：" f"{torrent_info.site_name} - {torrent_info.title}" ) @@ -723,7 +809,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): context=_context, priority=torrent_info.pri_order, ): - logger.info( + logger.debug( f"{subscribe.name} 正在洗版，{torrent_info.title} " f"优先级未达到当前模式的升级条件" ) @@ -733,13 +819,13 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): and subscribe.current_priority and torrent_info.pri_order <= subscribe.current_priority ): - logger.info( + logger.debug( f"{subscribe.name} 正在洗版，{torrent_info.title} 优先级低于或等于已下载优先级" ) continue # 匹配成功 - logger.info(f"{mediainfo.title_year} 匹配成功：{torrent_info.title}") + logger.debug(f"{mediainfo.title_year} 匹配成功：{torrent_info.title}") # 自定义属性 if subscribe.media_category: torrent_mediainfo.category = subscribe.media_category @@ -752,7 +838,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): if not _match_context: # 未匹配到资源 - logger.info(f"{mediainfo.title_year} 未匹配到符合条件的资源") + logger.debug(f"{mediainfo.title_year} 未匹配到符合条件的资源") self.finish_subscribe_or_not( subscribe=subscribe, meta=meta, mediainfo=mediainfo, lefts=no_exists ) @@ -762,7 +848,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): return "skipped" # 开始批量择优下载 - logger.info(f"{mediainfo.title_year} 匹配完成，共匹配到{len(_match_context)}个资源") + logger.debug(f"{mediainfo.title_year} 匹配完成，共匹配到{len(_match_context)}个资源") downloads, lefts = self._SubscribeChain__download_best_version_with_full_pack_first( contexts=_match_context, no_exists=no_exists, diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index fd9160b80..8068d4992 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -2,9 +2,10 @@ import random import time +from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from functools import partial -from typing import Any, Callable, Optional, cast +from typing import Any, Callable, Literal, Optional, cast from uuid import uuid4 from app.application.configuration import get_configured_system_config @@ -24,6 +25,7 @@ from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.sitebudget import ( SubscriptionSearchCancelled, SubscriptionSiteBudget, + SubscriptionSiteBudgetMetrics, ) from app.chain.media import MediaChain from app.chain.search.facade import SearchChain @@ -179,6 +181,9 @@ def _release_cancelled_or_stopped_search_task( ) +_SearchTaskOutcome = Literal["completed", "skipped", "failed", "cancelled", "requeued"] + + def _finish_returned_search_task( *, queue: SubscriptionSearchRepository, @@ -188,7 +193,7 @@ def _finish_returned_search_task( execution_context: SubscriptionExecutionContext, system_stopped: bool, cancel_requested: bool, -) -> Optional[int]: +) -> tuple[Optional[int], _SearchTaskOutcome, Optional[str]]: """按 TTL、停机和取消边界收口正常返回的搜索任务。""" download_started = execution_context.download_started if execution_context.is_expired() and not (system_stopped or cancel_requested): @@ -199,14 +204,14 @@ def _finish_returned_search_task( state="completed", error="执行截止时间晚于下载提交边界，已按实际结果完成", ) - return subscription_id + return subscription_id, "completed", "ttl_timeout" queue.finish_task( task_id=task_id, lease_token=lease_token, state="failed", error="订阅执行已超过协作截止时间", ) - return None + return None, "failed", "ttl_timeout" if system_stopped: if download_started: queue.finish_task( @@ -215,9 +220,9 @@ def _finish_returned_search_task( state="completed", error="停机请求晚于下载提交边界，已按实际结果完成", ) - return subscription_id + return subscription_id, "completed", "system_stop" queue.release_task(task_id=task_id, lease_token=lease_token) - return None + return None, "requeued", "system_stop" if cancel_requested: if download_started: queue.finish_task( @@ -226,11 +231,99 @@ def _finish_returned_search_task( state="completed", error="取消请求晚于下载提交边界，已按实际结果完成", ) - return subscription_id + return subscription_id, "completed", "cancelled" queue.release_task(task_id=task_id, lease_token=lease_token, cancelled=True) - return None + return None, "cancelled", "cancelled" queue.finish_task(task_id=task_id, lease_token=lease_token, state="completed") - return subscription_id + return subscription_id, "completed", None + + +@dataclass(slots=True) +class _SearchExecutionSummary: + """聚合一次 Search 消费轮次的任务、站点与候选观测。""" + + source: str + requested: int + run_id: str = field(default_factory=lambda: uuid4().hex) + batch_id: Optional[str] = None + coalesced: int = 0 + started_at: float = field(default_factory=time.perf_counter) + completed: int = 0 + skipped: int = 0 + failed: int = 0 + cancelled: int = 0 + requeued: int = 0 + admission_conflicts: int = 0 + ttl_timeouts: int = 0 + release_failures: int = 0 + consumer_conflicts: int = 0 + stopped: bool = False + site_metrics: SubscriptionSiteBudgetMetrics = field( + default_factory=SubscriptionSiteBudgetMetrics + ) + + @property + def processed(self) -> int: + """返回本轮实际取得并收口或退回队列的任务数。""" + return self.completed + self.skipped + self.failed + self.cancelled + self.requeued + + def record(self, outcome: _SearchTaskOutcome, reason: Optional[str] = None) -> None: + """记录一个任务结果，并单独统计冲突、取消和 TTL。""" + if outcome == "completed": + self.completed += 1 + elif outcome == "skipped": + self.skipped += 1 + elif outcome == "failed": + self.failed += 1 + elif outcome == "cancelled": + self.cancelled += 1 + else: + self.requeued += 1 + if reason == "admission_conflict": + self.admission_conflicts += 1 + elif reason == "ttl_timeout": + self.ttl_timeouts += 1 + + def start_log(self) -> str: + """构造 Search 轮次开始日志。""" + return ( + "订阅治理轮次开始: operation=search " + f"run_id={self.run_id} batch_id={self.batch_id or '-'} source={self.source} " + f"subscriptions={self.requested} coalesced={self.coalesced}" + ) + + def finish_log(self, batch: Optional[SearchBatchSnapshot] = None) -> str: + """构造 Search 轮次结束日志，任务完成与订阅完成保持分离。""" + sites = self.site_metrics.snapshot() + if batch is not None: + state = batch.state + elif self.consumer_conflicts: + state = "skipped" + elif self.requeued or self.stopped: + state = "stopped" + elif self.failed: + state = "failed" + elif self.cancelled: + state = "cancelled" + elif self.skipped: + state = "skipped" + else: + state = "completed" + elapsed_ms = (time.perf_counter() - self.started_at) * 1000 + return ( + "订阅治理轮次结束: operation=search " + f"run_id={self.run_id} batch_id={self.batch_id or '-'} source={self.source} " + f"state={state} subscriptions={self.requested} processed={self.processed} " + f"task_completed={self.completed} task_skipped={self.skipped} " + 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"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} " + f"release_failures={self.release_failures + sites.release_failure_count} " + f"duration_ms={elapsed_ms:.1f}" + ) class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): @@ -352,9 +445,12 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): return subscribes = [] processed = [] + summary: Optional[_SearchExecutionSummary] = None try: subscribes = self._load_search_subscriptions(sid=sid, sids=sids, state=state) total = len(subscribes) + summary = _SearchExecutionSummary(source="inline", requested=total) + logger.info(summary.start_log()) if progress_callback: progress_callback( value=0, @@ -364,9 +460,11 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): searchchain = SearchChain() for index, subscribe in enumerate(subscribes, start=1): if runtime_stop_state.is_system_stopped: + summary.stopped = True break self._report_search_progress(progress_callback, subscribe, index, total) if self._defer_recent_subscription(subscribe): + summary.record("skipped", "recent_subscription") continue self._wait_before_scheduled_search(sid, sids, state, progress_callback) lease = self._subscription_execution_admission.try_acquire( @@ -375,7 +473,8 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL, ) if lease is None: - logger.info(f"订阅 {subscribe.name} 正在由其他通道处理，本轮搜索已跳过") + logger.debug(f"订阅 {subscribe.name} 正在由其他通道处理，本轮搜索已跳过") + summary.record("skipped", "admission_conflict") continue execution_context = SubscriptionExecutionContext( lease=lease, @@ -383,11 +482,13 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): cancel_requested=lambda: runtime_stop_state.is_system_stopped, ) current = None + outcome: _SearchTaskOutcome = "skipped" + reason: Optional[str] = "not_eligible" try: current = self.subscription_repository.get(subscribe.id) if current is None or current.state == "S": if current and current.state == "S": - logger.info(f"订阅 {current.name} 已暂停，本轮搜索已跳过") + logger.debug(f"订阅 {current.name} 已暂停，本轮搜索已跳过") continue processed_result = self._process_search_subscription( current, @@ -396,9 +497,15 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): ) processed.append(processed_result or current) current = processed_result + outcome = "completed" + reason = None except SubscriptionSearchCancelled: - logger.info(f"订阅 {subscribe.name} 搜索已在安全边界取消") + reason = "ttl_timeout" if execution_context.is_expired() else "cancelled" + outcome = "failed" if reason == "ttl_timeout" else "cancelled" + logger.debug(f"订阅 {subscribe.name} 搜索已在安全边界取消") except Exception as err: + outcome = "failed" + reason = "error" logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True) finally: try: @@ -414,7 +521,14 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): exc_info=True, ) finally: - self._subscription_execution_admission.release(lease) + released = self._subscription_execution_admission.release(lease) + if not released: + summary.release_failures += 1 + logger.error( + "订阅准入释放失败: operation=search " + f"subscription_id={subscribe.id} run_id={summary.run_id}" + ) + summary.record(outcome, reason) self._report_search_progress( progress_callback, subscribe, @@ -429,6 +543,8 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): finally: subscribes.clear() self._search_queue_lock.release() + if summary is not None: + logger.info(summary.finish_log()) logger.debug(f"search Lock released at {datetime.now()}") def _execute_queued_search( @@ -460,6 +576,13 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): ), ) total = len(subscribes) + summary = _SearchExecutionSummary( + source=source, + requested=total, + batch_id=str(enqueued.batch.batch_id), + coalesced=enqueued.coalesced_count, + ) + logger.info(summary.start_log()) if progress_callback: progress_callback( value=0, @@ -475,11 +598,12 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): 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: - batch = queue.get_batch(enqueued.batch.batch_id) progress_callback( value=100, text=_batch_progress_text(batch), @@ -490,6 +614,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): "coalesced": enqueued.coalesced_count, }, ) + logger.info(summary.finish_log(batch)) return str(enqueued.batch.batch_id) def _drain_search_queue( @@ -498,10 +623,12 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): queue: SubscriptionSearchRepository, limit: int, progress_callback: Optional[Callable[..., None]], + summary: _SearchExecutionSummary, ) -> set[int]: """有界消费可恢复任务；单任务失败不得阻止后续订阅。""" if not self._search_queue_lock.acquire(blocking=False): logger.debug("订阅搜索队列已有消费者，本轮仅保留持久任务") + summary.consumer_conflicts += 1 return set() owner = f"subscribe-search:{uuid4().hex}" processed: set[int] = set() @@ -509,6 +636,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): searchchain = SearchChain() for index in range(1, limit + 1): if runtime_stop_state.is_system_stopped: + summary.stopped = True break task = queue.claim_next(owner=owner) if task is None: @@ -521,6 +649,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): index=index, limit=limit, progress_callback=progress_callback, + summary=summary, ) if subscription_id is not None: processed.add(subscription_id) @@ -538,10 +667,12 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): index: int, limit: int, progress_callback: Optional[Callable[..., None]], + summary: _SearchExecutionSummary, ) -> Optional[int]: """执行一条已认领任务，返回实际进入搜索处理的订阅 ID。""" if task.lease_token is None: logger.error(f"订阅搜索任务 {task.task_id} 缺少租约令牌，跳过执行") + summary.record("failed", "missing_lease") return None task_id = str(task.task_id) lease_token = task.lease_token @@ -549,6 +680,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): stop_state = getattr(self, "stop_state", runtime_stop_state) if cancelled(): queue.release_task(task_id=task_id, lease_token=lease_token, cancelled=True) + summary.record("cancelled", "cancelled") return None subscribe = self.subscription_repository.get(task.subscription_id) if subscribe is None: @@ -558,10 +690,12 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): state="cancelled", error="订阅已不存在", ) + summary.record("cancelled", "missing_subscription") return None self._report_search_progress(progress_callback, subscribe, index, limit) if self._defer_recent_subscription(subscribe): _skip_search_task(queue, task, "订阅仍在新增保护期，本轮搜索已跳过") + summary.record("skipped", "recent_subscription") return None lease = self._subscription_execution_admission.try_acquire( subscription_id=subscribe.id, @@ -569,8 +703,9 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL, ) if lease is None: - logger.info(f"订阅 {subscribe.name} 正在由其他通道处理，本轮搜索已跳过") + logger.debug(f"订阅 {subscribe.name} 正在由其他通道处理，本轮搜索已跳过") _skip_search_task(queue, task, "同一订阅正在由其他通道处理，本轮搜索已跳过") + summary.record("skipped", "admission_conflict") return None phase_changed = partial(_update_search_task_phase, queue, task_id, lease_token) execution_context = SubscriptionExecutionContext( @@ -590,9 +725,11 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): state="cancelled", error="订阅已不存在", ) + summary.record("cancelled", "missing_subscription") return None if current.state == "S": _skip_search_task(queue, task, "订阅已暂停，本轮搜索已跳过") + summary.record("skipped", "paused") return None searchchain.configure_subscription_site_budget( SubscriptionSiteBudget( @@ -601,6 +738,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): cancelled=execution_context.should_stop, stop_state=stop_state, phase_changed=phase_changed, + metrics=summary.site_metrics, ) ) current = self._process_search_subscription( @@ -610,7 +748,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): ) system_stopped = stop_state.is_system_stopped cancel_requested = False if system_stopped else cancelled() - return _finish_returned_search_task( + subscription_id, outcome, reason = _finish_returned_search_task( queue=queue, task_id=task_id, lease_token=lease_token, @@ -619,6 +757,8 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): system_stopped=system_stopped, cancel_requested=cancel_requested, ) + summary.record(outcome, reason) + return subscription_id except SubscriptionSearchCancelled: if execution_context.is_expired() and not execution_context.is_cancel_requested(): queue.finish_task( @@ -627,12 +767,18 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): state="failed", error="订阅执行已超过协作截止时间", ) + summary.record("failed", "ttl_timeout") else: + system_stopped = stop_state.is_system_stopped _release_cancelled_or_stopped_search_task( queue, task_id, lease_token, - stop_state.is_system_stopped, + system_stopped, + ) + summary.record( + "requeued" if system_stopped else "cancelled", + "system_stop" if system_stopped else "cancelled", ) except Exception as err: logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True) @@ -642,6 +788,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): state="failed", error=str(err), ) + summary.record("failed", "error") finally: self._cleanup_search_task( queue=queue, @@ -652,6 +799,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): progress_callback=progress_callback, index=index, limit=limit, + summary=summary, ) return None @@ -666,6 +814,7 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): progress_callback: Optional[Callable[..., None]], index: int, limit: int, + summary: _SearchExecutionSummary, ) -> None: """清理站点预算和订阅状态，并在所有异常路径释放 owner。""" try: @@ -682,7 +831,13 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): except Exception as err: logger.error(f"订阅 {subscribe.name} 搜索后状态重置失败：{str(err)}", exc_info=True) finally: - self._subscription_execution_admission.release(lease) + released = self._subscription_execution_admission.release(lease) + if not released: + summary.release_failures += 1 + logger.error( + "订阅准入释放失败: operation=search " + f"subscription_id={subscribe.id} run_id={summary.run_id}" + ) self._report_search_progress( progress_callback, subscribe, @@ -702,11 +857,15 @@ class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): ) if queue is None: 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()) def cancel_search_batch(self, batch_id: str) -> bool: """请求取消持久搜索批次；未注入队列时返回失败。""" @@ -763,7 +922,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): if sid or sids is not None or state not in {"R", "P"}: return sleep_time = random.randint(60, 300) - logger.info(f"订阅搜索随机休眠 {sleep_time} 秒 ...") + logger.debug(f"订阅搜索随机休眠 {sleep_time} 秒 ...") if progress_callback: progress_callback(text=f"订阅搜索随机休眠 {sleep_time} 秒后继续 ...") time.sleep(sleep_time) @@ -776,7 +935,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): ) -> Optional[SubscriptionSnapshot]: """处理单个订阅，并返回下载后重新读取的状态快照。""" _ensure_execution_active(execution_context) - logger.info(f"开始搜索订阅，标题：{subscribe.name} ...") + logger.debug(f"开始搜索订阅，标题：{subscribe.name} ...") if subscribe.type == MediaType.MUSIC.value: self._search_music_subscribe(subscribe, execution_context=execution_context) return subscribe @@ -828,7 +987,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): site_budget_failures = searchchain.consume_subscription_site_budget_failures() _ensure_execution_active(execution_context) if not contexts: - logger.warning(f"订阅 {subscribe.keyword or subscribe.name} 未搜索到资源") + logger.debug(f"订阅 {subscribe.keyword or subscribe.name} 未搜索到资源") self.finish_subscribe_or_not( subscribe=subscribe, meta=meta, @@ -839,7 +998,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): return subscribe matched = self._filter_search_contexts(subscribe, contexts) if not matched: - logger.warning(f"订阅 {subscribe.name} 没有符合过滤条件的资源") + logger.debug(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 @@ -891,17 +1050,17 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): media = context.media_info if subscribe.best_version and media.type == MediaType.TV: if not self._SubscribeChain__is_full_season_best_version_resource(torrent_meta, subscribe): - logger.info(f"{subscribe.name} 正在全集洗版，{torrent_info.title} 不是全集资源") + logger.debug(f"{subscribe.name} 正在全集洗版，{torrent_info.title} 不是全集资源") continue if not self._is_episode_range_covered(torrent_meta, subscribe): - logger.info(f"{subscribe.name} 正在洗版，{torrent_info.title} 不符合订阅集数范围") + logger.debug(f"{subscribe.name} 正在洗版，{torrent_info.title} 不符合订阅集数范围") continue if not self._SubscribeChain__prepare_best_version_tv_candidate( subscribe, context, torrent_info.pri_order, ): - logger.info(f"{subscribe.name} 正在洗版，{torrent_info.title} 优先级未达到当前模式的升级条件") + logger.debug(f"{subscribe.name} 正在洗版，{torrent_info.title} 优先级未达到当前模式的升级条件") continue if ( subscribe.best_version @@ -909,7 +1068,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): and subscribe.current_priority and torrent_info.pri_order <= subscribe.current_priority ): - logger.info(f"{subscribe.name} 正在洗版，{torrent_info.title} 优先级低于或等于已下载优先级") + logger.debug(f"{subscribe.name} 正在洗版，{torrent_info.title} 优先级低于或等于已下载优先级") continue if subscribe.media_category: media.category = subscribe.media_category diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index bee7cb845..64555b196 100644 --- a/docs/refactor/moviepilot-subscription-governance-roadmap.md +++ b/docs/refactor/moviepilot-subscription-governance-roadmap.md @@ -3,6 +3,7 @@ > 历史方案：本文件记录 2026-09-01 的上游接管与验证证据，不再作为当前实施合同。 > 其中 `CandidateBatch`、候选恢复账本、订阅下载提交持久账本和 `reconcile_required` > 已由 2026-09-03 的简化方案废止；当前行为以代码、迁移和回归测试为准。 +> 当前治理链路的运行日志验收以第 10 节为准。 > > 状态：`completed（2026-09-01）` > 当前叶：`none（全部叶已完成或按条件停止）` @@ -417,7 +418,7 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 5. 24 小时兜底搜索无结果、部分结果、站点超时和 `429/403`； 6. 日常 Match 与兜底搜索重叠； 7. 手工刷新或手工搜索与后台任务重叠； -8. 下载器已接受但本地写回失败； +8. 下载器拒绝、超时或本地写回失败，并在下轮按普通订阅流程重新判断； 9. 停机、取消和进程重启后的恢复； 10. 单媒体服务器可用、慢响应和不可用。 @@ -432,9 +433,9 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 | 本地计算 | 候选预处理、每订阅匹配、完成对账、提交耗时 | | TMDB | 请求数、缓存/租约命中、错误率、p50/p95 | | 媒体服务器 | 请求数、p50/p95、超时和合并命中 | -| PT 站点 | 按站点请求数、峰值在途、最小间隔、429/403/超时、重试数 | +| PT 站点 | 按站点请求数、峰值在途、最小间隔、429/403/超时、错误冷却跳过数 | | 正确性 | 命中集合、下载集合、缺集集合、完成/删除集合与基线一致 | -| 幂等 | 重复提交数、下载器不确定终态、恢复后重复任务数 | +| 并发 | 同订阅跨 Search/Match 准入冲突数、重复下载数、准入与站点租约释放失败数 | | 生命周期 | 取消延迟、event-loop lag、遗留 task/worker/锁、shutdown 上限 | ### 7.3 门禁 @@ -475,8 +476,9 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 - 现有测试只证明了参数或局部调用，没有证明 TMDB 请求减少、订阅完成正确、站点压力不变和下载集合一致； - 旧 checkpoint `c2d72280e` 与 10/100/500 数据只覆盖兜底搜索等待，不再是 A1B 的交付候选。 -候选回写语义、完整缓存与本轮 delta、无损索引和真实性测试仍需由后续实现证明。治理文档现位于 -`docs/refactor/` 并属于后端 Git 仓；文档状态与代码提交证据仍分别记录，不能用路线图状态替代行为验收。 +本节所述候选回写、完整缓存与本轮 delta 属于已废止的历史实施路径，不再作为后续实现或日志验收指标。 +治理文档现位于 `docs/refactor/` 并属于后端 Git 仓；文档状态与代码提交证据仍分别记录，不能用路线图 +状态替代行为验收。 ## 9. 停止条件 @@ -488,3 +490,36 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 4. 用户可以看到真实业务状态和失败终态； 5. 所有保留的并发和缓存都有可重放验证、版本级回滚路径和明确 owner； 6. 后续只剩最低优先级的重复订阅产品规则或无生产证据的推测项。 + +## 10. 当前运行日志验收 + +当前日志用于上线后确认新治理链路是否健康，不承担旧新链路 A/B 对照，也不引入独立日志聚合器。日志内容遵循 +现有主日志输出方式，以稳定 `key=value` 字段提供机器可检索的轮次摘要。 + +### 10.1 轮次摘要 + +- Search 与 Match 每次实际执行各输出一条 `订阅治理轮次开始` 和一条 `订阅治理轮次结束` 的 `INFO`； +- 两类摘要均包含 `operation`、`run_id`、订阅数和耗时；Search 额外包含 `batch_id` 与来源，Match 包含输入 + 站点数和完整候选数； +- 结束摘要使用 `task_completed`、`task_skipped`、`task_failed`、`task_cancelled` 等任务终态字段；其中 + `task_completed` 只表示本次处理任务正常收口，不表示订阅已经下载完成或已经删除； +- `admission_conflicts`、`cancelled`、`ttl_timeouts` 分开计数，避免把准入跳过、用户取消和协作 TTL 超时混为 + 同一种“未执行”； +- `release_failures` 汇总订阅准入与站点租约收口失败；任何单次释放失败同时立即输出 `ERROR`，不能只留在轮次 + 计数里。 + +### 10.2 站点与候选摘要 + +- Search 结束摘要聚合 `sites`、`site_requests`、`site_failures`、`site_cooldown_skips`、`candidates` 和 + `cooldown_seconds`；正常成功不产生额外冷却，错误冷却和已有在途租约导致的未发请求统一计入跳过数； +- Match 的 `candidates` 表示本轮完整候选快照数量，不记录候选标题、下载地址或逐项匹配明细； +- 逐订阅开始、正常无结果、准入冲突，逐候选匹配/过滤，以及逐页站点进度降为 `DEBUG`；真实异常、媒体识别 + 失败和下载业务结果仍保留原有错误或业务日志级别。 + +### 10.3 验证门禁 + +- 日志合同测试必须覆盖 Search/Match 开始与结束摘要、任务完成语义、准入冲突、取消、TTL、站点请求、错误冷却 + 和两类释放失败； +- `100/10/1000` 与 `200/20/1200` 两档受控 Match 快照中，Match 模块的 `INFO` 数量固定为两条轮次摘要， + 不得随订阅数、站点数或候选乘积增长； +- 固定快照只验证日志量、字段和本地治理合同，不把它解释为真实 PT/TMDB/媒体服务器的长期网络 SLO。 diff --git a/scripts/validation/subscription_governance_scale.py b/scripts/validation/subscription_governance_scale.py index cdb416205..eaf843d28 100644 --- a/scripts/validation/subscription_governance_scale.py +++ b/scripts/validation/subscription_governance_scale.py @@ -782,8 +782,31 @@ def _run_match_execution_case(case: ScaleCase) -> dict[str, Any]: settlement_ms: list[float] = [] recognition_calls = 0 progress_snapshots: list[dict[str, int]] = [] + info_logs: list[str] = [] started = time.perf_counter() + class _ScaleMatchLogger: + """只收集 Match 模块 INFO，用于验证日志量不随候选乘积增长。""" + + @staticmethod + def debug(*_args: Any, **_kwargs: Any) -> None: + """忽略受控样本的逐项诊断日志。""" + + @staticmethod + def warning(*_args: Any, **_kwargs: Any) -> None: + """忽略受控样本中预期不会出现的警告。""" + + warn = warning + + @staticmethod + def error(*_args: Any, **_kwargs: Any) -> None: + """错误仍由集合和终态门禁暴露，不污染 INFO 计数。""" + + @staticmethod + def info(message: Any, *_args: Any, **_kwargs: Any) -> None: + """保存结构化批次摘要。""" + info_logs.append(str(message)) + class _ScaleMediaChain: """用固定新鲜媒体事实替代目标规模下的真实 TMDB 请求。""" @@ -905,6 +928,7 @@ def _run_match_execution_case(case: ScaleCase) -> dict[str, Any]: "runtime_stop_state", SimpleNamespace(is_system_stopped=False), ), + patch.object(subscribe_match, "logger", _ScaleMatchLogger()), patch.object( download_batch, "runtime_stop_state", @@ -936,6 +960,12 @@ def _run_match_execution_case(case: ScaleCase) -> dict[str, Any]: "skipped": 0, "failed": 0, }, + "info_log_count": len(info_logs), + "info_log_bounded": bool( + len(info_logs) == 2 + and info_logs[0].startswith("订阅治理轮次开始: operation=match ") + and info_logs[1].startswith("订阅治理轮次结束: operation=match ") + ), "first_settlement_local_ms": round(settlement_ms[0], 3), "subscription_settlement_local_p50_ms": _percentile(settlement_ms, 50), "subscription_settlement_local_p95_ms": _percentile(settlement_ms, 95), @@ -1163,6 +1193,10 @@ def run_acceptance() -> dict[str, Any]: case["fresh_fact_loads"] == match_case["media_count"] for case, match_case in zip(match_execution_cases, match_cases) ), + "match_info_logs_bounded": all( + case["info_log_bounded"] + for case in match_execution_cases + ), "queue_completed": all( case["queue_state"] == "completed" and case["queue_finished_count"] == next( @@ -1210,7 +1244,7 @@ def run_acceptance() -> dict[str, Any]: ), } return { - "schema_version": 4, + "schema_version": 5, "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "revision": _git_revision(), "environment": { diff --git a/tests/test_subscription_governance_scale.py b/tests/test_subscription_governance_scale.py index dd8c5699e..921781b0e 100644 --- a/tests/test_subscription_governance_scale.py +++ b/tests/test_subscription_governance_scale.py @@ -22,7 +22,7 @@ def test_subscription_governance_controlled_scale_matrix() -> None: """两档最终矩阵必须同时满足正确性、压力、恢复和订阅准入门禁。""" result = run_acceptance() - assert result["schema_version"] == 4 + assert result["schema_version"] == 5 assert result["passed"] is True assert all(result["gates"].values()) assert result["method"]["match_entrypoint"] == "SubscribeChain.match" @@ -51,6 +51,7 @@ def test_subscription_governance_controlled_scale_matrix() -> None: assert result["gates"]["match_execution_download_sets_equal"] is True assert result["gates"]["match_execution_missing_sets_equal"] is True assert result["gates"]["match_execution_completion_sets_equal"] is True + assert result["gates"]["match_info_logs_bounded"] is True assert [case["subscription_count"] for case in result["match_cases"]] == [100, 200] assert [case["site_count"] for case in result["match_cases"]] == [10, 20] assert min(case["candidate_count"] for case in result["match_cases"]) >= 1000 @@ -69,6 +70,8 @@ def test_subscription_governance_controlled_scale_matrix() -> None: assert all( case["downloaded_candidates"]["duplicate_count"] == 0 and case["completed_subscriptions"]["duplicate_count"] == 0 + and case["info_log_count"] == 2 + and case["info_log_bounded"] is True for case in result["match_execution_cases"] ) assert [ diff --git a/tests/test_subscription_match_admission.py b/tests/test_subscription_match_admission.py index 099898a79..2b935c50c 100644 --- a/tests/test_subscription_match_admission.py +++ b/tests/test_subscription_match_admission.py @@ -229,3 +229,99 @@ def test_match_stop_does_not_report_unvisited_subscriptions_as_completed(monkeyp "failed": 0, }, } + + +def test_match_logs_one_bounded_start_and_finish_summary(monkeypatch) -> None: + """Match INFO 只保留轮次摘要，并明确任务完成不等于订阅成功。""" + completed = _subscribe(13) + conflicted = _subscribe(14) + cancelled = _subscribe(15) + chain = _chain([completed, conflicted, cancelled]) + search_lease = chain._subscription_execution_admission.try_acquire( + subscription_id=conflicted.id, + operation="search", + ttl_seconds=60, + ) + assert search_lease is not None + + def process(**kwargs): + """让一条正常处理完成，另一条在安全边界取消。""" + if kwargs["subscribe"].id == cancelled.id: + raise subscribe_match.SubscriptionSearchCancelled("cancelled") + return "completed" + + chain._match_subscription = process + info_logs = [] + monkeypatch.setattr(subscribe_match.logger, "info", info_logs.append) + + chain.match({"site-a.example": [object()], "site-b.example": [object(), object()]}) + + assert len(info_logs) == 2 + assert info_logs[0].startswith("订阅治理轮次开始: operation=match ") + assert "subscriptions=3" in info_logs[0] + assert "sites=2" in info_logs[0] + assert "candidates=3" in info_logs[0] + assert info_logs[1].startswith("订阅治理轮次结束: operation=match ") + assert "state=skipped" in info_logs[1] + assert "task_completed=1" in info_logs[1] + assert "task_skipped=2" in info_logs[1] + assert "admission_conflicts=1" in info_logs[1] + assert "cancelled=1" in info_logs[1] + assert "ttl_timeouts=0" in info_logs[1] + assert "订阅成功" not in "\n".join(info_logs) + assert chain._subscription_execution_admission.release(search_lease) is True + + +def test_match_summary_distinguishes_ttl_timeout(monkeypatch) -> None: + """Match 安全边界因 TTL 停止时必须单独计数。""" + subscribe = _subscribe(16) + chain = _chain([subscribe]) + now = [0.0] + chain._subscription_execution_admission = SubscriptionExecutionAdmission( + clock=lambda: now[0] + ) + + def expire(**_kwargs): + """让当前 owner 在单订阅执行返回前超过 TTL。""" + now[0] = chain._SUBSCRIPTION_EXECUTION_TTL + 1 + return "skipped" + + chain._match_subscription = expire + info_logs = [] + monkeypatch.setattr(subscribe_match.logger, "info", info_logs.append) + + chain.match({"site.example": [object()]}) + + assert len(info_logs) == 2 + assert "task_skipped=1" in info_logs[1] + assert "cancelled=0" in info_logs[1] + assert "ttl_timeouts=1" in info_logs[1] + + +def test_match_logs_subscription_admission_release_failure(monkeypatch) -> None: + """订阅 owner token 无法释放时必须同时进入错误日志和轮次摘要。""" + class _ReleaseRejectingAdmission(SubscriptionExecutionAdmission): + """模拟当前 owner 已被替换的释放冲突。""" + + def release(self, lease): + """拒绝释放以触发可见性合同。""" + del lease + return False + + subscribe = _subscribe(17) + chain = _chain([subscribe]) + chain._subscription_execution_admission = _ReleaseRejectingAdmission() + chain._match_subscription = lambda **_kwargs: "completed" + info_logs = [] + error_logs = [] + monkeypatch.setattr(subscribe_match.logger, "info", info_logs.append) + monkeypatch.setattr( + subscribe_match.logger, + "error", + lambda message, **_kwargs: error_logs.append(message), + ) + + chain.match({"site.example": [object()]}) + + assert any("订阅准入释放失败: operation=match subscription_id=17" in item for item in error_logs) + assert "release_failures=1" in info_logs[-1] diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index 52490840b..df4419c26 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -14,6 +14,7 @@ from app.application.subscription.contract import SubscriptionSnapshot from app.application.subscription.execution import SubscriptionExecutionAdmission from app.application.subscription.sitebudget import SubscriptionSearchCancelled from app.chain.search.facade import SearchChain +from app.chain.subscribe import search as subscribe_search from app.chain.subscribe.facade import SubscribeChain from app.chain.subscribe.search import _search_task_available_at from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository @@ -128,7 +129,6 @@ def test_inline_fallback_search_preserves_site_pressure_stagger(tmp_path, monkey "_process_search_subscription", lambda item, _searchchain, **_kwargs: item, ) - with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): chain.search(state="R") @@ -166,12 +166,19 @@ def test_successful_sites_remain_available_to_next_due_subscription(tmp_path, mo searchchain.search_site_torrents = search_site_torrents monkeypatch.setattr(chain, "_process_search_subscription", process) + info_logs = [] + monkeypatch.setattr(subscribe_search.logger, "info", info_logs.append) with patch("app.chain.subscribe.search.SearchChain", return_value=searchchain): batch_id = chain.search(state="R") assert calls == [(40, 11), (40, 12), (41, 11), (41, 12)] assert chain.get_search_batch(batch_id).state == "completed" + assert "sites=2" in info_logs[-1] + assert "site_requests=4" in info_logs[-1] + assert "site_failures=0" in info_logs[-1] + assert "site_cooldown_skips=0" in info_logs[-1] + assert "candidates=4" in info_logs[-1] def test_fallback_queue_executes_without_match_global_lock(tmp_path, monkeypatch): @@ -241,6 +248,69 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke assert chain._subscription_execution_admission.release(lease) is True +def test_search_logs_one_bounded_start_and_finish_summary(tmp_path, monkeypatch): + """Search INFO 只保留轮次摘要，并携带任务终态与耗时字段。""" + subscribes = [_subscribe(50), _subscribe(51)] + chain = _chain(tmp_path, subscribes) + _make_tasks_ready(monkeypatch) + + def process(subscribe, _searchchain, **_kwargs): + """让一条任务失败并保持下一条正常收口。""" + if subscribe.id == 50: + raise RuntimeError("provider timeout") + return subscribe + + monkeypatch.setattr(chain, "_process_search_subscription", process) + info_logs = [] + monkeypatch.setattr(subscribe_search.logger, "info", info_logs.append) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(state="R") + + assert batch_id + assert len(info_logs) == 2 + assert info_logs[0].startswith("订阅治理轮次开始: operation=search ") + assert f"batch_id={batch_id}" in info_logs[0] + assert "subscriptions=2" in info_logs[0] + assert info_logs[1].startswith("订阅治理轮次结束: operation=search ") + assert "state=failed" in info_logs[1] + assert "processed=2" in info_logs[1] + assert "task_completed=1" in info_logs[1] + assert "task_failed=1" in info_logs[1] + assert "admission_conflicts=0" in info_logs[1] + assert "ttl_timeouts=0" in info_logs[1] + assert "site_requests=0" in info_logs[1] + assert "duration_ms=" in info_logs[1] + assert "订阅成功" not in "\n".join(info_logs) + + +def test_search_logs_subscription_admission_release_failure(tmp_path, monkeypatch): + """Search 无法释放订阅 owner 时必须即时告警并写入轮次摘要。""" + subscribe = _subscribe(52) + chain = _chain(tmp_path, [subscribe]) + _make_tasks_ready(monkeypatch) + monkeypatch.setattr( + chain, + "_process_search_subscription", + lambda item, _searchchain, **_kwargs: item, + ) + monkeypatch.setattr(chain._subscription_execution_admission, "release", lambda _lease: False) + info_logs = [] + error_logs = [] + monkeypatch.setattr(subscribe_search.logger, "info", info_logs.append) + monkeypatch.setattr( + subscribe_search.logger, + "error", + lambda message, **_kwargs: error_logs.append(message), + ) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + chain.search(state="R") + + assert any("订阅准入释放失败: operation=search subscription_id=52" in item for item in error_logs) + assert "release_failures=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)] @@ -298,6 +368,8 @@ def test_swallowed_indexer_failure_marks_task_and_batch_failed_but_continues(tmp return subscribe monkeypatch.setattr(chain, "_process_search_subscription", process) + info_logs = [] + monkeypatch.setattr(subscribe_search.logger, "info", info_logs.append) with patch("app.chain.subscribe.search.SearchChain", return_value=searchchain): batch_id = chain.search(state="R") @@ -309,6 +381,10 @@ def test_swallowed_indexer_failure_marks_task_and_batch_failed_but_continues(tmp assert batch.failed_count == 1 assert "Flaky" in batch.last_error assert "HTTP 429" in batch.last_error + assert "sites=2" in info_logs[-1] + assert "site_requests=2" in info_logs[-1] + assert "site_failures=1" in info_logs[-1] + assert "cooldown_seconds=900.0" in info_logs[-1] def test_same_subscription_conflict_is_skipped_without_waiting(tmp_path, monkeypatch): @@ -324,6 +400,8 @@ def test_same_subscription_conflict_is_skipped_without_waiting(tmp_path, monkeyp ttl_seconds=60, ) assert match_lease is not None + info_logs = [] + monkeypatch.setattr(subscribe_search.logger, "info", info_logs.append) with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): batch_id = chain.search(state="R") @@ -334,6 +412,8 @@ def test_same_subscription_conflict_is_skipped_without_waiting(tmp_path, monkeyp assert batch.finished_count == 0 assert batch.skipped_count == 1 assert batch.last_error == "同一订阅正在由其他通道处理，本轮搜索已跳过" + assert "task_skipped=1" in info_logs[-1] + assert "admission_conflicts=1" in info_logs[-1] assert chain._subscription_execution_admission.release(match_lease) is True @@ -532,6 +612,8 @@ def test_ttl_expiry_after_normal_return_marks_task_and_batch_failed(tmp_path, mo "_process_search_subscription", lambda item, _searchchain, **_kwargs: item, ) + info_logs = [] + monkeypatch.setattr(subscribe_search.logger, "info", info_logs.append) with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): batch_id = chain.search(state="R") @@ -542,6 +624,8 @@ def test_ttl_expiry_after_normal_return_marks_task_and_batch_failed(tmp_path, mo assert batch.failed_count == 1 assert batch.cancelled_count == 0 assert batch.last_error == "订阅执行已超过协作截止时间" + assert "task_failed=1" in info_logs[-1] + assert "ttl_timeouts=1" in info_logs[-1] def test_ttl_expiry_after_download_started_completes_with_actual_result(tmp_path, monkeypatch): diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index 0cdb0c2a3..f67ae8191 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -14,6 +14,7 @@ from app.application.subscription.sitebudget import ( SiteBudgetClaim, SubscriptionSearchCancelled, SubscriptionSiteBudget, + SubscriptionSiteBudgetMetrics, SubscriptionSiteBudgetUnavailable, ) from app.chain.search.facade import SearchChain @@ -244,11 +245,13 @@ def test_skipped_search_releases_budget_without_external_interval(): def test_search_provider_reports_cooled_site_without_blocking_other_results(): """错误冷却中的站点返回空页并记录聚合失败，而非阻塞 provider。""" repository = _WaitingRepository() + metrics = SubscriptionSiteBudgetMetrics() budget = SubscriptionSiteBudget( repository=repository, owner="fallback-worker", cancelled=lambda: False, stop_state=ProcessStopState(), + metrics=metrics, ) chain = object.__new__(SearchChain) chain.configure_subscription_site_budget(budget) @@ -265,6 +268,9 @@ def test_search_provider_reports_cooled_site_without_blocking_other_results(): failures = chain.consume_subscription_site_budget_failures() assert len(failures) == 1 assert "站点 11" in failures[0] + snapshot = metrics.snapshot() + assert snapshot.request_count == 0 + assert snapshot.cooldown_skip_count == 1 def test_search_provider_releases_successful_site_budget(): @@ -290,11 +296,13 @@ def test_search_provider_releases_successful_site_budget(): captured.update(kwargs) return True + metrics = SubscriptionSiteBudgetMetrics() budget = SubscriptionSiteBudget( repository=_Repository(), owner="manual-worker", cancelled=lambda: False, stop_state=ProcessStopState(), + metrics=metrics, ) chain = object.__new__(SearchChain) chain.configure_subscription_site_budget(budget) @@ -317,6 +325,12 @@ def test_search_provider_releases_successful_site_budget(): assert captured["site_id"] == 12 assert captured["outcome"] == "success" assert captured["lease_token"] == "lease-token" + snapshot = metrics.snapshot() + assert snapshot.site_count == 1 + assert snapshot.request_count == 1 + assert snapshot.candidate_count == 1 + assert snapshot.failure_count == 0 + assert snapshot.release_failure_count == 0 def test_search_provider_aggregates_swallowed_indexer_failure(monkeypatch): @@ -368,11 +382,13 @@ def test_search_provider_aggregates_swallowed_indexer_failure(monkeypatch): staticmethod(lambda **_kwargs: []), ) + metrics = SubscriptionSiteBudgetMetrics() budget = SubscriptionSiteBudget( repository=_Repository(), owner="fallback-worker", cancelled=lambda: False, stop_state=ProcessStopState(), + metrics=metrics, ) chain = object.__new__(SearchChain) chain.configure_subscription_site_budget(budget) @@ -392,3 +408,53 @@ def test_search_provider_aggregates_swallowed_indexer_failure(monkeypatch): assert len(failures) == 1 assert "Flaky" in failures[0] assert "HTTP 429" in failures[0] + snapshot = metrics.snapshot() + assert snapshot.request_count == 1 + assert snapshot.failure_count == 1 + assert snapshot.cooldown_seconds == 900.0 + + +def test_search_provider_logs_site_budget_release_failure(monkeypatch): + """站点租约收口返回失败时必须进入摘要并即时记录错误。""" + class _Repository(_WaitingRepository): + """提供租约但拒绝以当前 token 收口。""" + + 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="stale-token", + ) + + def finish_site(self, **_kwargs) -> bool: + """模拟 token 已失效的 CAS 收口。""" + return False + + metrics = SubscriptionSiteBudgetMetrics() + budget = SubscriptionSiteBudget( + repository=_Repository(), + owner="fallback-worker", + cancelled=lambda: False, + stop_state=ProcessStopState(), + metrics=metrics, + ) + chain = object.__new__(SearchChain) + chain.configure_subscription_site_budget(budget) + chain.search_site_torrents = lambda **_kwargs: ["torrent"] + errors = [] + monkeypatch.setattr("app.chain.search.provider.logger.error", lambda message, **_kwargs: errors.append(message)) + + result = chain._search_site_torrents_with_budget( # pylint: disable=protected-access + site={"id": 14, "name": "Stale"}, + keyword="movie", + mtype=None, + page=0, + ) + + assert result == ["torrent"] + assert metrics.snapshot().release_failure_count == 1 + assert errors == ["订阅站点预算释放失败: site_id=14 site=Stale"]