diff --git a/app/runtime/log.py b/app/runtime/log.py index 6de2f1371..a6349362c 100644 --- a/app/runtime/log.py +++ b/app/runtime/log.py @@ -8,16 +8,17 @@ import sys import threading import time from collections import deque -from concurrent.futures import ThreadPoolExecutor from datetime import datetime from logging.handlers import RotatingFileHandler from pathlib import Path -from typing import Any, Callable, Dict, Optional, Protocol +from types import FrameType +from typing import Any, Callable, Dict, Optional, Protocol, Self import click from pydantic import BaseModel, ConfigDict -class LogConfigModel(BaseModel): +# strict mypy 跳过第三方实现导入，因此无法在本文件解析 Pydantic 元类类型。 +class LogConfigModel(BaseModel): # type: ignore[misc] """描述日志级别、格式和文件写入策略。""" model_config = ConfigDict(extra="ignore") @@ -34,6 +35,7 @@ class LogConfigModel(BaseModel): "【%(levelname)s】%(asctime)s [%(correlation_id)s] - %(message)s" ) ASYNC_FILE_QUEUE_SIZE: int = 1000 + # 保留历史配置解析兼容；协程环境文件日志已统一由单一有界队列 writer 执行。 ASYNC_FILE_WORKERS: int = 2 BATCH_WRITE_SIZE: int = 50 WRITE_TIMEOUT: float = 3.0 @@ -72,12 +74,13 @@ class LogWriter(Protocol): def write_log(self, level: str, message: str, file_path: Path) -> None: """将一条日志写入指定文件。""" - def shutdown(self) -> None: - """排空待写日志并释放写入资源。""" + def shutdown(self) -> Optional[bool]: + """排空待写日志并释放写入资源，未收敛时返回 False。""" log_settings = LogSettings() _correlation_id_provider: Callable[[], str | None] = lambda: None +_LOG_STOP_TIMEOUT_SECONDS = 10.0 def configure_correlation_id_provider(provider: Callable[[], str | None]) -> None: @@ -96,9 +99,9 @@ class NonBlockingFileHandler: _instance = None _lock = threading.Lock() - _stop_sentinel = object() + _stop_sentinel = None - def __new__(cls): + def __new__(cls) -> Self: """返回进程内唯一的文件写入器。""" if cls._instance is None: with cls._lock: @@ -114,12 +117,14 @@ class NonBlockingFileHandler: self._state_lock = threading.RLock() self._handlers_lock = threading.Lock() self._rotating_handlers: dict[Path, RotatingFileHandler] = {} - self._write_queue = queue.Queue(maxsize=log_settings.ASYNC_FILE_QUEUE_SIZE) - self._executor = ThreadPoolExecutor( - max_workers=log_settings.ASYNC_FILE_WORKERS, - thread_name_prefix="LogWriter", + self._write_queue: queue.Queue[Optional[LogEntry]] = queue.Queue( + maxsize=log_settings.ASYNC_FILE_QUEUE_SIZE, ) + self._stop_requested = threading.Event() self._running = True + self._closed = False + self._close_thread: Optional[threading.Thread] = None + self._close_error: Optional[BaseException] = None self._write_thread = threading.Thread( target=self._batch_writer, daemon=True, @@ -170,7 +175,8 @@ class NonBlockingFileHandler: try: self._write_queue.put_nowait(entry) except queue.Full: - self._executor.submit(self._write_sync, entry) + # 文件日志属于 E1 观测数据；队列达到显式上限时不能再创建无界线程池旁路。 + return False return True def _write_sync(self, entry: LogEntry) -> None: @@ -193,8 +199,10 @@ class NonBlockingFileHandler: msg=entry.message, args=(), exc_info=None, - created=entry.timestamp.timestamp(), ) + created_at = entry.timestamp.timestamp() + record.created = created_at + record.msecs = (created_at - int(created_at)) * 1000 record.correlation_id = entry.correlation_id return record @@ -202,15 +210,21 @@ class NonBlockingFileHandler: """持续收集队列日志，并在停止哨兵后排空已有批次。""" while True: try: - batch = [] + batch: list[LogEntry] = [] should_stop = False - end_time = time.time() + log_settings.WRITE_TIMEOUT + end_time = time.monotonic() + log_settings.WRITE_TIMEOUT while ( len(batch) < log_settings.BATCH_WRITE_SIZE - and time.time() < end_time + and time.monotonic() < end_time ): try: - remaining_time = max(0, end_time - time.time()) + if ( + self._stop_requested.is_set() + and self._write_queue.empty() + ): + should_stop = True + break + remaining_time = max(0, end_time - time.monotonic()) entry = self._write_queue.get(timeout=remaining_time) if entry is self._stop_sentinel: should_stop = True @@ -222,6 +236,8 @@ class NonBlockingFileHandler: self._write_batch(batch) if should_stop: break + if self._stop_requested.is_set() and self._write_queue.empty(): + break except Exception as err: print(f"批量写入线程错误: {err}") time.sleep(0.1) @@ -241,23 +257,86 @@ class NonBlockingFileHandler: for entry in entries: self._write_sync(entry) - def shutdown(self) -> None: - """停止接收新日志，排空队列并关闭线程池和文件处理器。""" + def _close_handlers(self) -> None: + """在独立 owner 中关闭文件处理器，保留失败项供后续重试。""" + first_error: Optional[BaseException] = None + with self._handlers_lock: + handlers = tuple(self._rotating_handlers.items()) + for file_path, handler in handlers: + try: + handler.flush() + handler.close() + except BaseException as err: # noqa: BLE001 需要保留关闭失败 owner + if first_error is None: + first_error = err + print(f"日志处理器关闭失败 {file_path}: {err}") + continue + with self._handlers_lock: + if self._rotating_handlers.get(file_path) is handler: + self._rotating_handlers.pop(file_path, None) with self._state_lock: - if not self._running: - return - self._running = False - if self._write_thread.is_alive(): - self._write_queue.put(self._stop_sentinel) - if self._write_thread.is_alive(): - self._write_thread.join() - self._executor.shutdown(wait=True) - for handler in self._rotating_handlers.values(): - handler.flush() - handler.close() - self._rotating_handlers.clear() + self._close_error = first_error -_LEVEL_NAME_COLORS = { + def _close_handlers_bounded(self, deadline: float) -> bool: + """复用关停总预算有限等待文件处理器关闭 owner。""" + with self._state_lock: + if self._closed: + return True + close_thread = self._close_thread + if close_thread is None: + self._close_error = None + close_thread = threading.Thread( + target=self._close_handlers, + daemon=True, + name="LogHandlerCloser", + ) + self._close_thread = close_thread + close_thread.start() + if close_thread is threading.current_thread(): + return False + close_thread.join(timeout=max(0.0, deadline - time.monotonic())) + if close_thread.is_alive(): + return False + with self._state_lock: + if self._close_error is not None: + if self._close_thread is close_thread: + self._close_thread = None + return False + self._closed = True + return True + + def shutdown( + self, + timeout: float = _LOG_STOP_TIMEOUT_SECONDS, + ) -> bool: + """ + 停止接收新日志，并在总预算内排空队列和关闭文件处理器。 + + :param timeout: 等待写线程和文件处理器收敛的最长秒数 + :return: 全部日志资源真实终止时返回 True，否则返回 False + """ + deadline = time.monotonic() + max(0.0, timeout) + with self._state_lock: + if self._closed: + return True + if self._running: + self._running = False + self._stop_requested.set() + if self._write_thread.is_alive(): + try: + self._write_queue.put_nowait(self._stop_sentinel) + except queue.Full: + # 队列非空会自然唤醒 writer；停止事件让其排空后退出。 + pass + if self._write_thread is threading.current_thread(): + return False + self._write_thread.join(timeout=max(0.0, deadline - time.monotonic())) + if self._write_thread.is_alive(): + return False + return self._close_handlers_bounded(deadline) + + +_LEVEL_NAME_COLORS: dict[int, Callable[[str], str]] = { logging.DEBUG: lambda level_name: click.style(str(level_name), fg="cyan"), logging.INFO: lambda level_name: click.style(str(level_name), fg="green"), logging.WARNING: lambda level_name: click.style(str(level_name), fg="yellow"), @@ -310,6 +389,7 @@ class LoggerManager: """ caller_name = None plugin_name = None + frame: Optional[FrameType] try: frame = sys._getframe(3) # noqa: SLF001 except (AttributeError, ValueError): @@ -368,12 +448,16 @@ class LoggerManager: """装配文件写入器，并补写装配前暂存的启动日志。""" with cls._lock: previous_writer = cls._writer + if previous_writer and previous_writer is not writer: + if previous_writer.shutdown() is False: + raise RuntimeError("既有日志写入器未收敛，拒绝丢失其资源 owner") + with cls._lock: + if cls._writer is not previous_writer: + raise RuntimeError("日志写入器在装配期间被并发替换") cls._writer = writer cls._log_path = Path(log_path) pending = list(cls._pending_file_logs) cls._pending_file_logs.clear() - if previous_writer and previous_writer is not writer: - previous_writer.shutdown() for level, message, logfile in pending: writer.write_log(level, message, Path(log_path) / logfile) @@ -456,14 +540,19 @@ class LoggerManager: self.logger("critical", msg, *args, **kwargs) @classmethod - def shutdown(cls) -> None: - """断开并关闭当前文件写入器。""" + def shutdown(cls) -> bool: + """关闭当前文件写入器，未收敛时保留 owner 供后续重试。""" with cls._lock: writer = cls._writer - cls._writer = None - cls._log_path = None - if writer: - writer.shutdown() + if writer is None: + return True + if writer.shutdown() is False: + return False + with cls._lock: + if cls._writer is writer: + cls._writer = None + cls._log_path = None + return True logger = LoggerManager() diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index 724f20eed..00796c9cc 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -582,7 +582,8 @@ async def lifespan(app: FastAPI): finally: try: # 日志最后关闭，确保其他组件的收尾信息已写入文件 - LoggerManager.shutdown() + if LoggerManager.shutdown() is False: + raise RuntimeError("日志写入资源未在关停预算内收敛") finally: if main_loop_owner is not None: global_vars.clear_loop(main_loop_owner) diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 20dc25d57..9d821029b 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -2,13 +2,13 @@ > 文档性质：当前架构复核、优秀 Python 后端实践对标、AI 可执行任务手册 > 适用仓库：`MoviePilot`，分支 `v3` -> 审计基线：`ad45bfac`（2026-08-24） +> 审计基线：`88262191`（2026-08-24） > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口；阶段 21 已收口监控历史端口；阶段 22 已统一服务配置应用边界；阶段 23 已补齐媒体服务器 API 遗留的类形配置读取路径；阶段 24 已清除 Scheduler 内部无 owner 的协程提交双轨；阶段 25 已补齐 TaskRegistry 跨线程 owner 并迁移整理 AI 接管；阶段 26 已统一 Agent 会话清理提交；阶段 27 已统一历史 AI 进度 owner；阶段 28 已托管旧插件订阅统计线程；阶段 29 已统一 Emby 系条目转换并清零重复代码白名单；阶段 30 已收口插件市场请求级子任务；阶段 31 已托管搜索 AI 推荐任务；阶段 32 已清除事件调度器绕过生命周期 owner 的投递回退；阶段 33 已统一宿主 Agent 运行时的获取路径；阶段 34 已统一 durable-required 事件与 Outbox topic 事实源；阶段 35 已统一 LLM provider 管理 API 的运行时解析路径；阶段 36 已统一 WebAgent 音频能力访问边界；阶段 37 已统一插件输入事件发布路径；阶段 38 已统一 WebAgent 通知事件监听与队列边界；阶段 39 已补齐搜索 SSE 断线时的上游任务清理；阶段 40 已补齐异步防抖取消的终态所有权；阶段 41 已统一优雅重启兜底线程的唯一所有权；阶段 42 已补齐 Telegram typing 的多实例隔离和终态 owner；阶段 43 已统一 Discord typing 的异步 owner 和 shutdown 收尾；阶段 44 已清除 WebAgent 测试临时事件循环提前关闭产生的 CI 红注解；阶段 45 已统一影视与字幕搜索的请求级逐页任务编排；阶段 46 已收口启动性能门禁的托管 runner 假失败与诊断输出；阶段 47 已补齐 Agent 渠道流式刷新任务的重入 owner；阶段 48 已统一工件上传 action 的 Node 24 主版本；阶段 49 已统一插件安装的同步/异步代际解析事实源；阶段 50 已统一插件市场 GitHub 请求降级策略；阶段 51 已统一插件索引请求与响应三态策略；阶段 52 已统一插件 Release 分页策略；阶段 53 已统一远端插件安装模式决策；阶段 54 已补齐同步安装成功后的临时回滚备份清理；阶段 55～56 已收口官方插件观察基线与报告保留策略；阶段 57 已统一进程级运行时 Facade 门禁并补齐 ModuleManager 边界；阶段 58 已消除 AgentTask 关闭回归的跨线程零时长等待竞态；阶段 59 已统一 Feishu 多实例长连接的 SDK 循环路由；阶段 60 已清除命令服务虚假的关停 owner 声明；阶段 61 已统一 Capability Runtime 同步/异步关闭的诚实收敛结果；阶段 62 已统一消息渠道长连接的多实例关闭收敛合同；阶段 63 已补齐应用消息队列线程的关闭收敛合同；阶段 64 已统一共享线程池的有界关闭 owner。 +> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口；阶段 21 已收口监控历史端口；阶段 22 已统一服务配置应用边界；阶段 23 已补齐媒体服务器 API 遗留的类形配置读取路径；阶段 24 已清除 Scheduler 内部无 owner 的协程提交双轨；阶段 25 已补齐 TaskRegistry 跨线程 owner 并迁移整理 AI 接管；阶段 26 已统一 Agent 会话清理提交；阶段 27 已统一历史 AI 进度 owner；阶段 28 已托管旧插件订阅统计线程；阶段 29 已统一 Emby 系条目转换并清零重复代码白名单；阶段 30 已收口插件市场请求级子任务；阶段 31 已托管搜索 AI 推荐任务；阶段 32 已清除事件调度器绕过生命周期 owner 的投递回退；阶段 33 已统一宿主 Agent 运行时的获取路径；阶段 34 已统一 durable-required 事件与 Outbox topic 事实源；阶段 35 已统一 LLM provider 管理 API 的运行时解析路径；阶段 36 已统一 WebAgent 音频能力访问边界；阶段 37 已统一插件输入事件发布路径；阶段 38 已统一 WebAgent 通知事件监听与队列边界；阶段 39 已补齐搜索 SSE 断线时的上游任务清理；阶段 40 已补齐异步防抖取消的终态所有权；阶段 41 已统一优雅重启兜底线程的唯一所有权；阶段 42 已补齐 Telegram typing 的多实例隔离和终态 owner；阶段 43 已统一 Discord typing 的异步 owner 和 shutdown 收尾；阶段 44 已清除 WebAgent 测试临时事件循环提前关闭产生的 CI 红注解；阶段 45 已统一影视与字幕搜索的请求级逐页任务编排；阶段 46 已收口启动性能门禁的托管 runner 假失败与诊断输出；阶段 47 已补齐 Agent 渠道流式刷新任务的重入 owner；阶段 48 已统一工件上传 action 的 Node 24 主版本；阶段 49 已统一插件安装的同步/异步代际解析事实源；阶段 50 已统一插件市场 GitHub 请求降级策略；阶段 51 已统一插件索引请求与响应三态策略；阶段 52 已统一插件 Release 分页策略；阶段 53 已统一远端插件安装模式决策；阶段 54 已补齐同步安装成功后的临时回滚备份清理；阶段 55～56 已收口官方插件观察基线与报告保留策略；阶段 57 已统一进程级运行时 Facade 门禁并补齐 ModuleManager 边界；阶段 58 已消除 AgentTask 关闭回归的跨线程零时长等待竞态；阶段 59 已统一 Feishu 多实例长连接的 SDK 循环路由；阶段 60 已清除命令服务虚假的关停 owner 声明；阶段 61 已统一 Capability Runtime 同步/异步关闭的诚实收敛结果；阶段 62 已统一消息渠道长连接的多实例关闭收敛合同；阶段 63 已补齐应用消息队列线程的关闭收敛合同；阶段 64 已统一共享线程池的有界关闭 owner；阶段 65 已统一异步文件日志的单一有界写入和关闭 owner。 > 当前 canonical 状态：API/Application 公共复杂度基线已清零，组合根外 `SystemConfigOper()` 构造和 Model/Oper 隐式事务均为 0；命名 Chain/Agent 数据端口、TaskRegistry owner、Module Contract V2、typed Event、Outbox durable intent、请求关联和插件运行时 getter 已形成当前路径。插件仓适配、未知第三方 fallback 和其它 E1/E3 副作用仍按风险持续治理。 -> 最新阶段：阶段 64 已统一共享线程池的有界关闭 owner。 +> 最新阶段：阶段 65 已统一异步文件日志的单一有界写入和关闭 owner。 ## 当前复核结论（2026-08-24） @@ -687,6 +687,26 @@ 故障注入；线程池/生命周期专项 78 项、架构与兼容调用链 164 项、Pylint 10.00/10、strict mypy 40 文件、宿主与质量 ratchet 及四分片全量 `5946 passed, 3 skipped` 均通过。 +### 长期整改阶段 65：异步文件日志单一有界写入与关闭 owner 统一（2026-08-24） + +- 协程环境的日志运行时原先同时保留批量队列 writer 和“队列满后提交独立 `ThreadPoolExecutor` 直写”两条路径； + 配置的队列容量并不是真实上限，溢出任务会进入 executor 的无界队列，写入顺序、资源 owner 和关闭 + 语义也分裂。当前异步文件日志只由一个有界队列 writer 执行；达到容量时拒绝新增文件副本，既有 + 控制台输出保持不变；无事件循环的同步调用仍保留直接写入语义，不再以第二套线程池掩盖 E1 过载。 +- 原 `shutdown()` 会无界等待批量线程、executor 和文件处理器；任一文件系统写入或 `close()` 阻塞都会 + 卡住 lifespan 所在事件循环。当前默认使用 10 秒共享 deadline，先封口新日志并排空已接受队列，再有限 + 等待 writer 和一次性文件处理器关闭 owner；超时返回 `False`，阻塞 owner 与 handler 均保留，释放后 + 可由同一实例重试到真实终态。 +- `LoggerManager` 不再先清空 writer 引用再关闭；未收敛时保留原 writer，重新装配也拒绝覆盖活动 owner。 + 日志仍在全部宿主组件之后最后关闭，但 `False` 会让 lifespan 以关闭失败结束，测试会话收尾也会报告 + 同一结果，不再把记录清理动作当成资源已经终止。 +- `NonBlockingFileHandler`、`LoggerManager`、无参数 `shutdown()`、历史 `ASYNC_FILE_WORKERS` 配置解析、 + `configure_log_writer()` 及 `app.log` 精确兼容映射均保留；新增 timeout 和布尔结果为向后兼容扩展， + 未修改 SDK/Compat 清单、插件 Hook 或插件仓，V1/V2/V3 插件观察面不变。 +- writer/handler 阻塞、队列过载、重试终态、重新装配拒绝覆盖和 lifespan 失败传播均有故障注入； + 日志/生命周期/旧导入专项 83 项、架构合同 94 项、兼容调用链 102 项、Pylint 10.00/10、strict mypy + 41 文件、全部宿主与质量 ratchet 及四分片全量 `5952 passed, 3 skipped` 均通过。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： @@ -757,7 +777,7 @@ ### P2：中长期可演进性债务 - **大型职责域仍偏重。** 代表性热点包括 `app/chain/subscribe.py`（约 `4141` 行）、`app/chain/transfer.py`（约 `2944` 行）、`app/agent/orchestrator.py`（约 `3540` 行）、`app/agent/llm/provider.py`（约 `3529` 行）、`app/adapters/external/market.py`（约 `3139` 行）和 `app/api/endpoints/agent.py`（约 `2489` 行）。复杂度 ratchet 只保证不超过当前基线，不代表这些文件已经易维护。只有在行为快照、调用命中和事务边界明确后，才值得按用例拆分。 -- **类型门禁覆盖面不足。** `mypy.ini` strict 文件清单目前为 `39` 个文件，Agent、Chain、Module、Adapter 大量代码仍依赖动态类型。应从模块契约、生命周期、Repository/Port 和关键 Chain 返回值开始扩展，而不是直接开启全仓 strict。 +- **类型门禁覆盖面不足。** `mypy.ini` strict 文件清单目前为 `41` 个文件，Agent、Chain、Module、Adapter 大量代码仍依赖动态类型。应从模块契约、生命周期、Repository/Port 和关键 Chain 返回值开始扩展，而不是直接开启全仓 strict。 - **Pylint 仍是增量硬门禁。** `.github/workflows/pylint.yml` 对改动 Python 文件执行硬检查，但全仓报告使用 `|| true` 仅作 advisory。该策略适合存量迁移，却没有形成全仓质量趋势约束；应增加按目录和新增问题数的 ratchet。 - **测试风格存在历史混用。** 当前有 `527` 个测试文件，仍有 `70` 个 `unittest.TestCase` 文件。它不是生产架构缺陷，但会增加 fixture、状态隔离和异步测试迁移成本，应在触碰相关模块时渐进迁移。 - **跨仓治理链路尚未完全闭环。** 前端已有 lint、typecheck、分片 Vitest 和构建门禁；插件仓有 V1/V2/V3 索引及版本/依赖检查；资源和 Rust 仓有独立构建发布链路。但插件 CI 本地复核因插件仓环境缺少主仓依赖 `httpx2` 无法完成收集，说明“插件仓测试环境与主仓锁定依赖”的可复现性仍需加强。资源构建通过 PR 同步到 `MoviePilot-Resources`，Rust 发布后自动向主仓发依赖 bump PR，链路合理但仍是多仓异步发布，需保留版本 provenance 和回滚点。 @@ -810,7 +830,7 @@ MoviePilot V3 当前不是“目录混乱、必须推倒重来”的状态。第 5. **模块与事件契约登记均已完成。**当前 212 个模块 spec 的宿主观察面已无 legacy aggregation，53 个事件全部绑定 typed payload，可见性、投递等级、错误行为和敏感字段均有基线，legacy event payload 为 `0`。六个 durable-required 事件的宿主正式生产者已通过业务同事务 Outbox 提供真实持久投递，事件与 topic 映射及恢复 handler 完整性已纳入 ratchet。后续重点是保持新增能力门禁和观察未知第三方 fallback 命中，不是重复创建契约、DTO 或 Outbox。 6. **后台副作用已有统一可靠性分类，但其他 E1/E3 机制仍需逐项收口。** ADR-0007 已分类事件队列、APScheduler、进程内任务、Agent task 与 transfer pending 的完成点、恢复和失败表达；不能因六个关键事件已接 Outbox，就把仍需定时重建、持久任务表或人工恢复的其他 E1/E3 机制误报为全部完成。 7. **核心关联与健康边界已落地，指标导出仍未收口。**HTTP/SSE correlation ID 已传播到线程池、事件、工作流、子进程、外部请求和日志；`/health/live`、`/health/ready` 已由部署入口消费，事件/数据库队列深度及模块/事件耗时使用低基数指标登记。当前缺口是稳定 exporter、运维查询面和跨进程聚合，而不是重新实现 request ID 或健康路由。 -8. **质量门禁已具备增量硬约束，但覆盖面仍需扩大。**push/PR 对变更 Python 文件执行 Pylint，CI 同时运行 host architecture、39 个 strict mypy 文件、复杂度、async 阻塞和 task owner ratchet；全仓 Pylint 仍是 advisory，strict 类型和复杂度拆分仍应随业务切片渐进扩展。 +8. **质量门禁已具备增量硬约束，但覆盖面仍需扩大。**push/PR 对变更 Python 文件执行 Pylint，CI 同时运行 host architecture、41 个 strict mypy 文件、复杂度、async 阻塞和 task owner ratchet；全仓 Pylint 仍是 advisory，strict 类型和复杂度拆分仍应随业务切片渐进扩展。 建议保持**模块化单体**，按以下顺序治理： diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 62737190e..ffa138da8 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -201,6 +201,9 @@ ModuleManager 与 startup 组合根继续关闭其余资源但必须向上返回 并向 startup 返回 `False`，不得用无界 `join()` 阻塞生命周期或把日志当作成功。 共享 `ThreadHelper` 必须追踪通过宿主 `submit()` 和旧兼容 `.pool.submit()` 接受的全部 Future；关闭时 先封口新任务，再有限等待且保留未终止 owner，结果由 startup 聚合，不得恢复无界 executor shutdown。 +协程环境文件日志属于有界 E1 观测能力，只允许单一队列 writer；队列满时不得再以无界 executor +形成第二条异步写入路径。日志关闭必须有限等待 writer 与文件处理器，未收敛时 `LoggerManager` +保留原 owner 并让 lifespan 以关闭失败结束，不得先清空引用或用无界 `join()` 掩盖失败。 API 中允许丢失或可重建的进程内任务必须登记到 `app/runtime/tasks.py`；登记器先于其他 运行资源启动，并在资源释放前停止接收、取消和有限等待。需要崩溃恢复的 E2/E3 副作用仍应 进入 Outbox 或持久任务表，不能把 TaskRegistry 当成 durable queue。 diff --git a/mypy.ini b/mypy.ini index e44180357..7ecbd4cf9 100644 --- a/mypy.ini +++ b/mypy.ini @@ -13,6 +13,7 @@ files = app/domain/meta/infopath.py, app/runtime/correlation.py, app/runtime/coalesce.py, + app/runtime/log.py, app/runtime/observability/__init__.py, app/runtime/thread.py, app/runtime/event/contracts.py, diff --git a/tests/conftest.py b/tests/conftest.py index 4ec119bcd..863601a89 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -493,6 +493,7 @@ def pytest_sessionfinish(session, exitstatus): try: from app.runtime.log import LoggerManager - LoggerManager.shutdown() + if LoggerManager.shutdown() is False: + raise RuntimeError("log writer did not converge") except Exception as err: _report_session_cleanup_error(session, "logger manager", err) diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 55d140714..7abab2fae 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -180,6 +180,26 @@ def test_lifespan_normal_mode_starts_full_runtime(monkeypatch): _assert_completed_once(step) +def test_lifespan_propagates_logger_nonconvergence(monkeypatch): + """最后一个日志 owner 未收敛时 lifespan 必须以关闭失败结束。""" + shutdown_steps = _patch_lifespan(monkeypatch) + shutdown_steps["logger"].return_value = False + + async def run_lifespan() -> None: + """运行完整生命周期并触发日志 writer 的诚实失败结果。""" + async with lifecycle.lifespan(FastAPI()): + pass + + with pytest.raises(RuntimeError, match="日志写入资源未在关停预算内收敛"): + asyncio.run(run_lifespan()) + + for step in shutdown_steps.values(): + _assert_completed_once(step) + lifecycle.global_vars.clear_loop.assert_called_once_with( + lifecycle.global_vars.set_loop.return_value + ) + + def test_lifespan_validation_failure_does_not_clear_outer_loop_owner(monkeypatch): """当前生命周期尚未取得 owner 时，启动失败不得清理外层登记。""" _patch_lifespan(monkeypatch) diff --git a/tests/test_log_shutdown.py b/tests/test_log_shutdown.py index cccaf83b4..3ce46a87a 100644 --- a/tests/test_log_shutdown.py +++ b/tests/test_log_shutdown.py @@ -2,7 +2,14 @@ import threading import time from unittest.mock import MagicMock -from app.runtime.log import LogEntry, NonBlockingFileHandler, log_settings +import pytest + +from app.runtime.log import ( + LogEntry, + LoggerManager, + NonBlockingFileHandler, + log_settings, +) def test_non_blocking_file_handler_shutdown_wakes_writer_and_closes_handlers(tmp_path): @@ -147,3 +154,163 @@ def test_non_blocking_file_handler_uses_handler_lock(monkeypatch, tmp_path): finally: handler.shutdown() NonBlockingFileHandler._instance = original_instance + + +def test_non_blocking_file_handler_shutdown_is_bounded_and_retryable( + monkeypatch, + tmp_path, +): + """批量写入阻塞时关闭必须有限返回，并保留同一 writer 供重试。""" + original_instance = NonBlockingFileHandler._instance + NonBlockingFileHandler._instance = None + monkeypatch.setattr(log_settings, "WRITE_TIMEOUT", 0.01) + handler = NonBlockingFileHandler() + entered = threading.Event() + release = threading.Event() + + def block_batch(_batch): + """模拟文件系统写入永久占用批量 writer。""" + entered.set() + release.wait() + + monkeypatch.setattr(handler, "_write_batch", block_batch) + try: + assert handler._write_non_blocking( + LogEntry("info", "blocked", tmp_path / "blocked.log") + ) is True + assert entered.wait(timeout=1) + + started_at = time.monotonic() + assert handler.shutdown(timeout=0.01) is False + assert time.monotonic() - started_at < 1 + assert handler._write_thread.is_alive() + assert handler._write_non_blocking( + LogEntry("info", "late", tmp_path / "blocked.log") + ) is False + release.set() + assert handler.shutdown(timeout=1) is True + assert not handler._write_thread.is_alive() + assert handler.shutdown(timeout=1) is True + finally: + release.set() + handler.shutdown(timeout=1) + NonBlockingFileHandler._instance = original_instance + + +def test_non_blocking_file_handler_does_not_bypass_full_queue( + monkeypatch, + tmp_path, +): + """队列达到显式容量后不得再通过无界线程池形成第二条写入路径。""" + original_instance = NonBlockingFileHandler._instance + NonBlockingFileHandler._instance = None + monkeypatch.setattr(log_settings, "ASYNC_FILE_QUEUE_SIZE", 1) + monkeypatch.setattr(log_settings, "BATCH_WRITE_SIZE", 1) + handler = NonBlockingFileHandler() + entered = threading.Event() + release = threading.Event() + written: list[str] = [] + + def block_batch(batch): + """占住唯一 writer，使后续日志稳定留在有界队列中。""" + written.extend(entry.message for entry in batch) + entered.set() + release.wait() + + monkeypatch.setattr(handler, "_write_batch", block_batch) + try: + assert handler._write_non_blocking( + LogEntry("info", "first", tmp_path / "bounded.log") + ) is True + assert entered.wait(timeout=1) + assert handler._write_non_blocking( + LogEntry("info", "queued", tmp_path / "bounded.log") + ) is True + assert handler._write_non_blocking( + LogEntry("info", "rejected", tmp_path / "bounded.log") + ) is False + release.set() + started_at = time.monotonic() + assert handler.shutdown(timeout=1) is True + assert time.monotonic() - started_at < 1 + assert written == ["first", "queued"] + finally: + release.set() + handler.shutdown(timeout=1) + NonBlockingFileHandler._instance = original_instance + + +def test_non_blocking_file_handler_bounds_handler_close(monkeypatch, tmp_path): + """文件处理器 close 阻塞时也必须保留关闭线程并支持最终重试。""" + original_instance = NonBlockingFileHandler._instance + NonBlockingFileHandler._instance = None + handler = NonBlockingFileHandler() + close_entered = threading.Event() + close_release = threading.Event() + log_handler = MagicMock() + + def block_close(): + """模拟文件系统在 flush 后阻塞关闭句柄。""" + close_entered.set() + close_release.wait() + + log_handler.close.side_effect = block_close + log_path = tmp_path / "close.log" + handler._rotating_handlers = {log_path: log_handler} + try: + assert handler.shutdown(timeout=0.05) is False + assert close_entered.wait(timeout=1) + assert handler._close_thread is not None + assert handler._close_thread.is_alive() + assert handler._rotating_handlers[log_path] is log_handler + close_release.set() + assert handler.shutdown(timeout=1) is True + assert handler._rotating_handlers == {} + finally: + close_release.set() + handler.shutdown(timeout=1) + NonBlockingFileHandler._instance = original_instance + + +def test_logger_manager_retains_nonconverged_writer_for_retry(tmp_path): + """平台日志门面不得在底层 writer 未收敛时丢失其 owner。""" + previous_writer = LoggerManager._writer + previous_log_path = LoggerManager._log_path + writer = MagicMock() + writer.shutdown.side_effect = [False, True] + LoggerManager._writer = writer + LoggerManager._log_path = tmp_path + try: + assert LoggerManager.shutdown() is False + assert LoggerManager._writer is writer + assert LoggerManager._log_path == tmp_path + + assert LoggerManager.shutdown() is True + assert LoggerManager._writer is None + assert LoggerManager._log_path is None + assert writer.shutdown.call_count == 2 + finally: + LoggerManager._writer = previous_writer + LoggerManager._log_path = previous_log_path + + +def test_logger_manager_refuses_to_replace_nonconverged_writer(tmp_path): + """重新装配不得用新 writer 覆盖仍持有资源的旧 owner。""" + original_writer = LoggerManager._writer + original_log_path = LoggerManager._log_path + previous_writer = MagicMock() + previous_writer.shutdown.return_value = False + replacement_writer = MagicMock() + old_path = tmp_path / "old" + LoggerManager._writer = previous_writer + LoggerManager._log_path = old_path + try: + with pytest.raises(RuntimeError, match="既有日志写入器未收敛"): + LoggerManager.configure_writer(replacement_writer, tmp_path / "new") + + assert LoggerManager._writer is previous_writer + assert LoggerManager._log_path == old_path + replacement_writer.write_log.assert_not_called() + finally: + LoggerManager._writer = original_writer + LoggerManager._log_path = original_log_path