From 8538a94ddc8d9abf9ce5a3d8196de28d3545f8b8 Mon Sep 17 00:00:00 2001 From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com> Date: Fri, 9 Oct 2026 11:44:38 -0400 Subject: [PATCH 1/5] feat: Windows support via Job Objects On Windows only qsv.exe itself was killed on a timeout or cancellation; everything it started survived. Each run is now assigned to its own Job Object (ctypes, no new dependency): CTRL_BREAK_EVENT to its console process group first, then TerminateJobObject after kill_grace. Children cannot leave the job, unlike a POSIX process group. - _process.py: a _ProcessTree per run (POSIX session or Windows job) with interrupt/kill/close; the teardown logic is now shared. - client.py: keep SYSTEMROOT with inherit_env=False on Windows. - Tests: the fake qsv is a .py behind a .cmd wrapper on Windows; process liveness uses GetExitCodeProcess (os.kill(pid, 0) terminates there); escaped-grandchild tests assert the job killed it. - CI: full suite on Windows for Python 3.10-3.14; lint also runs mypy --platform win32. - Docs: README platform notes, CHANGELOG, Windows classifier. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 10 +- CHANGELOG.md | 8 ++ CLAUDE.md | 9 +- README.md | 11 ++- pyproject.toml | 1 + src/qsv_client/_process.py | 183 +++++++++++++++++++++++++------------ src/qsv_client/client.py | 9 +- tests/conftest.py | 18 +++- tests/test_process.py | 96 +++++++++++++++---- 9 files changed, 254 insertions(+), 91 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 46043bf..c578bb2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,6 +18,7 @@ jobs: - run: uv run ruff check . - run: uv run ruff format --check . - run: uv run mypy + - run: uv run mypy --platform win32 test: strategy: @@ -50,13 +51,12 @@ jobs: if: runner.os == 'Linux' run: QSV_BIN="$RUNNER_TEMP/qsv/qsvdp" uv run pytest -q tests/test_integration.py - # Integration tests only: the unit tests' fake qsv relies on a shebang and skips on Windows, - # so process-group cleanup is not exercised here. + # Separate from `test` because Windows steps run in PowerShell and use the MSVC build. test-windows: strategy: fail-fast: false matrix: - python: ["3.10", "3.14"] + python: ["3.10", "3.11", "3.12", "3.13", "3.14"] runs-on: windows-latest defaults: run: @@ -76,10 +76,10 @@ jobs: gh release download --repo dathere/qsv --pattern "qsv-*-x86_64-pc-windows-msvc.zip" --dir $dir Get-ChildItem $dir -Filter *.zip | ForEach-Object { Expand-Archive $_.FullName -DestinationPath $dir -Force } Get-ChildItem $dir - - name: Integration tests (qsv) + - name: Unit tests and integration tests (qsv) run: | $env:QSV_BIN = Join-Path $env:RUNNER_TEMP "qsv\qsv.exe" - uv run pytest -v tests/test_integration.py + uv run pytest -v - name: Integration tests (qsvlite) run: | $env:QSV_BIN = Join-Path $env:RUNNER_TEMP "qsv\qsvlite.exe" diff --git a/CHANGELOG.md b/CHANGELOG.md index bba0f8f..891e649 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ All notable changes to this project are documented here. The format follows ## [Unreleased] +### Added + +- Windows support. Each run is placed in its own Job Object, so a timeout or cancellation + stops qsv *and* everything it started: `CTRL_BREAK_EVENT` first, then `TerminateJobObject` + after `kill_grace`. Previously only `qsv.exe` itself was killed. With `inherit_env=False`, + `SYSTEMROOT` is still passed on Windows, since many programs fail to start without it. CI + now runs the full test suite on Windows. + ### Changed - The sdist now contains only the package, tests, README, CHANGELOG and LICENSE. 0.1.0's also diff --git a/CLAUDE.md b/CLAUDE.md index f1aa58d..2830947 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,7 +15,9 @@ uv run ruff format . && uv run ruff check . && uv run mypy # must be CI runs lint, format check, and mypy, plus tests on ubuntu/macos x Python 3.10–3.14 against qsv, qsvlite, and qsvdp (Linux only). -On Windows (Python 3.10 and 3.14) it runs only the integration tests, against qsv and qsvlite. +On Windows (Python 3.10–3.14) it runs the full suite against qsv, plus the integration tests +against qsvlite. Windows teardown uses Job Objects via ctypes (`_ProcessTree` in `_process.py`); +type-check it with `uv run mypy --platform win32` (the lint job runs both). ## Layout (`src/qsv_client/`) @@ -32,8 +34,9 @@ On Windows (Python 3.10 and 3.14) it runs only the integration tests, against qs - `_Base._build_env` always sets `QSV_ERROR_FORMAT=json`. Older qsv versions ignore it, so their errors are classified by exit code. - Exit codes 0 and 255 (qsv's "warning") count as success. -- On timeout, cancellation, or KeyboardInterrupt, the whole process group is killed: SIGTERM, - then SIGKILL after `kill_grace`. A timeout raises `QsvTimeout` (exit code 124) even when +- On timeout, cancellation, or KeyboardInterrupt, the run's whole process tree is killed after + `kill_grace`: SIGTERM then SIGKILL to the process group on POSIX, `CTRL_BREAK_EVENT` then + `TerminateJobObject` on Windows. A timeout raises `QsvTimeout` (exit code 124) even when `check=False`. - LLM settings for describegpt are passed as env vars (`QSV_LLM_*`), never on argv. - `min_version` is enforced inside `_probe`, so it holds however the capabilities cache is diff --git a/README.md b/README.md index 0c23bac..e8fc81e 100644 --- a/README.md +++ b/README.md @@ -118,8 +118,15 @@ runs. ## Platform notes -Process-group cleanup uses POSIX sessions (Linux, macOS). On Windows, qsv is started in a new -process group and killed directly; its own children are not reaped. +On a timeout or cancellation, the client stops qsv and every process qsv started: + +- **Linux, macOS:** each run gets its own session. The client sends SIGTERM to the process + group, then SIGKILL after `kill_grace`. A process that leaves the group (by calling `setsid`) + survives. +- **Windows:** each run goes in its own Job Object. The client sends `CTRL_BREAK_EVENT` to + the run's process group, then calls `TerminateJobObject` after `kill_grace`. Child processes + can't leave the job. A child started in the instant between qsv starting and being added to + the job is missed. ## Development diff --git a/pyproject.toml b/pyproject.toml index 501a654..0f3b676 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ classifiers = [ "Intended Audience :: Developers", "Operating System :: POSIX", "Operating System :: MacOS", + "Operating System :: Microsoft :: Windows", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3 :: Only", "Topic :: Scientific/Engineering :: Information Analysis", diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py index 8cec986..4d9ef86 100644 --- a/src/qsv_client/_process.py +++ b/src/qsv_client/_process.py @@ -1,14 +1,17 @@ """Spawn qsv and reap it, including any children it started, on timeout or cancellation. qsv commands can start their own children (``validate`` runs pyshacl, ``viz`` a webdriver, -``describegpt`` may shell out). Killing only the qsv pid would orphan them, so on POSIX each -run gets its own session (process group) and the whole group is signalled: SIGTERM first, -then SIGKILL after a grace period. On Windows the process is started in a new process group -and killed directly; grandchildren are not reaped there. - -A descendant that leaves the group (e.g. by calling ``setsid``) survives the kill and may keep -the output pipes open; after a timeout, output is collected for at most ``kill_grace`` seconds -and anything that descendant still holds is dropped. +``describegpt`` may shell out). Killing only the qsv pid would orphan them, so each run's +process tree is tracked and stopped as a whole: first politely, then forcibly after a grace +period. + +- POSIX: each run gets its own session (process group). SIGTERM, then SIGKILL, to the group. + A descendant that leaves the group (e.g. by calling ``setsid``) survives the kill and may + keep the output pipes open; after a timeout, output is collected for at most ``kill_grace`` + seconds and anything that descendant still holds is dropped. +- Windows: each run is put in its own Job Object, which every process it starts joins and + cannot leave. ``CTRL_BREAK_EVENT`` to its process group, then ``TerminateJobObject``. A + child started in the instant between spawning qsv and assigning it to the job is missed. """ from __future__ import annotations @@ -36,27 +39,95 @@ class RawResult: timed_out: bool -def _spawn_kwargs() -> dict[str, Any]: - if IS_WINDOWS: - return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} # type: ignore[attr-defined,unused-ignore] - return {"start_new_session": True} - - -def _signal_group(pid: int, sig: int) -> None: - with contextlib.suppress(ProcessLookupError, PermissionError): - os.killpg(pid, sig) # type: ignore[attr-defined,unused-ignore] - - -def _terminate_sync(proc: subprocess.Popen[bytes], grace: float) -> None: - if IS_WINDOWS: - with contextlib.suppress(OSError): - proc.kill() - return - _signal_group(proc.pid, signal.SIGTERM) +if sys.platform == "win32": + import ctypes + from ctypes import wintypes + + _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + _kernel32.CreateJobObjectW.argtypes = (wintypes.LPVOID, wintypes.LPCWSTR) + _kernel32.CreateJobObjectW.restype = wintypes.HANDLE + _kernel32.OpenProcess.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD) + _kernel32.OpenProcess.restype = wintypes.HANDLE + _kernel32.AssignProcessToJobObject.argtypes = (wintypes.HANDLE, wintypes.HANDLE) + _kernel32.AssignProcessToJobObject.restype = wintypes.BOOL + _kernel32.TerminateJobObject.argtypes = (wintypes.HANDLE, wintypes.UINT) + _kernel32.TerminateJobObject.restype = wintypes.BOOL + _kernel32.CloseHandle.argtypes = (wintypes.HANDLE,) + _kernel32.CloseHandle.restype = wintypes.BOOL + + _PROCESS_TERMINATE = 0x0001 + _PROCESS_SET_QUOTA = 0x0100 + + def _spawn_kwargs() -> dict[str, Any]: + # its own console process group, so CTRL_BREAK_EVENT reaches qsv and its children only + return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} + + def _create_job(pid: int) -> int | None: + """A Job Object holding ``pid`` (and, from now on, everything it starts), or None.""" + job = _kernel32.CreateJobObjectW(None, None) + if not job: + return None + handle = _kernel32.OpenProcess(_PROCESS_SET_QUOTA | _PROCESS_TERMINATE, False, pid) + try: + if handle and _kernel32.AssignProcessToJobObject(job, handle): + return int(job) + finally: + if handle: + _kernel32.CloseHandle(handle) + _kernel32.CloseHandle(job) + return None + + class _ProcessTree: + def __init__(self, pid: int) -> None: + self.pid = pid + self._job = _create_job(pid) + + def interrupt(self) -> None: + with contextlib.suppress(OSError): + os.kill(self.pid, signal.CTRL_BREAK_EVENT) + + def kill(self) -> None: + if self._job is not None and _kernel32.TerminateJobObject(self._job, 1): + return + # no job (creation failed): the leader is all we can reach. os.kill on Windows + # is TerminateProcess for anything but the console events. + with contextlib.suppress(OSError): + os.kill(self.pid, signal.SIGTERM) + + def close(self) -> None: + if self._job is not None: + _kernel32.CloseHandle(self._job) + self._job = None + +else: + + def _spawn_kwargs() -> dict[str, Any]: + return {"start_new_session": True} + + def _signal_group(pid: int, sig: int) -> None: + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(pid, sig) + + class _ProcessTree: + def __init__(self, pid: int) -> None: + self.pid = pid + + def interrupt(self) -> None: + _signal_group(self.pid, signal.SIGTERM) + + def kill(self) -> None: + # the whole group, even if the leader already exited: its children may not have + _signal_group(self.pid, signal.SIGKILL) + + def close(self) -> None: + pass + + +def _terminate_sync(proc: subprocess.Popen[bytes], tree: _ProcessTree, grace: float) -> None: + tree.interrupt() with contextlib.suppress(subprocess.TimeoutExpired): proc.wait(timeout=grace) - # SIGKILL the group even if the leader already exited: its children may not have - _signal_group(proc.pid, signal.SIGKILL) # type: ignore[attr-defined,unused-ignore] + tree.kill() def run_sync( @@ -78,12 +149,13 @@ def run_sync( cwd=cwd, **_spawn_kwargs(), ) + tree = _ProcessTree(proc.pid) timed_out = False try: out, err = proc.communicate(input=stdin, timeout=timeout) except subprocess.TimeoutExpired: timed_out = True - _terminate_sync(proc, kill_grace) + _terminate_sync(proc, tree, kill_grace) try: out, err = proc.communicate(timeout=kill_grace) except subprocess.TimeoutExpired: @@ -94,10 +166,12 @@ def run_sync( proc.wait() out, err = b"", b"" except BaseException: - # KeyboardInterrupt etc.: never leave a qsv process group running behind us - _terminate_sync(proc, kill_grace) + # KeyboardInterrupt etc.: never leave a qsv process tree running behind us + _terminate_sync(proc, tree, kill_grace) proc.wait() raise + finally: + tree.close() return RawResult( exit_code=proc.returncode, stdout=out or b"", @@ -106,19 +180,15 @@ def run_sync( ) -async def _terminate_async(proc: asyncio.subprocess.Process, grace: float) -> None: - if IS_WINDOWS: - with contextlib.suppress(ProcessLookupError): - proc.kill() - with contextlib.suppress(asyncio.TimeoutError): - await asyncio.wait_for(proc.wait(), grace) - return +async def _terminate_async( + proc: asyncio.subprocess.Process, tree: _ProcessTree, grace: float +) -> None: # proc.wait() also waits for the output pipes to close, which a descendant that left the # group can hold open indefinitely, so every wait here is bounded - _signal_group(proc.pid, signal.SIGTERM) + tree.interrupt() with contextlib.suppress(asyncio.TimeoutError): await asyncio.wait_for(proc.wait(), grace) - _signal_group(proc.pid, signal.SIGKILL) # type: ignore[attr-defined,unused-ignore] + tree.kill() with contextlib.suppress(asyncio.TimeoutError): await asyncio.wait_for(proc.wait(), grace) @@ -139,37 +209,35 @@ def _close_pipes(proc: asyncio.subprocess.Process) -> None: transport.close() -def _kill_now(proc: asyncio.subprocess.Process) -> None: +def _kill_now(proc: asyncio.subprocess.Process, tree: _ProcessTree) -> None: """Teardown was itself cancelled (e.g. by ``asyncio.run()`` shutting down): skip the remaining grace and do the parts that matter synchronously.""" - if IS_WINDOWS: - with contextlib.suppress(ProcessLookupError): - proc.kill() - else: - _signal_group(proc.pid, signal.SIGKILL) # type: ignore[attr-defined,unused-ignore] + tree.kill() _close_pipes(proc) -async def _reap_async(proc: asyncio.subprocess.Process, grace: float) -> None: +async def _reap_async(proc: asyncio.subprocess.Process, tree: _ProcessTree, grace: float) -> None: try: - await _terminate_async(proc, grace) + await _terminate_async(proc, tree, grace) await _release_async(proc) except asyncio.CancelledError: - _kill_now(proc) + _kill_now(proc, tree) raise -async def _timeout_cleanup(proc: asyncio.subprocess.Process, grace: float) -> bytes: +async def _timeout_cleanup( + proc: asyncio.subprocess.Process, tree: _ProcessTree, grace: float +) -> bytes: """Kill and release a timed-out run; return whatever stderr arrives within ``grace``.""" try: - await _terminate_async(proc, grace) + await _terminate_async(proc, tree, grace) err = b"" if proc.stderr is not None: with contextlib.suppress(Exception): err = await asyncio.wait_for(proc.stderr.read(), grace) await _release_async(proc) except asyncio.CancelledError: - _kill_now(proc) + _kill_now(proc, tree) raise return err @@ -179,7 +247,7 @@ async def _finish_despite_cancel(coro: Coroutine[Any, Any, _T]) -> _T: ``asyncio.shield`` alone would propagate the cancellation at once, leaving the cleanup in a background task that an ``asyncio.run()`` returning right after would cancel, skipping - the SIGKILL. + the forced kill. """ task = asyncio.ensure_future(coro) cancel: asyncio.CancelledError | None = None @@ -212,17 +280,20 @@ async def run_async( cwd=cwd, **_spawn_kwargs(), ) + tree = _ProcessTree(proc.pid) timed_out = False try: out, err = await asyncio.wait_for(proc.communicate(input=stdin), timeout) except asyncio.TimeoutError: timed_out = True - # a cancellation arriving mid-cleanup must not skip the SIGKILL or the release - out, err = b"", await _finish_despite_cancel(_timeout_cleanup(proc, kill_grace)) + # a cancellation arriving mid-cleanup must not skip the forced kill or the release + out, err = b"", await _finish_despite_cancel(_timeout_cleanup(proc, tree, kill_grace)) except BaseException: - # task cancelled (or the loop is shutting down): reap the group, then propagate - await _finish_despite_cancel(_reap_async(proc, kill_grace)) + # task cancelled (or the loop is shutting down): reap the tree, then propagate + await _finish_despite_cancel(_reap_async(proc, tree, kill_grace)) raise + finally: + tree.close() assert proc.returncode is not None return RawResult( exit_code=proc.returncode, diff --git a/src/qsv_client/client.py b/src/qsv_client/client.py index 9a8d64d..dd5515d 100644 --- a/src/qsv_client/client.py +++ b/src/qsv_client/client.py @@ -118,13 +118,15 @@ def __init__( whole process group is sent SIGTERM, then SIGKILL after ``kill_grace`` seconds, and :class:`QsvTimeout` is raised. env: extra environment variables for every run. - inherit_env: start from ``os.environ`` (True) or from an empty environment. + inherit_env: start from ``os.environ`` (True) or from an empty environment + (on Windows, one holding only ``SYSTEMROOT``). cwd: working directory for every run. llm_api_key / llm_base_url / llm_model: passed to ``describegpt`` as ``QSV_LLM_APIKEY`` / ``QSV_LLM_BASE_URL`` / ``QSV_LLM_MODEL`` environment variables, so the key never appears in the process table. min_version: raise :class:`QsvVersionError` on first use if the binary is older. - kill_grace: seconds to wait after SIGTERM before SIGKILL on timeout or + kill_grace: seconds to wait between the polite stop (SIGTERM, or + ``CTRL_BREAK_EVENT`` on Windows) and the forced kill on timeout or cancellation, and the most spent collecting output after the kill. """ self.binary = find_qsv(binary) @@ -148,6 +150,9 @@ def __repr__(self) -> str: def _build_env(self, extra: Mapping[str, str] | None) -> dict[str, str]: env = dict(os.environ) if self.inherit_env else {} + if not self.inherit_env and _process.IS_WINDOWS and "SYSTEMROOT" in os.environ: + # Windows processes (the C runtime, Winsock) can fail to start without it + env["SYSTEMROOT"] = os.environ["SYSTEMROOT"] env.update(self._env) if extra: env.update(extra) diff --git a/tests/conftest.py b/tests/conftest.py index 10c9454..b9b4e12 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -40,13 +40,23 @@ def csv_files(tmp_path: Path) -> Path: @pytest.fixture def fake_qsv(tmp_path: Path) -> Callable[[str], str]: - """Write an executable stand-in for qsv running the given Python body; return its path.""" - if sys.platform == "win32": - pytest.skip("fake executables use a shebang") + """Write an executable stand-in for qsv running the given Python body; return its path. + + POSIX: a script with a shebang. Windows: the script plus a ``fake-qsv.cmd`` wrapper that + runs it (the extra ``cmd.exe`` is inside the run's Job Object, so it is killed with it). + """ + # qsv writes "\n" on every platform; without this, print() writes "\r\n" on Windows + prelude = 'import sys; sys.stdout.reconfigure(newline="\\n")\n' def make(body: str) -> str: + script = prelude + textwrap.dedent(body) + if sys.platform == "win32": + (tmp_path / "fake-qsv.py").write_text(script) + wrapper = tmp_path / "fake-qsv.cmd" + wrapper.write_text(f'@"{sys.executable}" "%~dp0fake-qsv.py" %*\n') + return str(wrapper) path = tmp_path / "fake-qsv" - path.write_text(f"#!{sys.executable}\n" + textwrap.dedent(body)) + path.write_text(f"#!{sys.executable}\n" + script) path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) return str(path) diff --git a/tests/test_process.py b/tests/test_process.py index 9f1532d..784035d 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -6,6 +6,7 @@ import contextlib import os import signal +import sys import time from collections.abc import Callable from pathlib import Path @@ -37,14 +38,46 @@ def _is_zombie(pid: int) -> bool: return stat.rsplit(")", 1)[1].split()[0] == "Z" -def _wait_dead(pid: int, within: float = 5.0) -> bool: - deadline = time.monotonic() + within - while time.monotonic() < deadline: +if sys.platform == "win32": + import ctypes + from ctypes import wintypes + + _k32 = ctypes.WinDLL("kernel32", use_last_error=True) + _k32.OpenProcess.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD) + _k32.OpenProcess.restype = wintypes.HANDLE + _k32.GetExitCodeProcess.argtypes = (wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)) + _k32.GetExitCodeProcess.restype = wintypes.BOOL + _k32.CloseHandle.argtypes = (wintypes.HANDLE,) + _k32.CloseHandle.restype = wintypes.BOOL + _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + _STILL_ACTIVE = 259 + + def _alive(pid: int) -> bool: + # not os.kill(pid, 0): on Windows that is TerminateProcess + handle = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if not handle: + return False + try: + code = wintypes.DWORD() + ok = _k32.GetExitCodeProcess(handle, ctypes.byref(code)) + return bool(ok) and code.value == _STILL_ACTIVE + finally: + _k32.CloseHandle(handle) + +else: + + def _alive(pid: int) -> bool: try: os.kill(pid, 0) except ProcessLookupError: - return True - if _is_zombie(pid): + return False + return not _is_zombie(pid) + + +def _wait_dead(pid: int, within: float = 5.0) -> bool: + deadline = time.monotonic() + within + while time.monotonic() < deadline: + if not _alive(pid): return True time.sleep(0.05) return False @@ -162,18 +195,36 @@ def test_warning_exit_code_is_success(fake_qsv: Callable[[str], str]) -> None: # Like HANG_WITH_GRANDCHILD, but the grandchild leaves the process group, so the group kill # misses it and it keeps the output pipes open. +# ignore the polite stop: SIGTERM on POSIX, CTRL_BREAK_EVENT (SIGBREAK) on Windows +IGNORE_STOP = """ +import signal +signal.signal(signal.SIGTERM, signal.SIG_IGN) +if hasattr(signal, "SIGBREAK"): + signal.signal(signal.SIGBREAK, signal.SIG_IGN) +""" + HANG_WITH_ESCAPED_GRANDCHILD = """ import subprocess, sys, time -child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], - start_new_session=True) +# POSIX: its own session, out of the group kill's reach. Windows: its own console process +# group, out of CTRL_BREAK_EVENT's reach (but still inside the run's Job Object). +escape = ({"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} if sys.platform == "win32" + else {"start_new_session": True}) +child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], **escape) open(sys.argv[-1], "w").write(str(child.pid)) time.sleep(60) """ def _kill(pid: int) -> None: - with contextlib.suppress(ProcessLookupError): - os.kill(pid, signal.SIGKILL) + # on Windows, os.kill with anything but the console events is TerminateProcess + with contextlib.suppress(OSError): + os.kill(pid, getattr(signal, "SIGKILL", signal.SIGTERM)) + + +def _check_job_killed(pid: int) -> None: + """On Windows nothing escapes the Job Object, so the 'escaped' grandchild must be dead.""" + if sys.platform == "win32": + assert _wait_dead(pid), "the Job Object missed a grandchild in its own process group" def test_timeout_bounded_when_grandchild_escapes_group( @@ -186,6 +237,7 @@ def test_timeout_bounded_when_grandchild_escapes_group( with pytest.raises(QsvTimeout): qsv.run("stats", pidfile, timeout=1) assert time.monotonic() - started < 10 + _check_job_killed(_read_pid(pidfile)) finally: _kill(_read_pid(pidfile)) @@ -200,6 +252,7 @@ async def test_async_timeout_bounded_when_grandchild_escapes_group( with pytest.raises(QsvTimeout): await qsv.run("stats", pidfile, timeout=1) assert time.monotonic() - started < 10 + _check_job_killed(_read_pid(pidfile)) finally: _kill(_read_pid(pidfile)) @@ -246,30 +299,31 @@ def test_relative_binary_with_cwd( assert Qsv(f"./{fake.name}", cwd=other).run("x").stdout == "ok\n" +@pytest.mark.skipif(sys.platform == "win32", reason="Win32 collapses `..` before symlinks") def test_relative_binary_through_symlinked_dir( fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: # work/link -> bin, so work/link/../fake-qsv is tmp_path/fake-qsv, not work/fake-qsv - fake_qsv("print('ok')") + fake = Path(fake_qsv("print('ok')")) (tmp_path / "bin").mkdir() work = tmp_path / "work" work.mkdir() (work / "link").symlink_to(tmp_path / "bin") monkeypatch.chdir(work) - assert Qsv("link/../fake-qsv").run("x").stdout == "ok\n" + assert Qsv(f"link/../{fake.name}").run("x").stdout == "ok\n" async def test_async_timeout_bounded_when_leader_ignores_sigterm_and_grandchild_escapes( fake_qsv: Callable[[str], str], tmp_path: Path ) -> None: pidfile = tmp_path / "child.pid" - body = "import signal; signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" - qsv = AsyncQsv(fake_qsv(body + HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.5) + qsv = AsyncQsv(fake_qsv(IGNORE_STOP + HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.5) started = time.monotonic() try: with pytest.raises(QsvTimeout): await qsv.run("stats", pidfile, timeout=1) assert time.monotonic() - started < 10 + _check_job_killed(_read_pid(pidfile)) finally: _kill(_read_pid(pidfile)) @@ -277,6 +331,8 @@ async def test_async_timeout_bounded_when_leader_ignores_sigterm_and_grandchild_ async def test_async_timeout_releases_pipes_held_by_escaped_grandchild( fake_qsv: Callable[[str], str], tmp_path: Path ) -> None: + if not os.path.isdir("/dev/fd"): + pytest.skip("no /dev/fd to count open descriptors") qsv = AsyncQsv(fake_qsv(HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.3) pidfiles = [tmp_path / f"child{i}.pid" for i in range(3)] open_fds = len(os.listdir("/dev/fd")) @@ -291,12 +347,14 @@ async def test_async_timeout_releases_pipes_held_by_escaped_grandchild( _kill(_read_pid(pidfile)) -IGNORE_SIGTERM_AND_HANG = """ -import os, signal, sys, time -signal.signal(signal.SIGTERM, signal.SIG_IGN) +IGNORE_STOP_AND_HANG = ( + IGNORE_STOP + + """ +import os, sys, time open(sys.argv[-1], "w").write(str(os.getpid())) time.sleep(60) """ +) async def test_async_cancel_during_timeout_cleanup_still_kills( @@ -305,7 +363,7 @@ async def test_async_cancel_during_timeout_cleanup_still_kills( # the run times out at 0.5s and starts its 2s SIGTERM grace wait; the outer deadline at # 1s cancels it mid-cleanup, which must not skip the SIGKILL pidfile = tmp_path / "leader.pid" - qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2) + qsv = AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=2) try: with pytest.raises(asyncio.TimeoutError): await asyncio.wait_for(qsv.run("stats", pidfile, timeout=0.5), 1) @@ -321,7 +379,7 @@ def test_asyncio_run_returning_mid_cleanup_still_kills( # as above, but asyncio.run() returns right after the outer deadline, so any cleanup still # running in a background task would be cancelled by the runner's shutdown pidfile = tmp_path / "leader.pid" - qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2) + qsv = AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=2) async def main() -> None: await asyncio.wait_for(qsv.run("stats", pidfile, timeout=0.5), 1) @@ -339,7 +397,7 @@ def test_asyncio_run_cancelling_background_run_mid_cleanup_still_kills( ) -> None: # the runner's shutdown cancels the run *and* its cleanup task directly, mid-grace-wait pidfile = tmp_path / "leader.pid" - qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2) + qsv = AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=2) async def main() -> None: background = asyncio.create_task(qsv.run("stats", pidfile, timeout=0.5)) From 4d225b46261ca22e375fc6d2ea0d366f781d4cb3 Mon Sep 17 00:00:00 2001 From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com> Date: Fri, 9 Oct 2026 11:59:14 -0400 Subject: [PATCH 2/5] fix: always force-kill on a second Ctrl-C; bound Windows stdin writes Addresses two findings of roborev review 5163: - A second Ctrl-C during the sync grace wait raised out of _terminate_sync before the forced kill, leaving a qsv that ignores SIGTERM / CTRL_BREAK_EVENT running (all platforms). The kill now runs in a finally, followed by a bounded reap; run_sync also closes output pipes an escaping exception would have left open. - Before Python 3.13, communicate() on Windows writes stdin synchronously before checking its timeout, so a child that stops reading could block a run forever. On Windows, stdin is now fed from a thread. Co-Authored-By: Claude Opus 5.5 --- src/qsv_client/_process.py | 49 ++++++++++++++++++++++++++--- tests/test_process.py | 63 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 4 deletions(-) diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py index 4d9ef86..f3d69df 100644 --- a/src/qsv_client/_process.py +++ b/src/qsv_client/_process.py @@ -22,6 +22,7 @@ import signal import subprocess import sys +import threading from collections.abc import Coroutine, Mapping, Sequence from dataclasses import dataclass from typing import IO, Any, TypeVar @@ -124,10 +125,40 @@ def close(self) -> None: def _terminate_sync(proc: subprocess.Popen[bytes], tree: _ProcessTree, grace: float) -> None: - tree.interrupt() - with contextlib.suppress(subprocess.TimeoutExpired): - proc.wait(timeout=grace) - tree.kill() + try: + tree.interrupt() + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(timeout=grace) + finally: + # even when a second Ctrl-C interrupts the grace wait; then reap the leader, which the + # forced kill ends at once, so it is not left behind as a zombie + tree.kill() + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(timeout=grace) + + +def _feed_stdin_in_thread(proc: subprocess.Popen[bytes], data: bytes) -> threading.Thread: + """Write ``data`` to the child's stdin from a thread, then close it. + + Before Python 3.13, ``communicate()`` on Windows writes stdin synchronously before it + checks its timeout, so a child that stops reading would block it forever. Detaching stdin + leaves ``communicate()`` only the reads, which do honor the timeout; killing the tree + breaks the pipe and ends this thread. + """ + pipe, proc.stdin = proc.stdin, None + + def feed() -> None: + assert pipe is not None + try: + with contextlib.suppress(OSError): + pipe.write(data) + finally: + with contextlib.suppress(OSError): + pipe.close() + + thread = threading.Thread(target=feed, name="qsv-stdin", daemon=True) + thread.start() + return thread def run_sync( @@ -150,6 +181,9 @@ def run_sync( **_spawn_kwargs(), ) tree = _ProcessTree(proc.pid) + feeder = None + if IS_WINDOWS and stdin is not None: + feeder, stdin = _feed_stdin_in_thread(proc, stdin), None timed_out = False try: out, err = proc.communicate(input=stdin, timeout=timeout) @@ -172,6 +206,13 @@ def run_sync( raise finally: tree.close() + if feeder is not None: + feeder.join(kill_grace) + # communicate() closes them when it finishes; an exception escaping first would not + for pipe in (proc.stdout, proc.stderr): + if pipe is not None and not pipe.closed: + with contextlib.suppress(OSError): + pipe.close() return RawResult( exit_code=proc.returncode, stdout=out or b"", diff --git a/tests/test_process.py b/tests/test_process.py index 784035d..51bb838 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -409,3 +409,66 @@ async def main() -> None: assert _wait_dead(_read_pid(pidfile), 6), "leader survived the runner's shutdown" finally: _kill(_read_pid(pidfile)) + + +def _wait_reaped(pid: int, within: float = 5.0) -> bool: + """Our own child that nothing else will reap: reap it here, so a SIGKILLed one counts.""" + deadline = time.monotonic() + within + while time.monotonic() < deadline: + try: + done, _ = os.waitpid(pid, os.WNOHANG) # type: ignore[attr-defined,unused-ignore] + except ChildProcessError: + return True + if done: + return True + time.sleep(0.05) + return False + + +@pytest.mark.skipif(not hasattr(signal, "setitimer"), reason="needs SIGALRM") +def test_second_interrupt_during_grace_wait_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + # times out at 0.5s and starts its 3s grace wait; a "second Ctrl-C" at 1s interrupts it + pidfile = tmp_path / "leader.pid" + qsv = Qsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=3) + + def interrupt(signum: int, frame: object) -> None: + raise KeyboardInterrupt + + previous = signal.signal(signal.SIGALRM, interrupt) # type: ignore[attr-defined,unused-ignore] + signal.setitimer(signal.ITIMER_REAL, 1.0) # type: ignore[attr-defined,unused-ignore] + try: + with pytest.raises(KeyboardInterrupt): + qsv.run("stats", pidfile, timeout=0.5) + assert _wait_reaped(_read_pid(pidfile)), "leader survived the second interrupt" + finally: + signal.setitimer(signal.ITIMER_REAL, 0) # type: ignore[attr-defined,unused-ignore] + signal.signal(signal.SIGALRM, previous) # type: ignore[attr-defined,unused-ignore] + _kill(_read_pid(pidfile)) + + +# 4 MiB, far more than any OS pipe buffer, to a fake that never reads stdin +HANG_WITHOUT_READING = """ +import time +time.sleep(60) +""" +BIG_STDIN = b"x" * (4 << 20) + + +def test_timeout_bounded_when_stdin_is_not_read(fake_qsv: Callable[[str], str]) -> None: + qsv = Qsv(fake_qsv(HANG_WITHOUT_READING), kill_grace=0.5) + started = time.monotonic() + with pytest.raises(QsvTimeout): + qsv.run("stats", stdin=BIG_STDIN, timeout=1) + assert time.monotonic() - started < 10 + + +async def test_async_timeout_bounded_when_stdin_is_not_read( + fake_qsv: Callable[[str], str], +) -> None: + qsv = AsyncQsv(fake_qsv(HANG_WITHOUT_READING), kill_grace=0.5) + started = time.monotonic() + with pytest.raises(QsvTimeout): + await qsv.run("stats", stdin=BIG_STDIN, timeout=1) + assert time.monotonic() - started < 10 From 04b3b606b764cfeff665f3877dcb1009a5c3a418 Mon Sep 17 00:00:00 2001 From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com> Date: Fri, 9 Oct 2026 12:30:37 -0400 Subject: [PATCH 3/5] fix: start qsv suspended on Windows so nothing escapes its Job Object Addresses the remaining finding of roborev review 5163. A child qsv started between CreateProcess and AssignProcessToJobObject landed outside the job; if it also ignored CTRL_BREAK_EVENT and held the output pipes, the timeout could block on Windows' pipe reader threads until it exited. qsv is now created with CREATE_SUSPENDED, assigned to the job, and only then resumed (its main thread found through a Toolhelp snapshot, since Popen closes the thread handle). If resuming fails, the process is killed and the run raises OSError rather than leaving it suspended. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 ++--- README.md | 8 ++-- src/qsv_client/_process.py | 83 ++++++++++++++++++++++++++++++++++---- 3 files changed, 85 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 891e649..e6ce86b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,11 +8,11 @@ All notable changes to this project are documented here. The format follows ### Added -- Windows support. Each run is placed in its own Job Object, so a timeout or cancellation - stops qsv *and* everything it started: `CTRL_BREAK_EVENT` first, then `TerminateJobObject` - after `kill_grace`. Previously only `qsv.exe` itself was killed. With `inherit_env=False`, - `SYSTEMROOT` is still passed on Windows, since many programs fail to start without it. CI - now runs the full test suite on Windows. +- Windows support. qsv is started suspended and placed in its own Job Object before it runs, + so a timeout or cancellation stops qsv *and* everything it started: `CTRL_BREAK_EVENT` + first, then `TerminateJobObject` after `kill_grace`. Previously only `qsv.exe` itself was + killed. With `inherit_env=False`, `SYSTEMROOT` is still passed on Windows, since many + programs fail to start without it. CI now runs the full test suite on Windows. ### Changed diff --git a/README.md b/README.md index e8fc81e..f1dd20d 100644 --- a/README.md +++ b/README.md @@ -123,10 +123,10 @@ On a timeout or cancellation, the client stops qsv and every process qsv started - **Linux, macOS:** each run gets its own session. The client sends SIGTERM to the process group, then SIGKILL after `kill_grace`. A process that leaves the group (by calling `setsid`) survives. -- **Windows:** each run goes in its own Job Object. The client sends `CTRL_BREAK_EVENT` to - the run's process group, then calls `TerminateJobObject` after `kill_grace`. Child processes - can't leave the job. A child started in the instant between qsv starting and being added to - the job is missed. +- **Windows:** qsv is started suspended, put in its own Job Object, and only then resumed, so + every process it starts is in the job and can't leave it. The client sends + `CTRL_BREAK_EVENT` to the run's process group, then calls `TerminateJobObject` after + `kill_grace`. ## Development diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py index f3d69df..64981c7 100644 --- a/src/qsv_client/_process.py +++ b/src/qsv_client/_process.py @@ -9,9 +9,9 @@ A descendant that leaves the group (e.g. by calling ``setsid``) survives the kill and may keep the output pipes open; after a timeout, output is collected for at most ``kill_grace`` seconds and anything that descendant still holds is dropped. -- Windows: each run is put in its own Job Object, which every process it starts joins and - cannot leave. ``CTRL_BREAK_EVENT`` to its process group, then ``TerminateJobObject``. A - child started in the instant between spawning qsv and assigning it to the job is missed. +- Windows: qsv is created suspended, put in its own Job Object, then resumed, so every + process it starts is in the job and cannot leave it. ``CTRL_BREAK_EVENT`` to its process + group, then ``TerminateJobObject``. """ from __future__ import annotations @@ -56,12 +56,67 @@ class RawResult: _kernel32.CloseHandle.argtypes = (wintypes.HANDLE,) _kernel32.CloseHandle.restype = wintypes.BOOL + class _THREADENTRY32(ctypes.Structure): + _fields_ = ( + ("dwSize", wintypes.DWORD), + ("cntUsage", wintypes.DWORD), + ("th32ThreadID", wintypes.DWORD), + ("th32OwnerProcessID", wintypes.DWORD), + ("tpBasePri", wintypes.LONG), + ("tpDeltaPri", wintypes.LONG), + ("dwFlags", wintypes.DWORD), + ) + + _kernel32.CreateToolhelp32Snapshot.argtypes = (wintypes.DWORD, wintypes.DWORD) + _kernel32.CreateToolhelp32Snapshot.restype = wintypes.HANDLE + _kernel32.Thread32First.argtypes = (wintypes.HANDLE, ctypes.POINTER(_THREADENTRY32)) + _kernel32.Thread32First.restype = wintypes.BOOL + _kernel32.Thread32Next.argtypes = (wintypes.HANDLE, ctypes.POINTER(_THREADENTRY32)) + _kernel32.Thread32Next.restype = wintypes.BOOL + _kernel32.OpenThread.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD) + _kernel32.OpenThread.restype = wintypes.HANDLE + _kernel32.ResumeThread.argtypes = (wintypes.HANDLE,) + _kernel32.ResumeThread.restype = wintypes.DWORD + + _CREATE_SUSPENDED = 0x00000004 _PROCESS_TERMINATE = 0x0001 _PROCESS_SET_QUOTA = 0x0100 + _TH32CS_SNAPTHREAD = 0x00000004 + _THREAD_SUSPEND_RESUME = 0x0002 + _INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value + _RESUME_FAILED = 0xFFFFFFFF def _spawn_kwargs() -> dict[str, Any]: - # its own console process group, so CTRL_BREAK_EVENT reaches qsv and its children only - return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} + # its own console process group, so CTRL_BREAK_EVENT reaches qsv and its children + # only; suspended, so it can start nothing before it is in its Job Object + return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP | _CREATE_SUSPENDED} + + def _resume(pid: int) -> bool: + """Resume the threads of a process created with CREATE_SUSPENDED. + + Popen closes the main thread's handle, so find it through a Toolhelp snapshot. + """ + snapshot = _kernel32.CreateToolhelp32Snapshot(_TH32CS_SNAPTHREAD, 0) + if not snapshot or snapshot == _INVALID_HANDLE_VALUE: + return False + resumed = False + try: + entry = _THREADENTRY32() + entry.dwSize = ctypes.sizeof(entry) + more = _kernel32.Thread32First(snapshot, ctypes.byref(entry)) + while more: + if entry.th32OwnerProcessID == pid: + thread = _kernel32.OpenThread(_THREAD_SUSPEND_RESUME, False, entry.th32ThreadID) + if thread: + try: + if _kernel32.ResumeThread(thread) != _RESUME_FAILED: + resumed = True + finally: + _kernel32.CloseHandle(thread) + more = _kernel32.Thread32Next(snapshot, ctypes.byref(entry)) + finally: + _kernel32.CloseHandle(snapshot) + return resumed def _create_job(pid: int) -> int | None: """A Job Object holding ``pid`` (and, from now on, everything it starts), or None.""" @@ -80,8 +135,14 @@ def _create_job(pid: int) -> int | None: class _ProcessTree: def __init__(self, pid: int) -> None: + """Put the (suspended) process in a new Job Object, then let it run.""" self.pid = pid self._job = _create_job(pid) + if not _resume(pid): + # never leave it suspended forever + self.kill() + self.close() + raise OSError(f"could not start qsv (pid {pid}): resuming its thread failed") def interrupt(self) -> None: with contextlib.suppress(OSError): @@ -180,7 +241,11 @@ def run_sync( cwd=cwd, **_spawn_kwargs(), ) - tree = _ProcessTree(proc.pid) + try: + tree = _ProcessTree(proc.pid) + except OSError: + proc.communicate() # already killed: reap it and close its pipes + raise feeder = None if IS_WINDOWS and stdin is not None: feeder, stdin = _feed_stdin_in_thread(proc, stdin), None @@ -321,7 +386,11 @@ async def run_async( cwd=cwd, **_spawn_kwargs(), ) - tree = _ProcessTree(proc.pid) + try: + tree = _ProcessTree(proc.pid) + except OSError: + await proc.communicate() # already killed: reap it and close its pipes + raise timed_out = False try: out, err = await asyncio.wait_for(proc.communicate(input=stdin), timeout) From ada93c2d0a3e8bad05e5da0ea722d4dccf0c6275 Mon Sep 17 00:00:00 2001 From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com> Date: Fri, 9 Oct 2026 12:37:55 -0400 Subject: [PATCH 4/5] fix: kill qsv if setup fails or is interrupted after the spawn Addresses roborev reviews 5164 and 5165: a BaseException raised between spawning qsv and entering the protected block (Ctrl-C during _ProcessTree setup, or Thread.start() failing for the Windows stdin feeder) bypassed teardown, and on Windows could leave qsv suspended. - _ProcessTree (Windows) kills and closes its job on any BaseException during setup, not just a failed resume. - run_sync / run_async catch any BaseException from setup, kill the leader, reap it and close its pipes before re-raising. - The stdin feeder starts inside the protected block and detaches proc.stdin only after its thread started; run_sync's finally now also closes stdin. Co-Authored-By: Claude Opus 5.5 --- src/qsv_client/_process.py | 37 +++++++++++++++------- tests/test_process.py | 64 +++++++++++++++++++++++++++++++++++++- 2 files changed, 88 insertions(+), 13 deletions(-) diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py index 64981c7..3998be9 100644 --- a/src/qsv_client/_process.py +++ b/src/qsv_client/_process.py @@ -137,12 +137,16 @@ class _ProcessTree: def __init__(self, pid: int) -> None: """Put the (suspended) process in a new Job Object, then let it run.""" self.pid = pid - self._job = _create_job(pid) - if not _resume(pid): - # never leave it suspended forever + self._job: int | None = None + try: + self._job = _create_job(pid) + if not _resume(pid): + raise OSError(f"could not start qsv (pid {pid}): resuming its thread failed") + except BaseException: + # failed or interrupted (Ctrl-C): never leave it suspended, or running untracked self.kill() self.close() - raise OSError(f"could not start qsv (pid {pid}): resuming its thread failed") + raise def interrupt(self) -> None: with contextlib.suppress(OSError): @@ -206,7 +210,7 @@ def _feed_stdin_in_thread(proc: subprocess.Popen[bytes], data: bytes) -> threadi leaves ``communicate()`` only the reads, which do honor the timeout; killing the tree breaks the pipe and ends this thread. """ - pipe, proc.stdin = proc.stdin, None + pipe = proc.stdin def feed() -> None: assert pipe is not None @@ -219,6 +223,8 @@ def feed() -> None: thread = threading.Thread(target=feed, name="qsv-stdin", daemon=True) thread.start() + # only now: if start() failed, the caller still owns stdin and closes it + proc.stdin = None return thread @@ -243,14 +249,18 @@ def run_sync( ) try: tree = _ProcessTree(proc.pid) - except OSError: - proc.communicate() # already killed: reap it and close its pipes + except BaseException: + # failed or interrupted (Ctrl-C) before there is a tree to tear down: _ProcessTree + # killed what it could; make sure of the leader, reap it and close its pipes + with contextlib.suppress(OSError): + proc.kill() + proc.communicate() raise feeder = None - if IS_WINDOWS and stdin is not None: - feeder, stdin = _feed_stdin_in_thread(proc, stdin), None timed_out = False try: + if IS_WINDOWS and stdin is not None: + feeder, stdin = _feed_stdin_in_thread(proc, stdin), None out, err = proc.communicate(input=stdin, timeout=timeout) except subprocess.TimeoutExpired: timed_out = True @@ -274,7 +284,7 @@ def run_sync( if feeder is not None: feeder.join(kill_grace) # communicate() closes them when it finishes; an exception escaping first would not - for pipe in (proc.stdout, proc.stderr): + for pipe in (proc.stdin, proc.stdout, proc.stderr): if pipe is not None and not pipe.closed: with contextlib.suppress(OSError): pipe.close() @@ -388,8 +398,11 @@ async def run_async( ) try: tree = _ProcessTree(proc.pid) - except OSError: - await proc.communicate() # already killed: reap it and close its pipes + except BaseException: + # failed or interrupted before there is a tree to tear down (see run_sync) + with contextlib.suppress(ProcessLookupError): + proc.kill() + await proc.communicate() raise timed_out = False try: diff --git a/tests/test_process.py b/tests/test_process.py index 51bb838..281ce6e 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -7,6 +7,7 @@ import os import signal import sys +import threading import time from collections.abc import Callable from pathlib import Path @@ -14,7 +15,7 @@ import pytest -from qsv_client import AsyncQsv, Qsv, QsvTimeout, QsvUsageError +from qsv_client import AsyncQsv, Qsv, QsvTimeout, QsvUsageError, _process # The fake starts a grandchild that inherits stdout and records its pid, then hangs. If only # the fake were killed, the grandchild would keep the stdout pipe open and the run would hang @@ -472,3 +473,64 @@ async def test_async_timeout_bounded_when_stdin_is_not_read( with pytest.raises(QsvTimeout): await qsv.run("stats", stdin=BIG_STDIN, timeout=1) assert time.monotonic() - started < 10 + + +class _Interrupted(BaseException): + """Stands in for Ctrl-C (or any BaseException) landing during setup after the spawn.""" + + +def _interrupting_tree(pidfile: Path) -> type: + class Tree(_process._ProcessTree): + def __init__(self, pid: int) -> None: + _read_pid(pidfile) # let the fake get going, so its death can be checked + raise _Interrupted + + return Tree + + +def test_interrupt_during_setup_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + pidfile = tmp_path / "leader.pid" + monkeypatch.setattr(_process, "_ProcessTree", _interrupting_tree(pidfile)) + try: + with pytest.raises(_Interrupted): + Qsv(fake_qsv(IGNORE_STOP_AND_HANG)).run("stats", pidfile) + assert _wait_reaped(_read_pid(pidfile)), "leader survived an interrupted setup" + finally: + _kill(_read_pid(pidfile)) + + +async def test_async_interrupt_during_setup_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + pidfile = tmp_path / "leader.pid" + monkeypatch.setattr(_process, "_ProcessTree", _interrupting_tree(pidfile)) + try: + with pytest.raises(_Interrupted): + await AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG)).run("stats", pidfile) + assert await asyncio.to_thread(_wait_dead, _read_pid(pidfile)), "leader survived" + finally: + _kill(_read_pid(pidfile)) + + +def test_stdin_feeder_failing_to_start_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # the Windows stdin feeder, forced on here, cannot get a thread + pidfile = tmp_path / "leader.pid" + + def no_thread(self: threading.Thread) -> None: + _read_pid(pidfile) + raise RuntimeError("can't start new thread") + + monkeypatch.setattr(_process, "IS_WINDOWS", True) + monkeypatch.setattr(threading.Thread, "start", no_thread) + try: + with pytest.raises(RuntimeError): + Qsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=0.5).run( + "stats", pidfile, stdin=b"a,b\n" + ) + assert _wait_reaped(_read_pid(pidfile)), "leader survived a failed feeder start" + finally: + _kill(_read_pid(pidfile)) From 2ee831a06e9fd583a84a12a0f823a3c77e89e05e Mon Sep 17 00:00:00 2001 From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com> Date: Fri, 9 Oct 2026 12:42:06 -0400 Subject: [PATCH 5/5] test: make the setup-failure tests work on Windows On Windows the interrupting _ProcessTree stand-in replaced the code that resumes the suspended qsv, so the fake never ran and never wrote the pidfile the tests waited for. Record the pid handed to the setup and check that instead, which covers a suspended (Windows) or running (POSIX) leader alike. _wait_reaped falls back to _wait_dead on Windows, which has no waitpid/WNOHANG. Co-Authored-By: Claude Opus 5.5 --- tests/test_process.py | 31 +++++++++++++++++++------------ 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/tests/test_process.py b/tests/test_process.py index 281ce6e..5f350e0 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -414,6 +414,8 @@ async def main() -> None: def _wait_reaped(pid: int, within: float = 5.0) -> bool: """Our own child that nothing else will reap: reap it here, so a SIGKILLed one counts.""" + if sys.platform == "win32": # no zombies (or waitpid) there + return _wait_dead(pid, within) deadline = time.monotonic() + within while time.monotonic() < deadline: try: @@ -479,10 +481,13 @@ class _Interrupted(BaseException): """Stands in for Ctrl-C (or any BaseException) landing during setup after the spawn.""" -def _interrupting_tree(pidfile: Path) -> type: +def _interrupting_tree(spawned: list[int]) -> type: + """A setup that is interrupted at once. On Windows qsv is then still suspended (it never + runs), so record the pid we were handed rather than wait for the fake to write one.""" + class Tree(_process._ProcessTree): def __init__(self, pid: int) -> None: - _read_pid(pidfile) # let the fake get going, so its death can be checked + spawned.append(pid) raise _Interrupted return Tree @@ -491,27 +496,29 @@ def __init__(self, pid: int) -> None: def test_interrupt_during_setup_still_kills( fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - pidfile = tmp_path / "leader.pid" - monkeypatch.setattr(_process, "_ProcessTree", _interrupting_tree(pidfile)) + spawned: list[int] = [] + monkeypatch.setattr(_process, "_ProcessTree", _interrupting_tree(spawned)) try: with pytest.raises(_Interrupted): - Qsv(fake_qsv(IGNORE_STOP_AND_HANG)).run("stats", pidfile) - assert _wait_reaped(_read_pid(pidfile)), "leader survived an interrupted setup" + Qsv(fake_qsv(IGNORE_STOP_AND_HANG)).run("stats", tmp_path / "leader.pid") + assert _wait_dead(spawned[0]), "leader survived an interrupted setup" finally: - _kill(_read_pid(pidfile)) + for pid in spawned: + _kill(pid) async def test_async_interrupt_during_setup_still_kills( fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - pidfile = tmp_path / "leader.pid" - monkeypatch.setattr(_process, "_ProcessTree", _interrupting_tree(pidfile)) + spawned: list[int] = [] + monkeypatch.setattr(_process, "_ProcessTree", _interrupting_tree(spawned)) try: with pytest.raises(_Interrupted): - await AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG)).run("stats", pidfile) - assert await asyncio.to_thread(_wait_dead, _read_pid(pidfile)), "leader survived" + await AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG)).run("stats", tmp_path / "leader.pid") + assert await asyncio.to_thread(_wait_dead, spawned[0]), "leader survived" finally: - _kill(_read_pid(pidfile)) + for pid in spawned: + _kill(pid) def test_stdin_feeder_failing_to_start_still_kills(