diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d68abdf..ed1dec9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,7 +24,7 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest, macos-latest] - python: ["3.10", "3.11", "3.12", "3.13"] + python: ["3.10", "3.11", "3.12", "3.13", "3.14"] runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v4 diff --git a/.serena/memories/tech_stack.md b/.serena/memories/tech_stack.md index fb13e4c..9d1836e 100644 --- a/.serena/memories/tech_stack.md +++ b/.serena/memories/tech_stack.md @@ -1,6 +1,6 @@ # Tech stack -- Python >=3.10 (ruff `target-version = py310`; CI matrix 3.10–3.13 on ubuntu + macos). Use `from __future__ import annotations`; no 3.11+ only stdlib features. +- Python >=3.10 (ruff `target-version = py310`; CI matrix 3.10–3.14 on ubuntu + macos). Use `from __future__ import annotations`; no 3.11+ only stdlib features. - No runtime dependencies — keep it that way (stdlib `subprocess`/`asyncio` only). - Package/env manager: uv (`uv.lock` committed); dev deps in `[dependency-groups].dev`: pytest, pytest-asyncio, mypy, ruff. - Build: hatchling, src layout (`src/qsv_client`). License AGPL-3.0-or-later. diff --git a/CHANGELOG.md b/CHANGELOG.md index 527f632..bba0f8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,21 @@ All notable changes to this project are documented here. The format follows included repository tooling files (`CLAUDE.md`, `.serena/`, `uv.lock`, `.github/`). The wheel is unchanged. ([#2](https://github.com/dathere/qsv-client/pull/2)) +### Fixed + +- A timeout now raises within a few `kill_grace` periods even when a process qsv started has + left the process group and still holds its output pipes, including when qsv ignores SIGTERM. + Previously the run waited for that process to exit. Output it still holds is dropped, and + `AsyncQsv` closes its pipe file descriptors instead of keeping them open until it exits. +- Cancelling an `AsyncQsv` run now always kills its process group, even when the cancellation + arrives during timeout cleanup or comes from `asyncio.run()` shutting down. A cancelled run + finishes its teardown before the cancellation propagates. +- A failed run whose stdout is not valid UTF-8 raises the typed `QsvError` instead of + `UnicodeDecodeError`. +- `count()` and `headers()` work with `stdout_path=` and `text=False` instead of raising + `IndexError`. +- A relative `binary` path now works together with `cwd=`. `find_qsv` returns absolute paths. + ## [0.1.0] - 2026-10-08 First release. ([#1](https://github.com/dathere/qsv-client/pull/1)) diff --git a/CLAUDE.md b/CLAUDE.md index 04b8b74..67ac574 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -13,7 +13,7 @@ QSV_BIN=/path/to/qsvlite uv run pytest -q tests/test_integration.py # other bi uv run ruff format . && uv run ruff check . && uv run mypy # must be clean before done ``` -CI runs lint, format check, and mypy, plus tests on ubuntu/macos x Python 3.10–3.13 against +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). ## Layout (`src/qsv_client/`) diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py index d6a3bd3..8cec986 100644 --- a/src/qsv_client/_process.py +++ b/src/qsv_client/_process.py @@ -5,6 +5,10 @@ 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. """ from __future__ import annotations @@ -15,12 +19,14 @@ import signal import subprocess import sys -from collections.abc import Mapping, Sequence +from collections.abc import Coroutine, Mapping, Sequence from dataclasses import dataclass -from typing import IO, Any +from typing import IO, Any, TypeVar IS_WINDOWS = sys.platform == "win32" +_T = TypeVar("_T") + @dataclass(frozen=True) class RawResult: @@ -78,7 +84,15 @@ def run_sync( except subprocess.TimeoutExpired: timed_out = True _terminate_sync(proc, kill_grace) - out, err = proc.communicate() + try: + out, err = proc.communicate(timeout=kill_grace) + except subprocess.TimeoutExpired: + # a descendant that left the group still holds the pipes: drop its output + for pipe in (proc.stdout, proc.stderr): + if pipe is not None: + pipe.close() + proc.wait() + out, err = b"", b"" except BaseException: # KeyboardInterrupt etc.: never leave a qsv process group running behind us _terminate_sync(proc, kill_grace) @@ -96,15 +110,89 @@ async def _terminate_async(proc: asyncio.subprocess.Process, grace: float) -> No if IS_WINDOWS: with contextlib.suppress(ProcessLookupError): proc.kill() - await proc.wait() + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(proc.wait(), grace) return + # 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) with contextlib.suppress(asyncio.TimeoutError): await asyncio.wait_for(proc.wait(), grace) _signal_group(proc.pid, signal.SIGKILL) # type: ignore[attr-defined,unused-ignore] + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(proc.wait(), grace) + + +async def _release_async(proc: asyncio.subprocess.Process) -> None: + """Close our ends of the pipes, then reap the (already killed) leader. + + Without the close, pipes held by an escaped descendant would keep their fds open and + ``wait()`` pending until it exits. asyncio has no public API for this. + """ + _close_pipes(proc) await proc.wait() +def _close_pipes(proc: asyncio.subprocess.Process) -> None: + transport = getattr(proc, "_transport", None) + if transport is not None: + transport.close() + + +def _kill_now(proc: asyncio.subprocess.Process) -> 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] + _close_pipes(proc) + + +async def _reap_async(proc: asyncio.subprocess.Process, grace: float) -> None: + try: + await _terminate_async(proc, grace) + await _release_async(proc) + except asyncio.CancelledError: + _kill_now(proc) + raise + + +async def _timeout_cleanup(proc: asyncio.subprocess.Process, grace: float) -> bytes: + """Kill and release a timed-out run; return whatever stderr arrives within ``grace``.""" + try: + await _terminate_async(proc, 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) + raise + return err + + +async def _finish_despite_cancel(coro: Coroutine[Any, Any, _T]) -> _T: + """Run ``coro`` to completion even if we are cancelled meanwhile, then re-raise the cancel. + + ``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. + """ + task = asyncio.ensure_future(coro) + cancel: asyncio.CancelledError | None = None + while not task.done(): + try: + await asyncio.wait({task}) + except asyncio.CancelledError as exc: + cancel = exc + if cancel is not None: + raise cancel + return task.result() + + async def run_async( argv: Sequence[str], *, @@ -129,14 +217,11 @@ async def run_async( out, err = await asyncio.wait_for(proc.communicate(input=stdin), timeout) except asyncio.TimeoutError: timed_out = True - await _terminate_async(proc, kill_grace) - out, err = b"", b"" - if proc.stderr is not None: - with contextlib.suppress(Exception): - err = await proc.stderr.read() + # 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)) except BaseException: # task cancelled (or the loop is shutting down): reap the group, then propagate - await asyncio.shield(_terminate_async(proc, kill_grace)) + await _finish_despite_cancel(_reap_async(proc, kill_grace)) raise assert proc.returncode is not None return RawResult( diff --git a/src/qsv_client/client.py b/src/qsv_client/client.py index 79f2991..9a8d64d 100644 --- a/src/qsv_client/client.py +++ b/src/qsv_client/client.py @@ -43,12 +43,14 @@ def find_qsv(binary: str | os.PathLike[str] | None = None) -> str: candidates.extend(DEFAULT_BINARIES) for cand in candidates: if os.sep in cand or (os.altsep and os.altsep in cand): + # absolute, so a relative path still resolves when the run sets cwd=. Not + # normalized: collapsing `..` lexically would be wrong after a symlinked dir. if Path(cand).is_file(): - return str(Path(cand)) + return str(Path(cand).absolute()) continue found = shutil.which(cand) if found: - return found + return str(Path(found).absolute()) raise QsvNotFound( f"no qsv binary found (tried {', '.join(candidates)}). Install qsv " "(https://github.com/dathere/qsv/releases), put it on PATH, or pass binary= / set QSV_BIN." @@ -122,6 +124,8 @@ def __init__( ``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 + cancellation, and the most spent collecting output after the kill. """ self.binary = find_qsv(binary) self.timeout = timeout @@ -187,8 +191,16 @@ def _finish( args=argv, stderr=stderr, ) + # before decoding stdout, so undecodable output can't mask the typed error + if check and raw.exit_code not in SUCCESS_EXIT_CODES: + raise error_from_run( + exit_code=raw.exit_code, + stderr=stderr, + args=argv, + command=command or None, + ) captured = stdout_path is None - result = QsvResult( + return QsvResult( args=tuple(argv), exit_code=raw.exit_code, stdout=raw.stdout.decode("utf-8") if captured and text else None, @@ -197,14 +209,6 @@ def _finish( duration=duration, stdout_path=stdout_path, ) - if check and not result.ok: - raise error_from_run( - exit_code=raw.exit_code, - stderr=stderr, - args=argv, - command=command or None, - ) - return result def _open_stdout( @@ -349,12 +353,12 @@ def _check_version_once(self) -> None: def count(self, path: Arg, *args: Arg, **kw: Any) -> int: """Number of records (``qsv count``).""" - out = self.run("count", path, *args, **kw).stdout or "" + out = self.run("count", path, *args, **kw)._text() return int(out.strip().split()[0].split(";")[0].replace(",", "")) def headers(self, path: Arg, *args: Arg, **kw: Any) -> list[str]: """Column names (``qsv headers --just-names``).""" - out = self.run("headers", "--just-names", path, *args, **kw).stdout or "" + out = self.run("headers", "--just-names", path, *args, **kw)._text() return [line for line in out.splitlines() if line] def stats(self, path: Arg, *args: Arg, **kw: Any) -> list[dict[str, str]]: @@ -474,11 +478,11 @@ async def _check_version_once(self) -> None: await self.capabilities() async def count(self, path: Arg, *args: Arg, **kw: Any) -> int: - out = (await self.run("count", path, *args, **kw)).stdout or "" + out = (await self.run("count", path, *args, **kw))._text() return int(out.strip().split()[0].split(";")[0].replace(",", "")) async def headers(self, path: Arg, *args: Arg, **kw: Any) -> list[str]: - out = (await self.run("headers", "--just-names", path, *args, **kw)).stdout or "" + out = (await self.run("headers", "--just-names", path, *args, **kw))._text() return [line for line in out.splitlines() if line] async def stats(self, path: Arg, *args: Arg, **kw: Any) -> list[dict[str, str]]: diff --git a/tests/test_process.py b/tests/test_process.py index dab3f15..9f1532d 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -3,14 +3,17 @@ from __future__ import annotations import asyncio +import contextlib import os +import signal import time from collections.abc import Callable from pathlib import Path +from typing import Any import pytest -from qsv_client import AsyncQsv, Qsv, QsvTimeout +from qsv_client import AsyncQsv, Qsv, QsvTimeout, QsvUsageError # 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 @@ -23,6 +26,17 @@ """ +def _is_zombie(pid: int) -> bool: + """Linux only: dead but not yet reaped. On Python 3.12+ asyncio reaps children from the + event loop (a pidfd reader), so a child killed as ``asyncio.run()`` shuts down is never + reaped by it and ``os.kill(pid, 0)`` keeps succeeding.""" + try: + stat = Path(f"/proc/{pid}/stat").read_text() + except OSError: + return False + 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: @@ -30,6 +44,8 @@ def _wait_dead(pid: int, within: float = 5.0) -> bool: os.kill(pid, 0) except ProcessLookupError: return True + if _is_zombie(pid): + return True time.sleep(0.05) return False @@ -142,3 +158,196 @@ def test_warning_exit_code_is_success(fake_qsv: Callable[[str], str]) -> None: res = Qsv(fake_qsv("import sys; print('done'); sys.exit(255)")).run("x") assert res.ok assert res.exit_code == 255 + + +# Like HANG_WITH_GRANDCHILD, but the grandchild leaves the process group, so the group kill +# misses it and it keeps the output pipes open. +HANG_WITH_ESCAPED_GRANDCHILD = """ +import subprocess, sys, time +child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], + start_new_session=True) +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) + + +def test_timeout_bounded_when_grandchild_escapes_group( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + pidfile = tmp_path / "child.pid" + qsv = Qsv(fake_qsv(HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.5) + started = time.monotonic() + try: + with pytest.raises(QsvTimeout): + qsv.run("stats", pidfile, timeout=1) + assert time.monotonic() - started < 10 + finally: + _kill(_read_pid(pidfile)) + + +async def test_async_timeout_bounded_when_grandchild_escapes_group( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + pidfile = tmp_path / "child.pid" + qsv = AsyncQsv(fake_qsv(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 + finally: + _kill(_read_pid(pidfile)) + + +def test_undecodable_stdout_still_raises_typed_error(fake_qsv: Callable[[str], str]) -> None: + qsv = Qsv(fake_qsv("import sys; sys.stdout.buffer.write(b'\\xff'); sys.exit(2)")) + with pytest.raises(QsvUsageError): + qsv.run("x") + + +COUNT_AND_HEADERS = """ +import sys +print("3" if sys.argv[1] == "count" else "name\\nn") +""" + + +def test_count_and_headers_without_captured_text( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + qsv = Qsv(fake_qsv(COUNT_AND_HEADERS)) + kws: list[dict[str, Any]] = [{"stdout_path": tmp_path / "out"}, {"text": False}] + for kw in kws: + assert qsv.count("in.csv", **kw) == 3 + assert qsv.headers("in.csv", **kw) == ["name", "n"] + + +async def test_async_count_and_headers_without_captured_text( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + qsv = AsyncQsv(fake_qsv(COUNT_AND_HEADERS)) + kws: list[dict[str, Any]] = [{"stdout_path": tmp_path / "out"}, {"text": False}] + for kw in kws: + assert await qsv.count("in.csv", **kw) == 3 + assert await qsv.headers("in.csv", **kw) == ["name", "n"] + + +def test_relative_binary_with_cwd( + fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fake = Path(fake_qsv("print('ok')")) + other = tmp_path / "other" + other.mkdir() + monkeypatch.chdir(fake.parent) + assert Qsv(f"./{fake.name}", cwd=other).run("x").stdout == "ok\n" + + +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')") + (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" + + +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) + started = time.monotonic() + try: + with pytest.raises(QsvTimeout): + await qsv.run("stats", pidfile, timeout=1) + assert time.monotonic() - started < 10 + finally: + _kill(_read_pid(pidfile)) + + +async def test_async_timeout_releases_pipes_held_by_escaped_grandchild( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + 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")) + try: + for pidfile in pidfiles: + with pytest.raises(QsvTimeout): + await qsv.run("stats", pidfile, timeout=0.5) + await asyncio.sleep(0) # asyncio closes pipe fds in a call_soon callback + assert len(os.listdir("/dev/fd")) <= open_fds + finally: + for pidfile in pidfiles: + _kill(_read_pid(pidfile)) + + +IGNORE_SIGTERM_AND_HANG = """ +import os, signal, sys, time +signal.signal(signal.SIGTERM, signal.SIG_IGN) +open(sys.argv[-1], "w").write(str(os.getpid())) +time.sleep(60) +""" + + +async def test_async_cancel_during_timeout_cleanup_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + # 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) + try: + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(qsv.run("stats", pidfile, timeout=0.5), 1) + pid = _read_pid(pidfile) + assert await asyncio.to_thread(_wait_dead, pid, 6), "leader survived cancellation" + finally: + _kill(_read_pid(pidfile)) + + +def test_asyncio_run_returning_mid_cleanup_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + # 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) + + async def main() -> None: + await asyncio.wait_for(qsv.run("stats", pidfile, timeout=0.5), 1) + + try: + with pytest.raises(asyncio.TimeoutError): + asyncio.run(main()) + assert _wait_dead(_read_pid(pidfile), 6), "leader survived the runner's shutdown" + finally: + _kill(_read_pid(pidfile)) + + +def test_asyncio_run_cancelling_background_run_mid_cleanup_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> 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) + + async def main() -> None: + background = asyncio.create_task(qsv.run("stats", pidfile, timeout=0.5)) + await asyncio.sleep(1) + assert not background.done() # still in its SIGTERM grace wait + + try: + asyncio.run(main()) + assert _wait_dead(_read_pid(pidfile), 6), "leader survived the runner's shutdown" + finally: + _kill(_read_pid(pidfile))