nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-10-31 20:34:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(subscription): add governance logging summaries

Browse Source
This commit is contained in:
InfinityPacer
2026-09-03 11:55:51 +08:00
parent 68880dbbf1
commit af47c1abee
10 changed files with 729 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
app/application/subscription/sitebudget.py
View File

@@ -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:
"""正常或本地跳过立即恢复，错误按连续失败次数退避。"""

27
app/chain/search/provider.py
View File

@@ -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",

126
app/chain/subscribe/match.py
View File

@@ -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,

211
app/chain/subscribe/search.py
View File

@@ -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

45
docs/refactor/moviepilot-subscription-governance-roadmap.md
View File

@@ -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。

36
scripts/validation/subscription_governance_scale.py
View File

@@ -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": {

5
tests/test_subscription_governance_scale.py
View File

@@ -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 [

96
tests/test_subscription_match_admission.py
View File

@@ -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]

86
tests/test_subscription_search_governance.py
View File

@@ -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):

66
tests/test_subscription_site_budget.py
View File

@@ -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"]