diff --git a/app/adapters/external/market.py b/app/adapters/external/market.py index 3b5b61a79..ecf98d20e 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] = {} @@ -2421,18 +2435,279 @@ 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 repair_target.exists(): + repair_target = None + if repair_target is None: + repair_target = settings.ROOT_PATH / "pyproject.toml" + repair_desc = "主程序 uv.lock" + if not repair_target.exists(): + return False, f"恢复依赖文件不存在：{repair_target}" + if snapshot_file is None and not (settings.ROOT_PATH / "uv.lock").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 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 candidate_path.is_dir(): + continue + candidate_key = str(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 _await_thread_operation( + cls.__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: + constraints_file.unlink(missing_ok=True) async def __async_backup_plugin(self, pid: str) -> str: """ @@ -2444,14 +2719,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 +2835,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 +2945,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..e120840f5 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,136 @@ 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: + """终止安装子进程及其同组子进程，并确保句柄已回收。""" + if process.returncode is None: + try: + if os.name == "nt": + 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): + try: + if os.name == "nt": + process.kill() + else: + os.killpg(process.pid, signal.SIGKILL) + except (ProcessLookupError, OSError): + pass + try: + await asyncio.wait_for( + asyncio.shield(communication_task), + timeout=grace_seconds, + ) + except (asyncio.TimeoutError, asyncio.CancelledError): + communication_task.cancel() + await asyncio.gather(communication_task, return_exceptions=True) + finally: + try: + await process.wait() + except (ProcessLookupError, OSError): + pass + + @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..f69de831b 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,7 +96,7 @@ 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: @@ -101,7 +114,7 @@ class PluginPackageManager: 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/application/plugin/install.py b/app/application/plugin/install.py index 90eddd7c7..ff844934b 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.application.plugin.lifecycle import plugin_lifecycle +from app.runtime.log import logger InstalledPluginsReader = Callable[[], list[str]] @@ -58,6 +61,20 @@ class PluginInstallResult: rollback: PluginInstallRollback = field(default_factory=PluginInstallRollback) +@dataclass +class _InstallState: + """记录取消补偿所需的事务阶段。""" + + checkpoint: Any = None + stage: str = "package_checkpoint" + package_installed: bool = False + installed_list_persisted: bool = False + runtime_touched: bool = False + registrations_touched: bool = False + committed: bool = False + original_plugins: list[str] = field(default_factory=list) + + class PluginInstallCommand: """协调插件检查、包事务、持久化、运行态刷新和安装上报。""" @@ -96,9 +113,38 @@ 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( @@ -112,8 +158,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 +175,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 +196,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, @@ -156,6 +212,7 @@ class PluginInstallCommand: try: 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, @@ -169,6 +226,8 @@ class PluginInstallCommand: raise return result + state.stage = "runtime_reload" + state.runtime_touched = True try: await self._plugin_reloader(plugin_id) except Exception as err: @@ -186,6 +245,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,13 +266,16 @@ class PluginInstallCommand: return result checkpoint_cleanup_error = "" + state.stage = "checkpoint_commit" try: await self._package_committer(checkpoint) + state.committed = True except Exception as err: checkpoint_cleanup_error = str(err) 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,6 +301,57 @@ 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.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_persisted, + 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, *, diff --git a/app/application/plugin/lifecycle.py b/app/application/plugin/lifecycle.py new file mode 100644 index 000000000..587979448 --- /dev/null +++ b/app/application/plugin/lifecycle.py @@ -0,0 +1,88 @@ +"""插件安装与启动同步之间共享的生命周期互斥。""" + +from __future__ import annotations + +import asyncio +import threading +from contextlib import asynccontextmanager, contextmanager +from typing import Iterator + + +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) + + @contextmanager + def hold_sync(self, plugin_id: str) -> Iterator[None]: + """同步持有单个插件的生命周期资格。""" + normalized_id = self._normalize(plugin_id) + if not normalized_id: + raise ValueError("插件ID不能为空") + with self._condition: + while self._startup_active or normalized_id in self._active_plugins: + self._condition.wait() + self._active_plugins.add(normalized_id) + try: + yield + finally: + self._release_plugin(normalized_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/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index de01e19a8..f6c1cc7c4 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 @@ -76,13 +77,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/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index 987d5ea20..ec69cac6c 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,70 @@ 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_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..7ae5c4691 100644 --- a/tests/test_plugin_install_command.py +++ b/tests/test_plugin_install_command.py @@ -1,3 +1,4 @@ +import asyncio from unittest.mock import AsyncMock, Mock import pytest @@ -306,6 +307,96 @@ 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_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_system_utils.py b/tests/test_system_utils.py index 374cd87d1..d7fe0d67b 100644 --- a/tests/test_system_utils.py +++ b/tests/test_system_utils.py @@ -1,9 +1,12 @@ +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 @@ -155,6 +158,53 @@ 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}") + + def test_execute_with_subprocess_redacts_userinfo_from_stdout_and_stderr(): error = subprocess.CalledProcessError( returncode=1,