diff --git a/app/adapters/external/market.py b/app/adapters/external/market.py index 3b5b61a79..81c3119cc 100644 --- a/app/adapters/external/market.py +++ b/app/adapters/external/market.py @@ -82,6 +82,19 @@ def _empty_installed_plugins() -> List[str]: _installed_plugins_provider: InstalledPluginsProvider = _empty_installed_plugins +async def _await_thread_operation(func, *args, **kwargs): + """取消请求到达时先等待同步插件操作收口，避免目录写入继续进行。""" + task = asyncio.create_task(asyncio.to_thread(func, *args, **kwargs)) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + try: + await asyncio.shield(task) + except BaseException: + pass + raise + + def configure_installed_plugins_provider( provider: InstalledPluginsProvider, ) -> None: @@ -177,6 +190,7 @@ class PluginHelper(metaclass=WeakSingleton): _base_url = "https://raw.githubusercontent.com/{user}/{repo}/main/" # 串行化运行期依赖安装，避免多个包安装子进程和导入缓存刷新互相踩踏。 _package_install_lock = threading.Lock() + PLUGIN_DEPENDENCY_INSTALL_TIMEOUT = 300 # 同仓库的并发 Release 请求共享任务；事件循环参与键控，避免热重载或测试循环切换后复用失效任务。 _release_task_lock = threading.Lock() _release_tasks: Dict[Tuple[asyncio.AbstractEventLoop, str, bool], asyncio.Task] = {} @@ -1472,6 +1486,42 @@ class PluginHelper(metaclass=WeakSingleton): temp_file.write(f"{cls.__format_package_name(package_name)}>={version}\n") return Path(temp_file.name) + @classmethod + async def __async_create_runtime_constraints_file( + cls, + protected_packages: Dict[str, Version], + ) -> Path: + """创建临时约束文件，取消时等待创建收口并删除已产生的文件。""" + create_task = asyncio.create_task( + asyncio.to_thread( + cls.__create_runtime_constraints_file, + protected_packages, + ) + ) + try: + return await asyncio.shield(create_task) + except asyncio.CancelledError: + async def cleanup_created_file() -> None: + try: + created_file = await create_task + except BaseException: + return + await asyncio.to_thread(created_file.unlink, missing_ok=True) + + cleanup_task = asyncio.create_task(cleanup_created_file()) + while not cleanup_task.done(): + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: + continue + except Exception: + break + try: + await cleanup_task + except Exception as err: + logger.warning(f"[UV] 取消后清理运行环境约束文件失败：{err}") + raise + @staticmethod def __refresh_import_system(): """ @@ -2421,18 +2471,284 @@ class PluginHelper(metaclass=WeakSingleton): return True, "" - async def __async_install_packages_with_fallback( - self, - dependency_file: Path, - find_links_dirs: Optional[List[Path]] = None) -> Tuple[bool, str]: - """ - 在线程池中执行插件依赖安装，避免同步包安装子进程阻塞事件循环。 - """ - return await asyncio.to_thread( - self.install_packages_with_fallback, - dependency_file, - find_links_dirs + @classmethod + async def __async_run_runtime_healthcheck(cls) -> Dict[str, Tuple[bool, str]]: + """异步执行插件安装后的运行环境检查。""" + health_snapshot: Dict[str, Tuple[bool, str]] = {} + uv_check = cls.__build_runtime_uv_command("check") + if uv_check: + checks = [("uv check", uv_check)] + else: + health_snapshot["uv check"] = (False, "未找到 uv 可执行文件") + checks = [] + checks.append(("核心依赖导入检查", [ + sys.executable, + "-c", + cls._runtime_import_probe, + ])) + for check_name, command in checks: + health_snapshot[check_name] = ( + await SystemUtils.execute_with_subprocess_async( + command, + timeout=30, + ) + ) + return health_snapshot + + @classmethod + async def __async_repair_main_runtime_dependencies( + cls, + snapshot_file: Optional[Path] = None, + ) -> Tuple[bool, str]: + """异步恢复主程序运行依赖，避免修复命令绕过可取消进程边界。""" + repair_target = snapshot_file + repair_desc = "主程序依赖快照" + if repair_target and not await _await_thread_operation(repair_target.exists): + repair_target = None + if repair_target is None: + repair_target = settings.ROOT_PATH / "pyproject.toml" + repair_desc = "主程序 uv.lock" + if not await _await_thread_operation(repair_target.exists): + return False, f"恢复依赖文件不存在：{repair_target}" + lock_file = settings.ROOT_PATH / "uv.lock" + if snapshot_file is None and not await _await_thread_operation(lock_file.exists): + return False, f"恢复依赖文件不存在：{settings.ROOT_PATH / 'uv.lock'}" + + request = cls.__build_package_install_request( + repair_target, + purpose="runtime-repair", ) + strategies = ( + build_package_install_strategies(request) + if snapshot_file is not None + else build_project_sync_strategies(request) + ) + last_error = "" + for strategy in strategies: + logger.warning( + f"[UV] 运行环境异常，尝试使用策略：{strategy.strategy_name} 恢复{repair_desc}" + ) + success, message = await SystemUtils.execute_with_subprocess_async( + strategy.command, + env=strategy.env, + safe_command=strategy.safe_log_command, + timeout=cls.PLUGIN_DEPENDENCY_INSTALL_TIMEOUT, + ) + if success: + cls.__refresh_import_system() + return True, message + last_error = message + logger.error( + f"[UV] 使用策略：{strategy.strategy_name} 恢复{repair_desc}失败：{message}" + ) + return False, last_error or f"恢复{repair_desc}失败" + + @classmethod + async def __async_repair_if_runtime_broken( + cls, + snapshot_file: Optional[Path], + baseline_health: Dict[str, Tuple[bool, str]], + ) -> Tuple[bool, str]: + """异步检查并修复安装过程中新增的主程序环境异常。""" + current_health = await cls.__async_run_runtime_healthcheck() + health_message = cls.__runtime_health_regression_message( + baseline_health, + current_health, + ) + if not health_message: + return True, "" + repair_ok, repair_message = ( + await cls.__async_repair_main_runtime_dependencies(snapshot_file) + ) + if not repair_ok: + return False, ( + f"插件依赖安装失败后主运行环境异常，且恢复失败：" + f"{health_message}; {repair_message}" + ) + restored_health = await cls.__async_run_runtime_healthcheck() + restored_message = cls.__runtime_health_regression_message( + baseline_health, + restored_health, + ) + if restored_message: + return False, ( + f"插件依赖安装失败后主运行环境异常，恢复后仍异常：" + f"{restored_message}" + ) + return True, "主运行环境已恢复" + + async def async_install_packages_with_fallback( + self, + dependency_files: Path | Sequence[Path], + find_links_dirs: Optional[List[Path]] = None, + ) -> Tuple[bool, str]: + """通过可取消子进程异步安装一组插件依赖清单。""" + return await self.__async_install_packages_with_fallback( + dependency_files, + find_links_dirs, + ) + + @classmethod + async def __async_install_packages_with_fallback( + cls, + dependency_files: Path | Sequence[Path], + find_links_dirs: Optional[List[Path]] = None, + ) -> Tuple[bool, str]: + """异步安装插件依赖，并让取消能够终止 uv 子进程。""" + if isinstance(dependency_files, Path): + resolved_dependency_files = (dependency_files,) + else: + resolved_dependency_files = tuple(Path(item) for item in dependency_files) + if not resolved_dependency_files: + return False, "没有传入插件依赖清单" + + candidate_dirs = [] + for dependency_file in resolved_dependency_files: + wheels_dir = dependency_file.parent / "wheels" + if await _await_thread_operation(wheels_dir.is_dir): + candidate_dirs.append(wheels_dir) + if find_links_dirs: + candidate_dirs.extend(find_links_dirs) + + resolved_dirs = [] + seen_dirs = set() + for candidate_dir in candidate_dirs: + candidate_path = Path(candidate_dir) + if not await _await_thread_operation(candidate_path.is_dir): + continue + candidate_key = str( + await _await_thread_operation(candidate_path.resolve) + ) + if candidate_key in seen_dirs: + continue + seen_dirs.add(candidate_key) + resolved_dirs.append(candidate_path) + + installed_packages = await _await_thread_operation( + cls.__get_installed_packages, + ) + protected_packages = await _await_thread_operation( + cls.__get_protected_runtime_packages, + installed_packages, + ) + for dependency_file in resolved_dependency_files: + check_ok, check_message = await _await_thread_operation( + cls.__validate_runtime_dependency_conflicts, + dependency_file, + protected_packages, + ) + if not check_ok: + logger.error(f"[UV] 运行环境冲突预检失败：{check_message}") + return False, check_message + + constraints_file = None + if protected_packages: + try: + constraints_file = await cls.__async_create_runtime_constraints_file( + protected_packages, + ) + except Exception as err: + logger.error(f"[UV] 创建运行环境约束文件失败：{err}") + return False, f"创建运行环境约束文件失败：{err}" + + request = cls.__build_package_install_request( + resolved_dependency_files, + find_links_dirs=resolved_dirs, + constraints_file=constraints_file, + purpose="plugin", + ) + strategies = build_package_install_strategies(request) + acquired = False + try: + while not cls._package_install_lock.acquire(blocking=False): + await asyncio.sleep(0.01) + acquired = True + baseline_health = await cls.__async_run_runtime_healthcheck() + baseline_health_message = cls.__runtime_health_regression_message( + {}, + baseline_health, + ) + if baseline_health_message: + logger.warning( + f"[UV] 安装前运行环境已存在异常，本次安装仅拦截新增异常：" + f"{baseline_health_message}" + ) + + last_error = "" + for strategy in strategies: + logger.debug( + f"[UV] 尝试使用策略：{strategy.strategy_name} 安装依赖，" + f"命令：{' '.join(strategy.safe_log_command)}" + ) + success, message = await SystemUtils.execute_with_subprocess_async( + strategy.command, + env=strategy.env, + safe_command=strategy.safe_log_command, + timeout=cls.PLUGIN_DEPENDENCY_INSTALL_TIMEOUT, + ) + if success: + current_health = await cls.__async_run_runtime_healthcheck() + health_message = cls.__runtime_health_regression_message( + baseline_health, + current_health, + ) + if health_message: + logger.error(f"[UV] 依赖安装后运行环境自检失败：{health_message}") + repair_ok, repair_message = ( + await cls.__async_repair_main_runtime_dependencies( + constraints_file if protected_packages else None + ) + ) + if repair_ok: + restored_health = await cls.__async_run_runtime_healthcheck() + restored_message = cls.__runtime_health_regression_message( + baseline_health, + restored_health, + ) + if not restored_message: + cls.__refresh_import_system() + return False, ( + f"依赖安装后运行环境自检失败，已自动恢复主程序依赖：" + f"{health_message}" + ) + return False, ( + f"依赖安装后运行环境自检失败，恢复主程序依赖后仍异常：" + f"{restored_message}" + ) + return False, ( + f"依赖安装后运行环境自检失败，且自动恢复主程序依赖失败：" + f"{repair_message}" + ) + + cls.__refresh_import_system() + return True, message + + last_error = message + repair_ok, repair_message = await cls.__async_repair_if_runtime_broken( + constraints_file if protected_packages else None, + baseline_health, + ) + logger.error( + f"[UV] 策略：{strategy.strategy_name} 安装依赖失败，错误信息：{message}" + ) + if not repair_ok or repair_message: + return False, ( + f"策略 {strategy.strategy_name} 安装依赖失败：{message}；" + f"{repair_message}" + ) + return False, ( + f"[UV] 所有策略均安装依赖失败：{last_error}" + if last_error + else "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束" + ) + finally: + if acquired: + cls._package_install_lock.release() + if constraints_file: + await _await_thread_operation( + constraints_file.unlink, + missing_ok=True, + ) async def __async_backup_plugin(self, pid: str) -> str: """ @@ -2444,14 +2760,16 @@ class PluginHelper(metaclass=WeakSingleton): backup_dir = AsyncPath(settings.TEMP_PATH) / "plugin_backup" / pid.lower() if await plugin_dir.exists(): - # 备份时清理已有的备份目录，防止残留文件影响 - if await backup_dir.exists(): - await aioshutil.rmtree(backup_dir, ignore_errors=True) - logger.debug(f"{pid} 旧的备份目录已清理 {backup_dir}") + try: + if await backup_dir.exists(): + await aioshutil.rmtree(backup_dir, ignore_errors=True) + logger.debug(f"{pid} 旧的备份目录已清理 {backup_dir}") - # 异步复制目录 - await self._async_copytree(plugin_dir, backup_dir) - logger.debug(f"{pid} 插件已备份到 {backup_dir}") + await self._async_copytree(plugin_dir, backup_dir) + logger.debug(f"{pid} 插件已备份到 {backup_dir}") + except asyncio.CancelledError: + await aioshutil.rmtree(backup_dir, ignore_errors=True) + raise return str(backup_dir) if await backup_dir.exists() else None @@ -2558,7 +2876,12 @@ class PluginHelper(metaclass=WeakSingleton): :return: (是否成功, 错误信息) """ if self.is_local_repo_url(repo_url): - return await asyncio.to_thread(self.install_local, pid, repo_url, force_install) + return await _await_thread_operation( + self.install_local, + pid, + repo_url, + force_install, + ) if SystemUtils.is_frozen(): return False, "可执行文件模式下，只能安装本地插件" @@ -2663,35 +2986,46 @@ class PluginHelper(metaclass=WeakSingleton): 异步安装流程，处理插件内容准备、依赖安装和注册 """ backup_dir = None - if not force_install: - backup_dir = await self.__async_backup_plugin(pid) + try: + if not force_install: + backup_dir = await self.__async_backup_plugin(pid) - await self.__async_remove_old_plugin(pid) + await self.__async_remove_old_plugin(pid) - success, message = await prepare_content() - if not success: - logger.error(f"{pid} 准备插件内容失败：{message}") + success, message = await prepare_content() + if not success: + logger.error(f"{pid} 准备插件内容失败：{message}") + if backup_dir: + await self.__async_restore_plugin(pid, backup_dir) + logger.warning(f"{pid} 插件安装失败，已还原备份插件") + else: + await self.__async_remove_old_plugin(pid) + logger.warning(f"{pid} 已清理对应插件目录，请尝试重新安装") + return False, message + + dependencies_exist, dep_ok, dep_msg = ( + await self.__async_install_dependencies_if_required(pid) + ) + if dependencies_exist and not dep_ok: + logger.error(f"{pid} 依赖安装失败：{dep_msg}") + if backup_dir: + await self.__async_restore_plugin(pid, backup_dir) + logger.warning(f"{pid} 插件安装失败，已还原备份插件") + else: + await self.__async_remove_old_plugin(pid) + logger.warning(f"{pid} 已清理对应插件目录，请尝试重新安装") + return False, dep_msg + + await _await_thread_operation(self.refresh_persistent_plugin_backup, pid) + return True, "" + except asyncio.CancelledError: + logger.warning( + f"{pid} 插件安装被取消，Python 依赖环境可能已经改变" + ) + raise + finally: if backup_dir: - await self.__async_restore_plugin(pid, backup_dir) - logger.warn(f"{pid} 插件安装失败，已还原备份插件") - else: - await self.__async_remove_old_plugin(pid) - logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装") - return False, message - - dependencies_exist, dep_ok, dep_msg = await self.__async_install_dependencies_if_required(pid) - if dependencies_exist and not dep_ok: - logger.error(f"{pid} 依赖安装失败：{dep_msg}") - if backup_dir: - await self.__async_restore_plugin(pid, backup_dir) - logger.warn(f"{pid} 插件安装失败，已还原备份插件") - else: - await self.__async_remove_old_plugin(pid) - logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装") - return False, dep_msg - - await asyncio.to_thread(self.refresh_persistent_plugin_backup, pid) - return True, "" + await aioshutil.rmtree(backup_dir, ignore_errors=True) def __prepare_content_via_filelist_sync(self, pid: str, user_repo: str, package_version: Optional[str]) -> Tuple[bool, str]: diff --git a/app/adapters/system/host.py b/app/adapters/system/host.py index c5ebff8b4..d7cb00870 100644 --- a/app/adapters/system/host.py +++ b/app/adapters/system/host.py @@ -1,8 +1,10 @@ +import asyncio import datetime import hashlib import os import platform import re +import signal import shutil import socket import struct @@ -101,6 +103,7 @@ class SystemUtils: command: list, env: Optional[dict[str, str]] = None, safe_command: Optional[list[str]] = None, + timeout: Optional[float] = 300, ) -> Tuple[bool, str]: """ 执行命令并捕获标准输出和错误输出，记录日志。 @@ -108,6 +111,7 @@ class SystemUtils: :param command: 要执行的命令，以列表形式提供 :param env: 传递给子进程的环境变量 :param safe_command: 用于错误信息展示的脱敏命令 + :param timeout: 子进程最长运行时间，None 表示不设置超时 :return: (命令是否成功, 输出信息或错误信息) """ display_command = safe_command or command @@ -120,6 +124,7 @@ class SystemUtils: stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, + timeout=timeout, ) # 合并 stdout 和 stderr output = SystemUtils.redact_url_userinfo(result.stdout + result.stderr) @@ -140,6 +145,26 @@ class SystemUtils: f"返回码：{e.returncode}，{'; '.join(output_parts)}" ) return False, error_message + except subprocess.TimeoutExpired as e: + stdout = SystemUtils.redact_url_userinfo( + SystemUtils._decode_subprocess_output(e.stdout) + ) + stderr = SystemUtils.redact_url_userinfo( + SystemUtils._decode_subprocess_output(e.stderr) + ) + output = "；".join( + part for part in ( + f"标准输出：{stdout}" if stdout else "", + f"错误输出：{stderr}" if stderr else "", + ) if part + ) + if output: + output = f"，{output}" + timeout_text = "未设置" if timeout is None else f"{timeout:g} 秒" + return False, ( + f"命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，" + f"执行超时（{timeout_text}）{output}" + ) except Exception as e: error_message = ( f"未知错误，命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，" @@ -147,6 +172,213 @@ class SystemUtils: ) return False, error_message + @staticmethod + def _decode_subprocess_output(value: object) -> str: + """把 subprocess 的文本或字节输出统一成可脱敏的字符串。""" + if value is None: + return "" + if isinstance(value, bytes): + return value.decode(errors="replace").strip() + return str(value).strip() + + @staticmethod + async def _terminate_async_subprocess( + process: asyncio.subprocess.Process, + communication_task: "asyncio.Task[tuple[bytes, bytes]]", + grace_seconds: float = 5, + ) -> None: + """终止安装子进程及其同组子进程，并确保句柄已回收。""" + process_tree = SystemUtils._process_tree(process.pid) + if process.returncode is None: + try: + if os.name == "nt": + SystemUtils._signal_process_tree(process_tree) + process.terminate() + else: + os.killpg(process.pid, signal.SIGTERM) + except (ProcessLookupError, OSError): + pass + + try: + await asyncio.wait_for( + asyncio.shield(communication_task), + timeout=grace_seconds, + ) + except (asyncio.TimeoutError, asyncio.CancelledError): + pass + + # 管道可能在父进程退出时立即关闭，而后代仍在运行或忽略终止信号。 + # 通信任务完成只代表 stdout/stderr 已收口，不能作为进程树已收口的依据。 + try: + known_pids = {item.pid for item in process_tree} + process_tree.extend( + process_item + for process_item in SystemUtils._process_tree(process.pid) + if process_item.pid not in known_pids + ) + except (ProcessLookupError, OSError): + pass + + alive_processes = SystemUtils._alive_processes(process_tree) + if alive_processes or process.returncode is None: + if os.name == "nt": + try: + SystemUtils._signal_process_tree(alive_processes, force=True) + process.kill() + except (ProcessLookupError, OSError): + pass + else: + try: + os.killpg(process.pid, signal.SIGKILL) + except (ProcessLookupError, OSError): + pass + try: + SystemUtils._signal_process_tree(alive_processes, force=True) + except (ProcessLookupError, OSError): + pass + await SystemUtils._wait_process_tree( + process_tree, + grace_seconds, + ) + if not communication_task.done(): + communication_task.cancel() + await asyncio.gather(communication_task, return_exceptions=True) + try: + await process.wait() + except (ProcessLookupError, OSError): + pass + + @staticmethod + def _process_tree(pid: int) -> list[psutil.Process]: + """收集安装进程及其后代，避免构建进程继续写运行环境。""" + try: + parent = psutil.Process(pid) + return [parent, *parent.children(recursive=True)] + except (psutil.Error, OSError): + return [] + + @staticmethod + def _alive_processes(processes: list[psutil.Process]) -> list[psutil.Process]: + """返回仍可能执行外部副作用的进程，忽略已退出和僵尸进程。""" + alive = [] + seen_pids = set() + for process in processes: + if process.pid in seen_pids: + continue + seen_pids.add(process.pid) + try: + if process.is_running() and process.status() != psutil.STATUS_ZOMBIE: + alive.append(process) + except (psutil.Error, OSError): + continue + return alive + + @staticmethod + async def _wait_process_tree( + processes: list[psutil.Process], + timeout: float, + ) -> list[psutil.Process]: + """在事件循环中有界等待整棵进程树退出，并返回残留进程。""" + deadline = asyncio.get_running_loop().time() + max(timeout, 0) + while True: + alive = SystemUtils._alive_processes(processes) + if not alive: + return [] + remaining = deadline - asyncio.get_running_loop().time() + if remaining <= 0: + return alive + await asyncio.sleep(min(0.05, remaining)) + + @staticmethod + def _signal_process_tree( + processes: list[psutil.Process], + force: bool = False, + ) -> None: + """向进程树发送终止或强制结束信号。""" + for child in reversed(processes): + try: + (child.kill if force else child.terminate)() + except (psutil.Error, OSError): + continue + + @staticmethod + async def execute_with_subprocess_async( + command: list, + env: Optional[dict[str, str]] = None, + safe_command: Optional[list[str]] = None, + timeout: Optional[float] = 300, + ) -> Tuple[bool, str]: + """异步执行可取消的子进程，超时或取消时回收整个进程组。""" + display_command = safe_command or command + process: Optional[asyncio.subprocess.Process] = None + communication_task: Optional["asyncio.Task[tuple[bytes, bytes]]"] = None + try: + creationflags = ( + getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + if os.name == "nt" + else 0 + ) + process = await asyncio.create_subprocess_exec( + *command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=env, + start_new_session=os.name != "nt", + creationflags=creationflags, + ) + communication_task = asyncio.create_task(process.communicate()) + try: + stdout, stderr = await asyncio.wait_for( + asyncio.shield(communication_task), + timeout=timeout, + ) + except asyncio.TimeoutError: + await SystemUtils._terminate_async_subprocess( + process, + communication_task, + ) + timeout_text = "未设置" if timeout is None else f"{timeout:g} 秒" + return False, ( + f"命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，" + f"执行超时（{timeout_text}）" + ) + except asyncio.CancelledError: + await SystemUtils._terminate_async_subprocess( + process, + communication_task, + ) + raise + + output = SystemUtils.redact_url_userinfo( + SystemUtils._decode_subprocess_output(stdout) + + SystemUtils._decode_subprocess_output(stderr) + ) + if process.returncode == 0: + return True, output + return False, ( + f"命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，" + f"执行失败，返回码：{process.returncode}，" + f"{output or '无标准输出或错误输出'}" + ) + except asyncio.CancelledError: + if process is not None and communication_task is not None: + await SystemUtils._terminate_async_subprocess( + process, + communication_task, + ) + raise + except Exception as e: + if process is not None and communication_task is not None: + await SystemUtils._terminate_async_subprocess( + process, + communication_task, + ) + error_message = ( + f"未知错误，命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，" + f"错误：{SystemUtils.redact_url_userinfo(str(e))}" + ) + return False, error_message + @staticmethod def is_docker() -> bool: """ diff --git a/app/adapters/system/plugin/dependency.py b/app/adapters/system/plugin/dependency.py index fefa9f5e9..80afb2db6 100644 --- a/app/adapters/system/plugin/dependency.py +++ b/app/adapters/system/plugin/dependency.py @@ -364,5 +364,17 @@ class PluginDependencyInstaller: return await asyncio.to_thread(self.find_missing) async def async_install(self, dependencies: list[str]) -> tuple[bool, str]: - """在线程池中安装依赖，复用同步包安装策略。""" - return await asyncio.to_thread(self.install, dependencies) + """异步安装依赖，使用可取消的包安装子进程。""" + if not dependencies: + return False, "没有传入需要安装的依赖项" + try: + manifest_paths = [manifest.path for manifest in self._plugin_manifests()] + if not manifest_paths: + return False, "没有找到已安装插件的依赖清单" + return await self._helper.async_install_packages_with_fallback( + manifest_paths, + self._wheels_dirs(), + ) + except Exception as err: + logger.error(f"安装依赖项时发生错误：{err}") + return False, f"安装依赖项时发生错误：{err}" diff --git a/app/adapters/system/plugin/package.py b/app/adapters/system/plugin/package.py index 531f4f8c6..c01e05fe1 100644 --- a/app/adapters/system/plugin/package.py +++ b/app/adapters/system/plugin/package.py @@ -19,6 +19,19 @@ from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() +async def _await_thread_operation(func, *args, **kwargs): + """取消请求到达时先等待文件操作收口，避免后台线程继续写运行目录。""" + task = asyncio.create_task(asyncio.to_thread(func, *args, **kwargs)) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + try: + await asyncio.shield(task) + except BaseException: + pass + raise + + @dataclass(frozen=True, slots=True) class PluginPackageCheckpoint: """记录一次插件包变更前可用于补偿恢复的文件快照。""" @@ -74,7 +87,7 @@ class PluginPackageManager: async def async_checkpoint(self, plugin_id: str) -> PluginPackageCheckpoint: """在线程池中创建插件包文件快照。""" - return await asyncio.to_thread(self.checkpoint, plugin_id) + return await _await_thread_operation(self.checkpoint, plugin_id) @staticmethod def commit(checkpoint: PluginPackageCheckpoint) -> None: @@ -83,25 +96,27 @@ class PluginPackageManager: async def async_commit(self, checkpoint: PluginPackageCheckpoint) -> None: """在线程池中清理已提交的插件包快照。""" - await asyncio.to_thread(self.commit, checkpoint) + await _await_thread_operation(self.commit, checkpoint) @staticmethod def rollback(checkpoint: PluginPackageCheckpoint) -> None: """删除当前包并把变更前文件快照恢复到运行目录。""" - if checkpoint.plugin_dir.exists(): - shutil.rmtree(checkpoint.plugin_dir) snapshot_dir = checkpoint.transaction_dir / "package" if checkpoint.existed: if not snapshot_dir.is_dir(): raise FileNotFoundError( f"插件 {checkpoint.plugin_id} 的补偿快照不存在：{snapshot_dir}" ) + if checkpoint.plugin_dir.exists(): + shutil.rmtree(checkpoint.plugin_dir) + if checkpoint.existed: shutil.copytree(snapshot_dir, checkpoint.plugin_dir) - shutil.rmtree(checkpoint.transaction_dir, ignore_errors=False) + if checkpoint.transaction_dir.exists(): + shutil.rmtree(checkpoint.transaction_dir, ignore_errors=False) async def async_rollback(self, checkpoint: PluginPackageCheckpoint) -> None: """在线程池中恢复插件包文件快照。""" - await asyncio.to_thread(self.rollback, checkpoint) + await _await_thread_operation(self.rollback, checkpoint) def install( self, diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index a5d1c5b6b..1550507e3 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -57,7 +57,7 @@ from app.api.dependencies.plugin import ( from app.adapters.external.server import MoviePilotServerHelper from app.adapters.external.market import PluginHelper from app.adapters.system.plugin.package import PluginPackageManager -from app.application.database import DatabaseWorkerOverloadedError +from app.schemas.exception import DatabaseWorkerOverloadedError from app.runtime.log import logger from app.schemas.types import SystemConfigKey from app.api.context import get_background_task_registry, resolve_background_task_registry diff --git a/app/application/database.py b/app/application/database.py index 78e6dafa5..d69414d27 100644 --- a/app/application/database.py +++ b/app/application/database.py @@ -18,14 +18,6 @@ DatabaseProbe = Callable[[], Optional[str]] T = TypeVar("T") -class DatabaseWorkerClosedError(RuntimeError): - """数据库执行器尚未启动或已经停止。""" - - -class DatabaseWorkerOverloadedError(RuntimeError): - """数据库执行器的运行与排队容量已经用尽。""" - - class AsyncDatabaseExecutor(Protocol): """让异步业务调用同步短事务而不阻塞事件循环。""" diff --git a/app/application/messaging/chat.py b/app/application/messaging/chat.py index bdb831348..5a8b3cb22 100644 --- a/app/application/messaging/chat.py +++ b/app/application/messaging/chat.py @@ -10,10 +10,12 @@ from weakref import WeakValueDictionary from app.application.database import ( AsyncDatabaseExecutor, +) +from app.schemas.agent import AgentChatSessionDetail, AgentChatSessionSummary +from app.schemas.exception import ( DatabaseWorkerClosedError, DatabaseWorkerOverloadedError, ) -from app.schemas.agent import AgentChatSessionDetail, AgentChatSessionSummary from app.runtime.observability import record_metric diff --git a/app/application/plugin/install.py b/app/application/plugin/install.py index 90eddd7c7..3c50bc547 100644 --- a/app/application/plugin/install.py +++ b/app/application/plugin/install.py @@ -2,11 +2,14 @@ from __future__ import annotations +import asyncio from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Any, Optional -from app.application.database import DatabaseWorkerOverloadedError +from app.schemas.exception import DatabaseWorkerOverloadedError +from app.application.plugin.lifecycle import plugin_lifecycle +from app.runtime.log import logger InstalledPluginsReader = Callable[[], list[str]] @@ -58,6 +61,22 @@ class PluginInstallResult: rollback: PluginInstallRollback = field(default_factory=PluginInstallRollback) +@dataclass +class _InstallState: + """记录取消补偿所需的事务阶段。""" + + checkpoint: Any = None + stage: str = "package_checkpoint" + package_installed: bool = False + installed_list_touched: bool = False + installed_list_persisted: bool = False + runtime_touched: bool = False + registrations_touched: bool = False + refresh_compensated: bool = False + committed: bool = False + original_plugins: list[str] = field(default_factory=list) + + class PluginInstallCommand: """协调插件检查、包事务、持久化、运行态刷新和安装上报。""" @@ -96,14 +115,44 @@ class PluginInstallCommand: repo_url: Optional[str], release_version: Optional[str] = None, force: bool = False, + ) -> PluginInstallResult: + """串行执行同一插件的完整安装生命周期，并保证取消后的补偿。""" + state = _InstallState() + async with plugin_lifecycle.hold(plugin_id): + try: + return await self._execute_locked( + plugin_id=plugin_id, + repo_url=repo_url, + release_version=release_version, + force=force, + state=state, + ) + except asyncio.CancelledError: + await self._rollback_cancelled( + plugin_id=plugin_id, + original_plugins=state.original_plugins, + state=state, + ) + raise + + async def _execute_locked( + self, + *, + plugin_id: str, + repo_url: Optional[str], + release_version: Optional[str], + force: bool, + state: _InstallState, ) -> PluginInstallResult: """执行插件安装，并在关键阶段失败时恢复可补偿状态。""" installed_plugins = list(self._installed_plugins_reader() or []) + state.original_plugins = installed_plugins refreshed_only = not force and plugin_id in self._plugin_ids_provider() if refreshed_only: return await self._refresh_existing( plugin_id=plugin_id, repo_url=repo_url, + state=state, ) if not repo_url: return PluginInstallResult( @@ -112,8 +161,16 @@ class PluginInstallCommand: failure_stage="validation", ) + checkpoint_task = asyncio.create_task(self._package_checkpointer(plugin_id)) try: - checkpoint = await self._package_checkpointer(plugin_id) + checkpoint = await asyncio.shield(checkpoint_task) + state.checkpoint = checkpoint + except asyncio.CancelledError: + try: + state.checkpoint = await asyncio.shield(checkpoint_task) + except BaseException: + pass + raise except Exception as err: return PluginInstallResult( success=False, @@ -121,13 +178,15 @@ class PluginInstallCommand: failure_stage="package_checkpoint", ) + state.stage = "package_install" try: - state, message = await self._package_installer( + package_installed, message = await self._package_installer( plugin_id, repo_url, release_version, force, ) + state.package_installed = package_installed except Exception as err: result = await self._failure( plugin_id=plugin_id, @@ -140,7 +199,7 @@ class PluginInstallCommand: if isinstance(err, DatabaseWorkerOverloadedError): raise return result - if not state: + if not package_installed: return await self._failure( plugin_id=plugin_id, original_plugins=installed_plugins, @@ -154,8 +213,11 @@ class PluginInstallCommand: if plugin_id not in installed_plugins: updated_plugins = [*installed_plugins, plugin_id] try: + # 写入方可能在返回前已经提交；取消时按已触碰处理，恢复原清单是幂等的。 + state.installed_list_touched = True await self._installed_plugins_writer(updated_plugins) installed_list_persisted = True + state.installed_list_persisted = True except Exception as err: result = await self._failure( plugin_id=plugin_id, @@ -164,11 +226,14 @@ class PluginInstallCommand: stage="installed_list_persistence", message=str(err), package_installed=True, + installed_list_persisted=state.installed_list_touched, ) if isinstance(err, DatabaseWorkerOverloadedError): raise return result + state.stage = "runtime_reload" + state.runtime_touched = True try: await self._plugin_reloader(plugin_id) except Exception as err: @@ -186,6 +251,8 @@ class PluginInstallCommand: raise return result + state.stage = "registration_refresh" + state.registrations_touched = True try: await self._registration_refresher(plugin_id) except Exception as err: @@ -205,6 +272,9 @@ class PluginInstallCommand: return result checkpoint_cleanup_error = "" + state.stage = "checkpoint_commit" + # 运行态和注册已完成，后续只清理临时快照，不再把取消当作未提交安装回滚。 + state.committed = True try: await self._package_committer(checkpoint) except Exception as err: @@ -212,6 +282,7 @@ class PluginInstallCommand: reported = False report_error = "" + state.stage = "report" try: report_result = await self._install_reporter(plugin_id, repo_url) reported = report_result is not False @@ -237,11 +308,65 @@ class PluginInstallCommand: checkpoint_cleanup_error=checkpoint_cleanup_error, ) + async def _rollback_cancelled( + self, + *, + plugin_id: str, + original_plugins: list[str], + state: _InstallState, + ) -> None: + """在保留取消语义的同时完成文件、清单和运行态补偿。""" + if state.committed: + logger.warning( + f"插件 {plugin_id} 在安装提交后被取消，Python 依赖环境可能已经改变" + ) + return + if state.refresh_compensated: + return + if state.checkpoint is None: + logger.warning( + f"插件 {plugin_id} 在创建安装快照前被取消，无法执行文件补偿" + ) + return + + rollback_task = asyncio.create_task( + self._failure( + plugin_id=plugin_id, + original_plugins=original_plugins, + checkpoint=state.checkpoint, + stage=state.stage, + message="插件安装已取消", + package_installed=state.package_installed, + installed_list_persisted=state.installed_list_touched, + runtime_touched=state.runtime_touched, + registrations_touched=state.registrations_touched, + ) + ) + try: + result = await asyncio.shield(rollback_task) + except asyncio.CancelledError: + try: + result = await asyncio.shield(rollback_task) + except BaseException as err: + logger.error(f"插件 {plugin_id} 取消后的补偿未完成：{err}") + return + except Exception as err: + logger.error(f"插件 {plugin_id} 取消后的补偿失败：{err}") + return + if result.rollback.errors: + logger.error( + f"插件 {plugin_id} 取消后的补偿存在错误：{'；'.join(result.rollback.errors)}" + ) + logger.warning( + f"插件 {plugin_id} 安装已取消，插件文件已尝试恢复，Python 依赖环境可能已经改变" + ) + async def _refresh_existing( self, *, plugin_id: str, repo_url: Optional[str], + state: _InstallState, ) -> PluginInstallResult: """刷新已存在插件，不触碰包文件和已安装列表。""" if repo_url: @@ -261,33 +386,31 @@ class PluginInstallCommand: await self._plugin_reloader(plugin_id) failure_stage = "registration_refresh" await self._registration_refresher(plugin_id) - except Exception as err: - rollback_errors = [] - runtime_restored = False - registrations_restored = False - try: - await self._plugin_reloader(plugin_id) - runtime_restored = True - except Exception as rollback_err: - rollback_errors.append(f"运行态恢复失败：{rollback_err}") - if runtime_restored: + except asyncio.CancelledError: + cleanup_task = asyncio.create_task( + self._restore_refreshed_runtime(plugin_id) + ) + while not cleanup_task.done(): try: - await self._registration_refresher(plugin_id) - registrations_restored = True - except Exception as rollback_err: - rollback_errors.append(f"路由和服务注册恢复失败：{rollback_err}") + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: + continue + rollback = await cleanup_task + state.refresh_compensated = True + if rollback.errors: + logger.error( + f"插件 {plugin_id} 取消刷新后的运行态补偿存在错误：" + f"{'；'.join(rollback.errors)}" + ) + raise + except Exception as err: + rollback = await self._restore_refreshed_runtime(plugin_id) result = PluginInstallResult( success=False, message=f"刷新插件运行态失败：{err}", refreshed_only=True, failure_stage=failure_stage, - rollback=PluginInstallRollback( - runtime_attempted=True, - runtime_restored=runtime_restored, - registrations_attempted=True, - registrations_restored=registrations_restored, - errors=tuple(rollback_errors), - ), + rollback=rollback, ) if isinstance(err, DatabaseWorkerOverloadedError): raise @@ -316,6 +439,33 @@ class PluginInstallCommand: report_error=report_error, ) + async def _restore_refreshed_runtime( + self, + plugin_id: str, + ) -> PluginInstallRollback: + """重新加载插件并刷新注册，使中断的运行态切换恢复到完整状态。""" + errors = [] + runtime_restored = False + registrations_restored = False + try: + await self._plugin_reloader(plugin_id) + runtime_restored = True + except Exception as err: + errors.append(f"运行态恢复失败：{err}") + if runtime_restored: + try: + await self._registration_refresher(plugin_id) + registrations_restored = True + except Exception as err: + errors.append(f"路由和服务注册恢复失败：{err}") + return PluginInstallRollback( + runtime_attempted=True, + runtime_restored=runtime_restored, + registrations_attempted=True, + registrations_restored=registrations_restored, + errors=tuple(errors), + ) + async def _failure( self, *, diff --git a/app/application/plugin/lifecycle.py b/app/application/plugin/lifecycle.py new file mode 100644 index 000000000..04b85a551 --- /dev/null +++ b/app/application/plugin/lifecycle.py @@ -0,0 +1,71 @@ +"""插件安装与启动同步之间共享的生命周期互斥。""" + +from __future__ import annotations + +import asyncio +import threading +from contextlib import asynccontextmanager + + +class PluginLifecycleCoordinator: + """在事件循环和同步启动线程之间协调插件生命周期操作。""" + + def __init__(self) -> None: + self._condition = threading.Condition() + self._active_plugins: set[str] = set() + self._startup_active = False + + @staticmethod + def _normalize(plugin_id: str) -> str: + return (plugin_id or "").strip().lower() + + def _try_acquire_plugin(self, plugin_id: str) -> bool: + normalized_id = self._normalize(plugin_id) + if not normalized_id: + raise ValueError("插件ID不能为空") + with self._condition: + if self._startup_active or normalized_id in self._active_plugins: + return False + self._active_plugins.add(normalized_id) + return True + + def _release_plugin(self, plugin_id: str) -> None: + normalized_id = self._normalize(plugin_id) + with self._condition: + self._active_plugins.discard(normalized_id) + self._condition.notify_all() + + def _try_acquire_startup(self) -> bool: + with self._condition: + if self._startup_active or self._active_plugins: + return False + self._startup_active = True + return True + + def _release_startup(self) -> None: + with self._condition: + self._startup_active = False + self._condition.notify_all() + + @asynccontextmanager + async def hold(self, plugin_id: str): + """异步持有单个插件的生命周期资格，不在线程池中等待锁。""" + while not self._try_acquire_plugin(plugin_id): + await asyncio.sleep(0.01) + try: + yield + finally: + self._release_plugin(plugin_id) + + @asynccontextmanager + async def hold_startup(self): + """异步持有启动同步的全局资格，阻止安装请求穿过启动收口。""" + while not self._try_acquire_startup(): + await asyncio.sleep(0.01) + try: + yield + finally: + self._release_startup() + + +plugin_lifecycle = PluginLifecycleCoordinator() diff --git a/app/db/worker.py b/app/db/worker.py index 1690df663..122366f42 100644 --- a/app/db/worker.py +++ b/app/db/worker.py @@ -10,7 +10,7 @@ from contextvars import copy_context from dataclasses import dataclass from typing import Callable, TypeVar -from app.application.database import ( +from app.schemas.exception import ( DatabaseWorkerClosedError, DatabaseWorkerOverloadedError, ) diff --git a/app/factory.py b/app/factory.py index 7db6a5e82..12a06b8bd 100644 --- a/app/factory.py +++ b/app/factory.py @@ -14,7 +14,7 @@ from app.adapters.observability.otel import build_observation_port from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry from app.adapters.web.health import install_health_routes from app.application.plugin.routes import configure_plugin_routes -from app.application.database import ( +from app.schemas.exception import ( DatabaseWorkerClosedError, DatabaseWorkerOverloadedError, ) diff --git a/app/runtime/extensions/plugin/dependency.py b/app/runtime/extensions/plugin/dependency.py index 691c9c6b1..5a026f4b2 100644 --- a/app/runtime/extensions/plugin/dependency.py +++ b/app/runtime/extensions/plugin/dependency.py @@ -63,6 +63,27 @@ class PluginDependencyService: """安装当前环境缺失的插件依赖并保持历史列表返回合同。""" return self.install_missing_with_status().missing + async def async_install_missing_with_status(self) -> PluginDependencyInstallResult: + """在异步启动链中恢复缺失依赖，确保安装子进程可取消。""" + installer = self._system().dependency + missing = await installer.async_find_missing() + if not missing: + return PluginDependencyInstallResult(missing=[], success=True) + self._logger.debug(f"检测到缺失的依赖项: {missing}") + self._logger.info(f"开始安装缺失的依赖项，共 {len(missing)} 个...") + started = time.time() + success, _message = await installer.async_install(missing) + elapsed = time.time() - started + if success: + self._logger.info( + f"已完成 {len(missing)} 个依赖项安装，总耗时：{elapsed:.2f} 秒" + ) + else: + self._logger.warning( + f"存在缺失依赖项安装失败，请尝试手动安装，总耗时：{elapsed:.2f} 秒" + ) + return PluginDependencyInstallResult(missing=missing, success=success) + def classify_plugins(self) -> PluginDependencyClassification: """返回启动编排使用的轻量插件分类。""" ready, missing_dependencies, missing_source = ( diff --git a/app/runtime/extensions/plugin_manager.py b/app/runtime/extensions/plugin_manager.py index d593e6e71..746b3a00d 100644 --- a/app/runtime/extensions/plugin_manager.py +++ b/app/runtime/extensions/plugin_manager.py @@ -603,6 +603,14 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): log=logger, ).install_missing_with_status() + @staticmethod + async def async_install_plugin_missing_dependencies_with_status() -> PluginDependencyInstallResult: + """在异步启动链中恢复插件依赖并保留取消语义。""" + return await PluginDependencyService( + system=get_plugin_system, + log=logger, + ).async_install_missing_with_status() + def classify_plugins(self) -> PluginDependencyClassification: """按源码依赖状态分类物理插件，并把结果映射到虚拟实例。""" source_classification = PluginDependencyService( diff --git a/app/schemas/exception.py b/app/schemas/exception.py index f51b5759e..f14b74795 100644 --- a/app/schemas/exception.py +++ b/app/schemas/exception.py @@ -1,3 +1,14 @@ +__all__ = ( + "ImmediateException", + "LimitException", + "APIRateLimitException", + "RateLimitExceededException", + "OperationInterrupted", + "StorageQueryError", + "TMDbException", +) + + class ImmediateException(Exception): """ 用于立即抛出异常而不重试的特殊异常类。 @@ -47,6 +58,14 @@ class StorageQueryError(Exception): pass +class DatabaseWorkerClosedError(RuntimeError): + """数据库执行器尚未启动或已经停止。""" + + +class DatabaseWorkerOverloadedError(RuntimeError): + """数据库执行器的运行与排队容量已经用尽。""" + + class TMDbException(Exception): """ 用于表示TheMovieDB数据源请求失败的跨层异常契约。 diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index 09bde9822..493516afa 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -25,6 +25,7 @@ except Exception: pass from app.chain.system import SystemChain +from app.application.plugin.lifecycle import plugin_lifecycle from app.application.plugin.runtime import get_plugin_manager from app.runtime.config import global_vars from app.runtime.settings import RuntimeSettingsCompat @@ -77,13 +78,14 @@ async def init_extra(): return plugin_manager = get_plugin_manager() try: - if await sync_plugins(): - await execute_task( - global_vars.loop, - init_plugin_scheduler, - "插件定时服务刷新", - ) - await asyncio.wrap_future(restart_command()) + async with plugin_lifecycle.hold_startup(): + if await sync_plugins(): + await execute_task( + global_vars.loop, + init_plugin_scheduler, + "插件定时服务刷新", + ) + await asyncio.wrap_future(restart_command()) finally: plugin_manager.set_plugin_settling(False) plugin_manager.start_monitor() diff --git a/app/startup/plugins_initializer.py b/app/startup/plugins_initializer.py index 583feafb3..f82a6f26b 100644 --- a/app/startup/plugins_initializer.py +++ b/app/startup/plugins_initializer.py @@ -148,10 +148,8 @@ async def sync_plugins() -> bool: plugin_manager.set_plugin_settling(True) sync_result = await execute_task(loop, plugin_manager.sync, "插件同步到本地") - dependency_result = await execute_task( - loop, - plugin_manager.install_plugin_missing_dependencies_with_status, - "缺失依赖项安装", + dependency_result = await ( + plugin_manager.async_install_plugin_missing_dependencies_with_status() ) if dependency_result is None: return False diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 31b182b18..9cb439b3c 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6463, - "edge_sha256": "086c14fef27199ae6f19fbe5e6a07c348cd1dc8a406ab3129a341647fd90a28f", + "edge_count": 6470, + "edge_sha256": "571b66b75b51e801a053e4164c5dbe0eb580b91f20cf84fa554e13b9469dd650", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -2062,7 +2062,6 @@ "app.api.endpoints.plugin -> app.application", "app.api.endpoints.plugin -> app.application.commands", "app.api.endpoints.plugin -> app.application.configuration", - "app.api.endpoints.plugin -> app.application.database", "app.api.endpoints.plugin -> app.application.plugin", "app.api.endpoints.plugin -> app.application.plugin.config", "app.api.endpoints.plugin -> app.application.plugin.folders", @@ -2079,6 +2078,7 @@ "app.api.endpoints.plugin -> app.runtime.tasks", "app.api.endpoints.plugin -> app.schemas", "app.api.endpoints.plugin -> app.schemas.common", + "app.api.endpoints.plugin -> app.schemas.exception", "app.api.endpoints.plugin -> app.schemas.plugin", "app.api.endpoints.plugin -> app.schemas.response", "app.api.endpoints.plugin -> app.schemas.token", @@ -2579,6 +2579,7 @@ "app.application.messaging.chat -> app.runtime.observability", "app.application.messaging.chat -> app.schemas", "app.application.messaging.chat -> app.schemas.agent", + "app.application.messaging.chat -> app.schemas.exception", "app.application.messaging.interaction -> app.schemas", "app.application.messaging.interaction -> app.schemas.message", "app.application.messaging.interaction -> app.schemas.notification", @@ -2665,7 +2666,12 @@ "app.application.plugin.folders -> app.schemas", "app.application.plugin.folders -> app.schemas.types", "app.application.plugin.install -> app.application", - "app.application.plugin.install -> app.application.database", + "app.application.plugin.install -> app.application.plugin", + "app.application.plugin.install -> app.application.plugin.lifecycle", + "app.application.plugin.install -> app.runtime", + "app.application.plugin.install -> app.runtime.log", + "app.application.plugin.install -> app.schemas", + "app.application.plugin.install -> app.schemas.exception", "app.application.recognition -> app.application", "app.application.recognition -> app.application.configuration", "app.application.recognition -> app.schemas", @@ -3694,10 +3700,10 @@ "app.db.session -> app.runtime.config", "app.db.session -> app.runtime.log", "app.db.session -> app.runtime.observability", - "app.db.worker -> app.application", - "app.db.worker -> app.application.database", "app.db.worker -> app.runtime", "app.db.worker -> app.runtime.observability", + "app.db.worker -> app.schemas", + "app.db.worker -> app.schemas.exception", "app.doctor.checks -> app.adapters", "app.doctor.checks -> app.adapters.system", "app.doctor.checks -> app.adapters.system.backup", @@ -3821,7 +3827,6 @@ "app.factory -> app.api", "app.factory -> app.api.response", "app.factory -> app.application", - "app.factory -> app.application.database", "app.factory -> app.application.plugin", "app.factory -> app.application.plugin.routes", "app.factory -> app.application.security", @@ -3835,6 +3840,7 @@ "app.factory -> app.runtime.observability", "app.factory -> app.runtime.settings", "app.factory -> app.schemas", + "app.factory -> app.schemas.exception", "app.factory -> app.schemas.mcp", "app.factory -> app.schemas.openai", "app.factory -> app.schemas.response", @@ -6056,6 +6062,7 @@ "app.startup.lifecycle -> app.adapters.network.http", "app.startup.lifecycle -> app.application", "app.startup.lifecycle -> app.application.plugin", + "app.startup.lifecycle -> app.application.plugin.lifecycle", "app.startup.lifecycle -> app.application.plugin.runtime", "app.startup.lifecycle -> app.chain", "app.startup.lifecycle -> app.chain.system", @@ -6480,7 +6487,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 799, + "module_count": 800, "modules": [ "app", "app.adapters", @@ -6772,6 +6779,7 @@ "app.application.plugin.data", "app.application.plugin.folders", "app.application.plugin.install", + "app.application.plugin.lifecycle", "app.application.plugin.routes", "app.application.plugin.runtime", "app.application.recognition", diff --git a/tests/test_agent_chat_persistence.py b/tests/test_agent_chat_persistence.py index 513a21172..bee1bd917 100644 --- a/tests/test_agent_chat_persistence.py +++ b/tests/test_agent_chat_persistence.py @@ -11,7 +11,7 @@ from uuid import uuid4 import pytest from sqlalchemy import delete, select -from app.application.database import ( +from app.schemas.exception import ( DatabaseWorkerClosedError, DatabaseWorkerOverloadedError, ) diff --git a/tests/test_api_response.py b/tests/test_api_response.py index 78a3ee335..715aadf9e 100644 --- a/tests/test_api_response.py +++ b/tests/test_api_response.py @@ -23,7 +23,7 @@ from app.factory import ( localized_unhandled_exception_handler, localized_validation_exception_handler, ) -from app.application.database import ( +from app.schemas.exception import ( DatabaseWorkerClosedError, DatabaseWorkerOverloadedError, ) diff --git a/tests/test_database_worker.py b/tests/test_database_worker.py index 156862610..4e755980c 100644 --- a/tests/test_database_worker.py +++ b/tests/test_database_worker.py @@ -6,8 +6,8 @@ from unittest.mock import patch import pytest -from app.db.worker import ( - DatabaseWorker, +from app.db.worker import DatabaseWorker +from app.schemas.exception import ( DatabaseWorkerClosedError, DatabaseWorkerOverloadedError, ) diff --git a/tests/test_plugin_dependency_service.py b/tests/test_plugin_dependency_service.py index b18981756..c762ffad6 100644 --- a/tests/test_plugin_dependency_service.py +++ b/tests/test_plugin_dependency_service.py @@ -1,7 +1,13 @@ +from unittest.mock import AsyncMock from types import SimpleNamespace from unittest.mock import MagicMock -from app.runtime.extensions.plugin.dependency import PluginDependencyService +import pytest + +from app.runtime.extensions.plugin.dependency import ( + PluginDependencyInstallResult, + PluginDependencyService, +) def test_install_missing_skips_installer_when_environment_is_satisfied() -> None: @@ -35,3 +41,25 @@ def test_install_missing_preserves_list_return_contract() -> None: assert service.install_missing() == ["demo>=1"] installer.install.assert_called_once_with(["demo>=1"]) + + +@pytest.mark.asyncio +async def test_async_install_missing_uses_async_installer() -> None: + """异步启动恢复必须调用可取消的依赖安装入口。""" + installer = SimpleNamespace( + async_find_missing=AsyncMock(return_value=["demo>=1"]), + async_install=AsyncMock(return_value=(True, "")), + ) + service = PluginDependencyService( + system=lambda: SimpleNamespace(dependency=installer), + log=MagicMock(), + ) + + result = await service.async_install_missing_with_status() + + assert result == PluginDependencyInstallResult( + missing=["demo>=1"], + success=True, + ) + installer.async_find_missing.assert_awaited_once() + installer.async_install.assert_awaited_once_with(["demo>=1"]) diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index 987d5ea20..61a45b046 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -9,7 +9,7 @@ import time import zipfile from pathlib import Path from types import ModuleType, SimpleNamespace -from unittest.mock import patch +from unittest.mock import AsyncMock, Mock, patch import pytest @@ -1615,39 +1615,251 @@ demo = { index = "private" } assert env["HTTPS_PROXY"] == "http://proxy.example:7890" assert "user:pass" not in " ".join(safe_command) - def test_async_package_install_runs_in_threadpool(self): - """ - 验证异步安装路径会把同步包安装派发到线程池，避免阻塞事件循环。 - """ + def test_async_package_install_uses_cancellable_subprocess(self): + """异步依赖安装应直接使用可取消的子进程执行器。""" try: from app.adapters.external.market import PluginHelper except ModuleNotFoundError as exc: pytest.skip(f"missing dependency: {exc}") helper = PluginHelper() - requirements_file = Path("/tmp/demo-requirements.txt") - find_links_dirs = [Path("/tmp/demo-wheels")] - calls = [] + with tempfile.TemporaryDirectory() as temp_dir: + requirements_file = Path(temp_dir) / "demo-requirements.txt" + requirements_file.write_text("demo-package\n", encoding="utf-8") + find_links_dirs = [Path(temp_dir) / "wheels"] - async def run_install(): - return await helper._PluginHelper__async_install_packages_with_fallback( - requirements_file, - find_links_dirs + async def run_install(): + return await helper._PluginHelper__async_install_packages_with_fallback( + requirements_file, + find_links_dirs, + ) + + health = { + "uv check": (True, "ok"), + "核心依赖导入检查": (True, "ok"), + } + strategy = Mock( + strategy_name="uv:test", + command=["uv", "pip", "install"], + env={}, + safe_log_command=["uv", "pip", "install"], ) - async def fake_to_thread(func, *args, **kwargs): - calls.append((func, args, kwargs)) - return True, "ok" - - with patch("app.adapters.external.market.asyncio.to_thread", side_effect=fake_to_thread): - success, message = asyncio.run(run_install()) + with patch.object( + PluginHelper, + "_PluginHelper__get_installed_packages", + return_value={}, + ), patch.object( + PluginHelper, + "_PluginHelper__get_protected_runtime_packages", + return_value={}, + ), patch.object( + PluginHelper, + "_PluginHelper__validate_runtime_dependency_conflicts", + return_value=(True, ""), + ), patch( + "app.adapters.external.market.build_package_install_strategies", + return_value=[strategy], + ), patch.object( + PluginHelper, + "_PluginHelper__async_run_runtime_healthcheck", + side_effect=[health, health], + ), patch.object( + PluginHelper, + "_PluginHelper__refresh_import_system", + ), patch( + "app.adapters.external.market.SystemUtils.execute_with_subprocess_async", + new=AsyncMock(return_value=(True, "ok")), + ) as execute_mock: + success, message = asyncio.run(run_install()) assert success assert "ok" == message - assert 1 == len(calls) - assert helper.install_packages_with_fallback == calls[0][0] - assert (requirements_file, find_links_dirs) == calls[0][1] - assert {} == calls[0][2] + execute_mock.assert_awaited_once() + assert execute_mock.await_args.kwargs["timeout"] == ( + PluginHelper.PLUGIN_DEPENDENCY_INSTALL_TIMEOUT + ) + + def test_async_package_install_cancellation_closes_full_lifecycle(self, tmp_path): + """取消真实安装进程后必须回收进程树、临时约束和安装锁。""" + import psutil + + from app.adapters.external.market import PluginHelper + + helper = PluginHelper() + requirements_file = tmp_path / "requirements.txt" + requirements_file.write_text("demo-package\n", encoding="utf-8") + constraints_file = tmp_path / "runtime-constraints.txt" + marker = tmp_path / "install-pids" + child_code = "import time; time.sleep(60)" + install_code = ( + "from pathlib import Path; import os, subprocess, time; " + f"child = subprocess.Popen([{sys.executable!r}, '-c', {child_code!r}]); " + f"Path({str(marker)!r}).write_text(str(os.getpid()) + ':' + str(child.pid)); " + "time.sleep(60)" + ) + strategy = Mock( + strategy_name="uv:test", + command=[sys.executable, "-c", install_code], + env=os.environ.copy(), + safe_log_command=[sys.executable, "-c", ""], + ) + health = { + "uv check": (True, "ok"), + "核心依赖导入检查": (True, "ok"), + } + + def create_constraints(_protected_packages): + constraints_file.write_text("fastapi==0\n", encoding="utf-8") + return constraints_file + + async def run_install(): + task = asyncio.create_task( + helper.async_install_packages_with_fallback(requirements_file) + ) + deadline = time.monotonic() + 2 + while not marker.exists() and time.monotonic() < deadline: + await asyncio.sleep(0.01) + assert marker.exists() + + pids = [int(value) for value in marker.read_text().split(":")] + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert not constraints_file.exists() + assert PluginHelper._package_install_lock.acquire(blocking=False) + PluginHelper._package_install_lock.release() + for _ in range(100): + alive = [] + for pid in pids: + try: + process = psutil.Process(pid) + if ( + process.is_running() + and process.status() != psutil.STATUS_ZOMBIE + ): + alive.append(pid) + except (psutil.Error, OSError): + continue + if not alive: + break + await asyncio.sleep(0.01) + else: + pytest.fail(f"安装进程树仍在运行：{alive}") + + with patch.object( + PluginHelper, + "_PluginHelper__get_installed_packages", + return_value={}, + ), patch.object( + PluginHelper, + "_PluginHelper__get_protected_runtime_packages", + return_value={"fastapi": "0"}, + ), patch.object( + PluginHelper, + "_PluginHelper__validate_runtime_dependency_conflicts", + return_value=(True, ""), + ), patch.object( + PluginHelper, + "_PluginHelper__create_runtime_constraints_file", + side_effect=create_constraints, + ), patch( + "app.adapters.external.market.build_package_install_strategies", + return_value=[strategy], + ), patch.object( + PluginHelper, + "_PluginHelper__async_run_runtime_healthcheck", + new=AsyncMock(return_value=health), + ): + asyncio.run(run_install()) + + def test_constraints_created_during_cancellation_are_removed(self, tmp_path): + """约束文件创建线程收口后仍须响应取消并删除临时文件。""" + from app.adapters.external.market import PluginHelper + + helper = PluginHelper() + requirements_file = tmp_path / "requirements.txt" + requirements_file.write_text("demo-package\n", encoding="utf-8") + constraints_file = tmp_path / "runtime-constraints.txt" + created = threading.Event() + release = threading.Event() + + def create_constraints(_protected_packages): + constraints_file.write_text("fastapi==0\n", encoding="utf-8") + created.set() + release.wait(timeout=2) + return constraints_file + + async def run_install(): + task = asyncio.create_task( + helper.async_install_packages_with_fallback(requirements_file) + ) + assert await asyncio.to_thread(created.wait, 2) + task.cancel() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + + with patch.object( + PluginHelper, + "_PluginHelper__get_installed_packages", + return_value={}, + ), patch.object( + PluginHelper, + "_PluginHelper__get_protected_runtime_packages", + return_value={"fastapi": "0"}, + ), patch.object( + PluginHelper, + "_PluginHelper__validate_runtime_dependency_conflicts", + return_value=(True, ""), + ), patch.object( + PluginHelper, + "_PluginHelper__create_runtime_constraints_file", + side_effect=create_constraints, + ): + asyncio.run(run_install()) + + assert not constraints_file.exists() + + def test_constraints_cleanup_failure_preserves_cancellation(self, tmp_path): + """临时文件删除失败只记录日志，不得替换调用方的取消异常。""" + from app.adapters.external.market import PluginHelper + + constraints_file = tmp_path / "runtime-constraints.txt" + created = threading.Event() + release = threading.Event() + + def create_constraints(_protected_packages): + constraints_file.write_text("fastapi==0\n", encoding="utf-8") + created.set() + release.wait(timeout=2) + return constraints_file + + async def run_create(): + task = asyncio.create_task( + PluginHelper._PluginHelper__async_create_runtime_constraints_file( + {"fastapi": Version("0")} + ) + ) + assert await asyncio.to_thread(created.wait, 2) + task.cancel() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + + with patch.object( + PluginHelper, + "_PluginHelper__create_runtime_constraints_file", + side_effect=create_constraints, + ), patch.object( + Path, + "unlink", + side_effect=PermissionError("locked"), + ), patch("app.adapters.external.market.logger.warning") as warning: + asyncio.run(run_create()) + + warning.assert_called_once() def test_install_uses_release_package_when_asset_is_available(self, monkeypatch): """ diff --git a/tests/test_plugin_install_command.py b/tests/test_plugin_install_command.py index 8b989d222..c77d8a4df 100644 --- a/tests/test_plugin_install_command.py +++ b/tests/test_plugin_install_command.py @@ -1,8 +1,9 @@ -from unittest.mock import AsyncMock, Mock +import asyncio +from unittest.mock import AsyncMock, Mock, patch import pytest -from app.application.database import DatabaseWorkerOverloadedError +from app.schemas.exception import DatabaseWorkerOverloadedError from app.application.plugin.install import PluginInstallCommand @@ -155,6 +156,42 @@ async def test_existing_plugin_checks_compatibility_without_reinstalling_package checkpointer.assert_not_awaited() +@pytest.mark.asyncio +async def test_cancelled_existing_plugin_refresh_restores_runtime_and_registrations(): + """已存在插件刷新被取消时，必须重新收敛运行态和注册。""" + registration_started = asyncio.Event() + calls: list[str] = [] + + async def reload_plugin(_plugin_id: str) -> None: + calls.append("reload") + + async def refresh_registrations(_plugin_id: str) -> None: + calls.append("registrations") + if calls.count("registrations") == 1: + registration_started.set() + await asyncio.Event().wait() + + with patch("app.application.plugin.install.logger.warning") as warning: + task = asyncio.create_task( + _command( + installed=["DemoPlugin"], + plugin_ids=["DemoPlugin"], + reloader=reload_plugin, + refresher=refresh_registrations, + ).execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + ) + await registration_started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert calls == ["reload", "registrations", "reload", "registrations"] + warning.assert_not_called() + + @pytest.mark.asyncio async def test_persistence_failure_restores_package_without_touching_runtime(): """已安装列表保存失败时恢复文件，且运行态尚未开始切换。""" @@ -176,12 +213,41 @@ async def test_persistence_failure_restores_package_without_touching_runtime(): assert result.success is False assert result.failure_stage == "installed_list_persistence" assert result.rollback.file_restored is True - assert result.rollback.installed_list_attempted is False + assert result.rollback.installed_list_attempted is True assert result.rollback.runtime_attempted is False rollback.assert_awaited_once_with(checkpoint) reloader.assert_not_awaited() +@pytest.mark.asyncio +async def test_persistence_exception_after_write_restores_installed_list(): + """清单写入已提交后抛异常时，文件和清单必须一起恢复。""" + persisted: list[list[str]] = [] + checkpoint = object() + rollback = AsyncMock() + + async def write(plugin_ids: list[str]) -> None: + persisted.append(list(plugin_ids)) + if len(persisted) == 1: + raise RuntimeError("write acknowledgement lost") + + result = await _command( + checkpointer=AsyncMock(return_value=checkpoint), + writer=write, + rollback=rollback, + ).execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + + assert result.success is False + assert result.failure_stage == "installed_list_persistence" + assert result.rollback.installed_list_attempted is True + assert result.rollback.installed_list_restored is True + assert persisted == [["DemoPlugin"], []] + rollback.assert_awaited_once_with(checkpoint) + + @pytest.mark.asyncio async def test_database_worker_overload_rolls_back_and_reaches_api_boundary(): """配置 worker 背压完成补偿后继续抛出，交由 API 映射为 503。""" @@ -306,6 +372,148 @@ async def test_registration_failure_restores_instance_files_and_routes() -> None ] +@pytest.mark.asyncio +async def test_same_plugin_install_lifecycle_is_serialized() -> None: + """同一插件的两个安装调用不得同时修改包、运行态和注册信息。""" + first_started = asyncio.Event() + release_first = asyncio.Event() + calls: list[str] = [] + + async def install(plugin_id, *_args): + calls.append(plugin_id) + if len(calls) == 1: + first_started.set() + await release_first.wait() + return True, "ok" + + command = _command(installer=install) + first = asyncio.create_task( + command.execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + ) + await first_started.wait() + second = asyncio.create_task( + command.execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + ) + await asyncio.sleep(0.02) + assert calls == ["DemoPlugin"] + + release_first.set() + results = await asyncio.gather(first, second) + assert all(result.success for result in results) + assert calls == ["DemoPlugin", "DemoPlugin"] + + +@pytest.mark.asyncio +async def test_cancelled_install_waits_for_rollback_before_releasing_lifecycle() -> None: + """取消安装后先完成包快照补偿，再允许同一插件的新调用进入。""" + install_started = asyncio.Event() + release_install = asyncio.Event() + rollback = AsyncMock() + + async def install(*_args): + install_started.set() + await release_install.wait() + return True, "ok" + + command = _command(installer=install, rollback=rollback) + task = asyncio.create_task( + command.execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + ) + await install_started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + rollback.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_cancelled_persisted_list_is_restored_conservatively() -> None: + """清单写入已产生副作用但尚未返回时取消，也必须恢复原清单。""" + persisted: list[list[str]] = [] + writer_started = asyncio.Event() + rollback = AsyncMock() + + async def writer(plugin_ids: list[str]) -> None: + persisted.append(list(plugin_ids)) + if len(persisted) == 1: + writer_started.set() + await asyncio.Event().wait() + + task = asyncio.create_task( + _command(writer=writer, rollback=rollback).execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + ) + await writer_started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert persisted == [["DemoPlugin"], []] + rollback.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_cancelled_snapshot_cleanup_does_not_rollback_committed_plugin() -> None: + """运行态提交后清理快照期间取消，不得删除已生效插件。""" + cleanup_started = asyncio.Event() + rollback = AsyncMock() + + async def committer(_checkpoint) -> None: + cleanup_started.set() + await asyncio.Event().wait() + + task = asyncio.create_task( + _command(committer=committer, rollback=rollback).execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + ) + await cleanup_started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + rollback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_startup_lifecycle_lock_blocks_plugin_install_until_settlement() -> None: + """启动同步持有全局资格时，插件安装不得穿过启动收口。""" + from app.application.plugin.lifecycle import plugin_lifecycle + + entered = asyncio.Event() + release = asyncio.Event() + + async def startup_scope(): + async with plugin_lifecycle.hold_startup(): + entered.set() + await release.wait() + + startup = asyncio.create_task(startup_scope()) + await entered.wait() + plugin_context = plugin_lifecycle.hold("DemoPlugin") + plugin_scope = asyncio.create_task(plugin_context.__aenter__()) + await asyncio.sleep(0.02) + assert plugin_scope.done() is False + + release.set() + await plugin_scope + await plugin_context.__aexit__(None, None, None) + await startup + + @pytest.mark.asyncio async def test_report_failure_does_not_rollback_completed_local_install(): """统计上报失败属于非关键副作用，不得撤销已成功的本地安装。""" diff --git a/tests/test_plugin_monitor_lifecycle.py b/tests/test_plugin_monitor_lifecycle.py index c76952a14..11a621ed1 100644 --- a/tests/test_plugin_monitor_lifecycle.py +++ b/tests/test_plugin_monitor_lifecycle.py @@ -2,7 +2,7 @@ import asyncio import threading import time from types import SimpleNamespace -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest @@ -139,6 +139,12 @@ def _patch_sync_plugins(monkeypatch, manager: MagicMock) -> MagicMock: monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager) monkeypatch.setattr(plugins_initializer, "execute_task", execute) monkeypatch.setattr(plugins_initializer, "register_plugin_api", register) + dependency_result = ( + manager.async_install_plugin_missing_dependencies_with_status.return_value + ) + manager.async_install_plugin_missing_dependencies_with_status = AsyncMock( + return_value=dependency_result, + ) manager.get_plugin_runtime_statuses.return_value = {} return register @@ -150,7 +156,7 @@ async def test_sync_plugins_activates_ready_plugins_when_dependencies_fail( """依赖恢复失败时仍激活无关的已就绪插件。""" manager = MagicMock() manager.sync.return_value = ["demo"] - manager.install_plugin_missing_dependencies_with_status.return_value = ( + manager.async_install_plugin_missing_dependencies_with_status.return_value = ( PluginDependencyInstallResult(missing=["demo>=1"], success=False) ) manager.classify_plugins.return_value = PluginDependencyClassification( @@ -175,7 +181,7 @@ async def test_sync_plugins_loads_only_plugins_that_become_ready( """后台依赖恢复后只启动尚未运行且当前已就绪的插件。""" manager = MagicMock() manager.sync.return_value = [] - manager.install_plugin_missing_dependencies_with_status.return_value = ( + manager.async_install_plugin_missing_dependencies_with_status.return_value = ( PluginDependencyInstallResult(missing=["demo>=1"], success=True) ) manager.classify_plugins.return_value = PluginDependencyClassification( @@ -204,7 +210,7 @@ async def test_sync_plugins_reloads_only_updated_running_plugins(monkeypatch) -> """源码同步只重载对应运行实例，不重启其他插件。""" manager = MagicMock() manager.sync.return_value = ["UpdatedPlugin"] - manager.install_plugin_missing_dependencies_with_status.return_value = ( + manager.async_install_plugin_missing_dependencies_with_status.return_value = ( PluginDependencyInstallResult(missing=[], success=True) ) manager.classify_plugins.return_value = PluginDependencyClassification( @@ -232,7 +238,7 @@ async def test_sync_plugins_reloads_running_plugin_after_dependency_recovery( """依赖恢复后，已运行的旧实例必须切换到新源码。""" manager = MagicMock() manager.sync.return_value = [] - manager.install_plugin_missing_dependencies_with_status.return_value = ( + manager.async_install_plugin_missing_dependencies_with_status.return_value = ( PluginDependencyInstallResult(missing=["demo>=1"], success=True) ) manager.classify_plugins.return_value = PluginDependencyClassification( @@ -258,7 +264,7 @@ async def test_sync_plugins_keeps_runtime_when_nothing_changed(monkeypatch) -> N """源码和依赖均无变化时保留首次初始化结果。""" manager = MagicMock() manager.sync.return_value = [] - manager.install_plugin_missing_dependencies_with_status.return_value = ( + manager.async_install_plugin_missing_dependencies_with_status.return_value = ( PluginDependencyInstallResult(missing=[], success=True) ) manager.classify_plugins.return_value = PluginDependencyClassification( @@ -283,7 +289,7 @@ async def test_sync_plugins_keeps_event_loop_responsive_during_activation( """插件初始化运行在线程池时，Web 事件循环仍可继续调度。""" manager = MagicMock() manager.sync.return_value = [] - manager.install_plugin_missing_dependencies_with_status.return_value = ( + manager.async_install_plugin_missing_dependencies_with_status.return_value = ( PluginDependencyInstallResult(missing=[], success=True) ) manager.classify_plugins.return_value = PluginDependencyClassification( @@ -299,6 +305,9 @@ async def test_sync_plugins_keeps_event_loop_responsive_during_activation( time.sleep(0.1) manager.start.side_effect = slow_start + manager.async_install_plugin_missing_dependencies_with_status = AsyncMock( + return_value=PluginDependencyInstallResult(missing=[], success=True), + ) monkeypatch.setattr(plugins_initializer, "configure_plugin_services", lambda: None) monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager) monkeypatch.setattr(plugins_initializer, "register_plugin_api", MagicMock()) diff --git a/tests/test_plugin_package_manager.py b/tests/test_plugin_package_manager.py index 03fa7dbdc..ad48f91c4 100644 --- a/tests/test_plugin_package_manager.py +++ b/tests/test_plugin_package_manager.py @@ -1,7 +1,10 @@ +import shutil from pathlib import Path from types import SimpleNamespace from unittest.mock import Mock +import pytest + from app.adapters.system.plugin.package import PluginPackageManager @@ -46,6 +49,23 @@ def test_checkpoint_rollback_removes_new_package(monkeypatch, tmp_path): assert not checkpoint.transaction_dir.exists() +def test_rollback_does_not_delete_package_when_snapshot_is_missing(monkeypatch, tmp_path): + """补偿快照损坏时先失败，不能先删除当前可用插件。""" + manager = _manager(monkeypatch, tmp_path) + plugin_dir = tmp_path / "app" / "plugins" / "demoplugin" + plugin_dir.mkdir(parents=True) + (plugin_dir / "__init__.py").write_text("old", encoding="utf-8") + + checkpoint = manager.checkpoint("DemoPlugin") + shutil.rmtree(checkpoint.transaction_dir / "package") + (plugin_dir / "__init__.py").write_text("new", encoding="utf-8") + + with pytest.raises(FileNotFoundError): + manager.rollback(checkpoint) + + assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "new" + + def test_local_sync_failure_restores_previous_runtime_copy(monkeypatch, tmp_path): """本地来源不可复制时不得丢失已经运行的插件副本。""" manager = _manager(monkeypatch, tmp_path) diff --git a/tests/test_system_utils.py b/tests/test_system_utils.py index 374cd87d1..df4bd6107 100644 --- a/tests/test_system_utils.py +++ b/tests/test_system_utils.py @@ -1,13 +1,17 @@ +import asyncio import errno import itertools import os import struct import subprocess +import sys import tempfile +import time from pathlib import Path from unittest import TestCase from unittest.mock import MagicMock, call, patch +import psutil import pytest from app.runtime.state import SystemHelper @@ -155,6 +159,148 @@ def test_execute_with_subprocess_uses_safe_command_in_failure_message(): assert run_mock.call_args.args[0] == command +@pytest.mark.asyncio +async def test_async_subprocess_timeout_reaps_process(): + """异步安装命令超时后应终止并回收子进程。""" + success, message = await SystemUtils.execute_with_subprocess_async( + [sys.executable, "-c", "import time; time.sleep(60)"], + timeout=0.05, + ) + + assert success is False + assert "执行超时" in message + + +@pytest.mark.asyncio +async def test_async_subprocess_cancellation_reaps_process(tmp_path): + """调用方取消安装任务时，底层子进程不得继续运行。""" + marker = tmp_path / "pid" + command = [ + sys.executable, + "-c", + ( + "from pathlib import Path; import os, time; " + f"Path({str(marker)!r}).write_text(str(os.getpid())); time.sleep(60)" + ), + ] + task = asyncio.create_task( + SystemUtils.execute_with_subprocess_async(command, timeout=30) + ) + deadline = time.monotonic() + 2 + while not marker.exists() and time.monotonic() < deadline: + await asyncio.sleep(0.01) + assert marker.exists() + + pid = int(marker.read_text()) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + for _ in range(100): + try: + os.kill(pid, 0) + except ProcessLookupError: + break + await asyncio.sleep(0.01) + else: + pytest.fail(f"子进程仍在运行：{pid}") + + +@pytest.mark.asyncio +async def test_async_subprocess_cancellation_reaps_process_tree(tmp_path): + """取消安装命令时，子进程派生的构建进程也不得继续运行。""" + marker = tmp_path / "pids" + child_code = "import time; time.sleep(60)" + command = [ + sys.executable, + "-c", + ( + "from pathlib import Path; import subprocess, os, time; " + f"child = subprocess.Popen([{sys.executable!r}, '-c', {child_code!r}]); " + f"Path({str(marker)!r}).write_text(str(os.getpid()) + ':' + str(child.pid)); " + "time.sleep(60)" + ), + ] + task = asyncio.create_task( + SystemUtils.execute_with_subprocess_async(command, timeout=30) + ) + deadline = time.monotonic() + 2 + while not marker.exists() and time.monotonic() < deadline: + await asyncio.sleep(0.01) + assert marker.exists() + + pids = [int(value) for value in marker.read_text().split(":")] + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + for _ in range(100): + alive = [] + for pid in pids: + try: + process = psutil.Process(pid) + if not process.is_running() or process.status() == psutil.STATUS_ZOMBIE: + continue + except (psutil.Error, OSError): + continue + alive.append(pid) + if not alive: + break + await asyncio.sleep(0.01) + else: + pytest.fail(f"进程树仍在运行：{alive}") + + +@pytest.mark.skipif(os.name == "nt", reason="Windows 没有 POSIX 进程组信号语义") +@pytest.mark.asyncio +async def test_async_subprocess_reaps_descendant_after_early_pipe_close(tmp_path): + """父进程关闭管道后，忽略终止信号的后代也必须被强制回收。""" + marker = tmp_path / "pids" + child_code = ( + "import os, signal, time; os.close(1); os.close(2); " + "signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(60)" + ) + command = [ + sys.executable, + "-c", + ( + "from pathlib import Path; import os, signal, subprocess, time; " + f"child = subprocess.Popen([{sys.executable!r}, '-c', {child_code!r}], " + "start_new_session=True); " + f"Path({str(marker)!r}).write_text(str(os.getpid()) + ':' + str(child.pid)); " + "signal.signal(signal.SIGTERM, lambda *_: os._exit(0)); time.sleep(60)" + ), + ] + task = asyncio.create_task( + SystemUtils.execute_with_subprocess_async(command, timeout=30) + ) + deadline = time.monotonic() + 2 + while not marker.exists() and time.monotonic() < deadline: + await asyncio.sleep(0.01) + assert marker.exists() + + pids = [int(value) for value in marker.read_text().split(":")] + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + for _ in range(100): + alive = [] + for pid in pids: + try: + process = psutil.Process(pid) + if not process.is_running() or process.status() == psutil.STATUS_ZOMBIE: + continue + except (psutil.Error, OSError): + continue + alive.append(pid) + if not alive: + break + await asyncio.sleep(0.01) + else: + pytest.fail(f"通信已结束但进程树仍在运行：{alive}") + + def test_execute_with_subprocess_redacts_userinfo_from_stdout_and_stderr(): error = subprocess.CalledProcessError( returncode=1,