From a7965d3f7d35a541d6e35187c9ad8cb20b05899e Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 07:16:07 +0800 Subject: [PATCH] fix(plugin): close process tree cleanup gaps --- app/adapters/system/host.py | 92 ++++++++++++++++++++++------ app/application/plugin/install.py | 1 + tests/test_plugin_install_command.py | 32 +++++++++- tests/test_system_utils.py | 50 +++++++++++++++ 4 files changed, 153 insertions(+), 22 deletions(-) diff --git a/app/adapters/system/host.py b/app/adapters/system/host.py index ec1d9ace9..d7cb00870 100644 --- a/app/adapters/system/host.py +++ b/app/adapters/system/host.py @@ -205,28 +205,48 @@ class SystemUtils: timeout=grace_seconds, ) except (asyncio.TimeoutError, asyncio.CancelledError): - try: - if os.name == "nt": - SystemUtils._signal_process_tree(process_tree, force=True) + 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() - else: + except (ProcessLookupError, OSError): + pass + else: + try: os.killpg(process.pid, signal.SIGKILL) - SystemUtils._signal_process_tree(process_tree, force=True) - 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 + 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]: @@ -237,6 +257,38 @@ class SystemUtils: 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], diff --git a/app/application/plugin/install.py b/app/application/plugin/install.py index 14ef8ebf4..c33f40bd3 100644 --- a/app/application/plugin/install.py +++ b/app/application/plugin/install.py @@ -224,6 +224,7 @@ class PluginInstallCommand: stage="installed_list_persistence", message=str(err), package_installed=True, + installed_list_persisted=state.installed_list_touched, ) if isinstance(err, DatabaseWorkerOverloadedError): raise diff --git a/tests/test_plugin_install_command.py b/tests/test_plugin_install_command.py index f7c90c9b8..7d8112abf 100644 --- a/tests/test_plugin_install_command.py +++ b/tests/test_plugin_install_command.py @@ -3,7 +3,6 @@ from unittest.mock import AsyncMock, Mock import pytest -from app.application.database import DatabaseWorkerOverloadedError from app.application.plugin.install import PluginInstallCommand @@ -177,12 +176,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。""" diff --git a/tests/test_system_utils.py b/tests/test_system_utils.py index d899989e7..df4bd6107 100644 --- a/tests/test_system_utils.py +++ b/tests/test_system_utils.py @@ -251,6 +251,56 @@ async def test_async_subprocess_cancellation_reaps_process_tree(tmp_path): 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,