diff --git a/loopx/capabilities/change_quality/scope.py b/loopx/capabilities/change_quality/scope.py index e312315752..5a66a75a42 100644 --- a/loopx/capabilities/change_quality/scope.py +++ b/loopx/capabilities/change_quality/scope.py @@ -1,14 +1,43 @@ from __future__ import annotations import hashlib +import os import subprocess from pathlib import Path from typing import Any +from ...extensions.process_runtime import retire_owned_process CHANGE_QUALITY_SCOPE_SCHEMA_VERSION = "change_quality_scope_v0" +# The scope reads a checkout the operator may not control. A repository blocked +# on a stale index lock, a network mount or an oversized history must not hold +# the receipt open with no answer, so every read shares one budget. +GIT_PROBE_TIMEOUT_SECONDS = 10.0 + + +class GitProbeUnavailable(ValueError): + """Git did not answer inside the read budget, or could not start.""" + + +def _retire_owned_probe(process: subprocess.Popen[Any]) -> None: + """Stop the read's own session, then drop its pipes unread. + + Git runs helpers -- fsmonitor hooks, pagers -- inside the session it starts + in, so killing the leader alone leaves them behind. An inherited pipe also + outlives the leader, so the pipes are closed instead of drained: a drain + would wait on work this call no longer owns. + """ + + try: + retire_owned_process(process, process_tree_owned=True) + except (OSError, RuntimeError, ValueError, subprocess.SubprocessError): + # Cleanup is best effort. The read's own timeout or cancellation is the + # result the caller must receive, even when termination is unconfirmed. + pass + + def _git( repo_root: Path, *args: str, @@ -18,14 +47,35 @@ def _git( # text mode: a non-None encoding would silently flip bytes-mode callers # into text mode because CPython treats any of text/encoding/errors as # text-mode requests. - return subprocess.run( - ["git", "-C", str(repo_root), *args], - check=False, - text=text, - encoding="utf-8" if text else None, - errors="replace" if text else None, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, + argv = ["git", "-C", str(repo_root), *args] + budget = GIT_PROBE_TIMEOUT_SECONDS + try: + process = subprocess.Popen( + argv, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=text, + encoding="utf-8" if text else None, + errors="replace" if text else None, + start_new_session=os.name == "posix", + ) + except OSError as exc: + raise GitProbeUnavailable(f"git {' '.join(args)} could not start: {exc}") from exc + try: + stdout, stderr = process.communicate(timeout=budget) + except subprocess.TimeoutExpired: + _retire_owned_probe(process) + raise GitProbeUnavailable( + f"git {' '.join(args)} did not answer within {budget:.1f}s" + ) from None + except BaseException: + # A terminal Ctrl-C reaches this process's group, not the session the + # read started, so the cancellation retires its helpers here and then + # keeps travelling to its own handler. + _retire_owned_probe(process) + raise + return subprocess.CompletedProcess( + args=argv, returncode=process.returncode, stdout=stdout, stderr=stderr ) diff --git a/loopx/extensions/process_runtime.py b/loopx/extensions/process_runtime.py index aa74afa28e..cbb4a28d56 100644 --- a/loopx/extensions/process_runtime.py +++ b/loopx/extensions/process_runtime.py @@ -1,7 +1,5 @@ from __future__ import annotations -from collections.abc import Callable, Mapping, Sequence -from dataclasses import dataclass import math import os import shutil @@ -11,9 +9,10 @@ import sys import threading import time +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass from pathlib import Path -from typing import BinaryIO - +from typing import Any, BinaryIO _PROCESS_IO_CHUNK_BYTES = 64 * 1024 _PROCESS_TERMINATE_GRACE_SECONDS = 1.0 @@ -310,7 +309,15 @@ def _force_terminate_windows_process_tree( stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) try: - if taskkill.wait(timeout=max(0, deadline - time.monotonic()) / 2) != 0: + try: + exit_code = taskkill.wait( + timeout=max(0, deadline - time.monotonic()) / 2 + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError( + "owned Windows process-tree cleanup unconfirmed" + ) from exc + if exit_code != 0: raise RuntimeError("owned Windows process-tree cleanup unconfirmed") finally: # subprocess.run(timeout=...) internally waits without a timeout @@ -345,6 +352,71 @@ def force_terminate_process( +def _close_owned_streams_before_deadline( + process: subprocess.Popen[bytes] | subprocess.Popen[str], deadline: float, +) -> None: + """Close a process's pipe ends without waiting on a blocked reader. + + ``communicate`` retains reader threads on Windows; a descendant holding the + other end of a pipe can keep that reader, and the stream's lock, alive. + Closing from the caller would then wait past the read budget, so each close + runs on a daemon thread and is joined only until the deadline. A close that + cannot finish is left to the reader it is waiting for. + """ + + threads: list[threading.Thread] = [] + for stream in (process.stdin, process.stdout, process.stderr): + if stream is None: + continue + + def close(stream: Any = stream) -> None: + try: + stream.close() + except (OSError, ValueError): + pass + + thread = threading.Thread( + target=close, + name="loopx-process-stream-close", + daemon=True, + ) + thread.start() + threads.append(thread) + for thread in threads: + thread.join(timeout=max(0, deadline - time.monotonic())) + + +def retire_owned_process( + process: subprocess.Popen[bytes] | subprocess.Popen[str], *, + process_tree_owned: bool, timeout: float = _PROCESS_TERMINATE_GRACE_SECONDS, +) -> None: + """Force-retire an owned process and release its pipes within a deadline. + + ``force_terminate_process`` bounds the native termination path, including + Windows taskkill and its helper. ``communicate`` may still retain a reader + thread, so pipe closure uses the same deadline and cannot turn a timed-out + read into an unbounded wait. The original timeout or cancellation remains + the caller's result even when termination cannot be confirmed. + """ + + if not math.isfinite(timeout) or timeout <= 0: + raise ValueError("native process cleanup timeout must be positive and finite") + deadline = time.monotonic() + timeout + # Split one budget: native termination gets most of it, the remainder is + # left for pipe handles a Windows reader thread may still own. The inner + # budget is derived from what is left of the caller's deadline and never + # widened to a floor, so a sub-50 ms cleanup stays sub-50 ms. + terminate_timeout = min(timeout * 0.8, deadline - time.monotonic()) + try: + force_terminate_process( + process, + process_tree_owned=process_tree_owned, + timeout=max(terminate_timeout, 1e-6), + ) + finally: + _close_owned_streams_before_deadline(process, deadline) + + def prepare_owned_process_cleanup(process: subprocess.Popen[bytes]) -> Callable[[], None]: """Retain diagnostic tree ownership even if its leader exits. diff --git a/tests/extensions/test_process_runtime.py b/tests/extensions/test_process_runtime.py index 8f5d4344fa..be66449e2f 100644 --- a/tests/extensions/test_process_runtime.py +++ b/tests/extensions/test_process_runtime.py @@ -17,6 +17,7 @@ import loopx.extensions.process_runtime as process_runtime from loopx.extensions.process_runtime import ( force_terminate_process, + retire_owned_process, remove_owned_tree, run_capped_process, terminate_process_tree, @@ -224,6 +225,34 @@ def collect(*, timeout): observer.stdout.close.assert_called_once_with() +class _NoStreamsProcess: + """Stand-in for a Popen whose pipes ``communicate`` already released.""" + + stdin = None + stdout = None + stderr = None + + +def test_retire_owned_process_never_widens_a_sub_grace_budget(monkeypatch): + """A 20 ms cleanup stays inside 20 ms; it is not raised to a 50 ms floor.""" + + process = _NoStreamsProcess() + captured = {} + + def record_termination(process, *, process_tree_owned, timeout): + captured["timeout"] = timeout + captured["owned"] = process_tree_owned + + monkeypatch.setattr(process_runtime, "force_terminate_process", record_termination) + started = time.monotonic() + retire_owned_process(process, process_tree_owned=True, timeout=.02) + elapsed = time.monotonic() - started + + assert captured["owned"] is True + assert 0 < captured["timeout"] <= .016 + assert elapsed < .05 + + @pytest.mark.parametrize("timeout", [0, -1, float("nan"), float("inf")]) def test_force_cleanup_rejects_an_unbounded_or_invalid_budget(timeout): process = Mock(spec=subprocess.Popen) diff --git a/tests/test_change_quality_git_budget.py b/tests/test_change_quality_git_budget.py new file mode 100644 index 0000000000..6ebe0c9bdb --- /dev/null +++ b/tests/test_change_quality_git_budget.py @@ -0,0 +1,268 @@ +"""The change-quality scope reads Git under one budget. + +The scope used to call `subprocess.run` with no timeout, so a checkout blocked +on a stale index lock, a network mount or an oversized history held the receipt +open with no answer. These tests pin the budget, the retirement of the session +a blocked read started, and the cancellation path a terminal Ctrl-C takes. +""" + +from __future__ import annotations + +import os +import shlex +import shutil +import signal +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from loopx.capabilities.change_quality import scope +from loopx.extensions import process_runtime + + +class _WindowsNameProxy: + """Report Windows only to the runtime under test, keeping pathlib host-native.""" + + def __init__(self, module): + self._module = module + + def __getattr__(self, name): + return getattr(self._module, name) + + @property + def name(self): + return "nt" + + +def _worktree(tmp_path: Path) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + for command in ( + ["git", "init", "-q", str(repo)], + ["git", "-C", str(repo), "config", "user.email", "fixture@example.invalid"], + ["git", "-C", str(repo), "config", "user.name", "Scope Fixture"], + ): + subprocess.run(command, check=True) + (repo / "tracked.txt").write_text("original\n", encoding="utf-8") + subprocess.run(["git", "-C", str(repo), "add", "tracked.txt"], check=True) + subprocess.run(["git", "-C", str(repo), "commit", "-qm", "fixture"], check=True) + return repo + + +def _fake_git(tmp_path: Path, script: str, monkeypatch: pytest.MonkeyPatch) -> Path: + bin_dir = tmp_path / "fakebin" + bin_dir.mkdir(exist_ok=True) + fake = bin_dir / "git" + fake.write_text(script, encoding="utf-8") + fake.chmod(0o755) + monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}") + return bin_dir + + +@pytest.mark.skipif(os.name != "posix", reason="the fake git is a POSIX script") +def test_a_git_that_never_answers_stops_inside_the_budget( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + repo = _worktree(tmp_path) + _fake_git(tmp_path, "#!/bin/sh\nsleep 30\n", monkeypatch) + monkeypatch.setattr(scope, "GIT_PROBE_TIMEOUT_SECONDS", 0.3) + + started = time.monotonic() + with pytest.raises(ValueError, match="did not answer"): + scope.resolve_git_root(repo) + elapsed = time.monotonic() - started + + assert elapsed < 6, f"a blocked Git held the read for {elapsed:.1f}s" + + +@pytest.mark.skipif(os.name != "posix", reason="the fake git is a POSIX script") +def test_a_blocked_read_takes_the_helper_it_started( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A blocked query retires the session, so its child stops too.""" + + repo = _worktree(tmp_path) + heartbeats = tmp_path / "heartbeats" + heartbeats.write_text("", encoding="utf-8") + stop = tmp_path / "stop" + real_git = shutil.which("git") + assert real_git is not None + _fake_git( + tmp_path, + "#!/bin/sh\n" + "for argument in \"$@\"; do\n" + " if [ \"$argument\" = \"--show-toplevel\" ]; then\n" + f" while [ ! -f {shlex.quote(str(stop))} ]; do\n" + f" echo tick >> {shlex.quote(str(heartbeats))}\n" + " sleep 0.1\n" + " done\n" + " exit 1\n" + " fi\n" + "done\n" + f'exec {shlex.quote(real_git)} "$@"\n', + monkeypatch, + ) + monkeypatch.setattr(scope, "GIT_PROBE_TIMEOUT_SECONDS", 0.8) + + try: + started = time.monotonic() + with pytest.raises(ValueError, match="did not answer"): + scope.resolve_git_root(repo) + assert time.monotonic() - started < 6 + retired = heartbeats.stat().st_size + assert retired > 0, "the fixture never blocked, so this proves nothing" + time.sleep(0.6) + assert heartbeats.stat().st_size == retired, ( + "the blocked read's helper outlived the budget" + ) + finally: + stop.write_text("stop\n", encoding="utf-8") + + +@pytest.mark.skipif(os.name != "posix", reason="a terminal Ctrl-C is a POSIX group signal") +def test_a_cancelled_read_takes_the_helper_it_started(tmp_path: Path) -> None: + """Ctrl-C during the read retires the session that read owns.""" + + repo = _worktree(tmp_path) + heartbeats = tmp_path / "heartbeats" + heartbeats.write_text("", encoding="utf-8") + stop = tmp_path / "stop" + real_git = shutil.which("git") + assert real_git is not None + bin_dir = tmp_path / "blocking-bin" + bin_dir.mkdir() + fake = bin_dir / "git" + fake.write_text( + f"#!{sys.executable}\n" + "import os, sys, time\n" + "from pathlib import Path\n" + "if '--show-toplevel' not in sys.argv:\n" + f" os.execv({real_git!r}, [{real_git!r}] + sys.argv[1:])\n" + f"heartbeats, stop = Path({str(heartbeats)!r}), Path({str(stop)!r})\n" + "while not stop.exists():\n" + " with heartbeats.open('a', encoding='utf-8') as stream:\n" + " stream.write('tick\\n')\n" + " time.sleep(0.1)\n", + encoding="utf-8", + ) + fake.chmod(0o755) + environment = { + key: value + for key, value in os.environ.items() + if not key.startswith(("PYTEST", "COVERAGE", "COV_CORE")) + } + environment.update({ + "PYTHONPATH": str(Path(__file__).resolve().parents[1]), + "LOOPX_USAGE_PING": "0", + "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}", + }) + command = subprocess.Popen( + [ + sys.executable, "-c", + ( + "import sys\n" + "from pathlib import Path\n" + "from loopx.capabilities.change_quality.scope import build_change_quality_scope\n" + "build_change_quality_scope(repo_path=Path(sys.argv[1]))\n" + ), + str(repo), + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=environment, + start_new_session=True, + ) + try: + deadline = time.monotonic() + 25 + while heartbeats.stat().st_size == 0 and time.monotonic() < deadline: + time.sleep(0.05) + assert heartbeats.stat().st_size > 0, "the fixture never blocked, so this proves nothing" + assert command.poll() is None, "the read had already returned" + os.killpg(os.getpgid(command.pid), signal.SIGINT) + command.communicate(timeout=25) + assert command.returncode != 0 + retired = heartbeats.stat().st_size + time.sleep(0.6) + assert heartbeats.stat().st_size == retired, ( + "the blocked read and its helper outlived the cancellation" + ) + finally: + stop.write_text("stop\n", encoding="utf-8") + if command.poll() is None: + os.killpg(os.getpgid(command.pid), signal.SIGKILL) + command.communicate(timeout=25) + + +def test_a_healthy_repository_still_builds_its_scope(tmp_path: Path) -> None: + repo = _worktree(tmp_path) + (repo / "untracked.txt").write_text("new\n", encoding="utf-8") + + result = scope.build_change_quality_scope(repo_path=repo, base_ref="HEAD") + + assert result["base_commit"] == result["head_commit"] + assert result["sources"]["untracked"] == ["untracked.txt"] + assert result["changed_file_count"] == len(result["changed_files"]) + assert scope.resolve_git_root(repo) == repo.resolve() + +_TASKKILL_FAILURES = ( + "#!/bin/sh\nwhile :; do :; done\n", + "#!/bin/sh\nexit 1\n", +) + + +@pytest.mark.skipif(os.name != "posix", reason="the fake binaries are POSIX scripts") +@pytest.mark.parametrize("taskkill_script", _TASKKILL_FAILURES) +def test_a_failed_windows_cleanup_still_returns_the_git_timeout( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + taskkill_script: str, +) -> None: + """A stuck or refused taskkill must not replace the read's timeout.""" + + repo = _worktree(tmp_path) + bin_dir = _fake_git( + tmp_path, "#!/bin/sh\nwhile :; do :; done\n", monkeypatch + ) + taskkill = bin_dir / "taskkill" + taskkill.write_text(taskkill_script, encoding="utf-8") + taskkill.chmod(0o755) + monkeypatch.setattr(process_runtime, "os", _WindowsNameProxy(os)) + monkeypatch.setattr(scope, "GIT_PROBE_TIMEOUT_SECONDS", 0.25) + + started = time.monotonic() + with pytest.raises(ValueError, match="did not answer"): + scope.resolve_git_root(repo) + assert time.monotonic() - started < 3 + + +@pytest.mark.skipif(os.name != "posix", reason="the fake binaries are POSIX scripts") +@pytest.mark.parametrize("taskkill_script", _TASKKILL_FAILURES) +def test_a_failed_windows_cleanup_still_propagates_cancellation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + taskkill_script: str, +) -> None: + """Ctrl-C keeps travelling even when Windows termination is unconfirmed.""" + + repo = _worktree(tmp_path) + bin_dir = _fake_git( + tmp_path, "#!/bin/sh\nwhile :; do :; done\n", monkeypatch + ) + taskkill = bin_dir / "taskkill" + taskkill.write_text(taskkill_script, encoding="utf-8") + taskkill.chmod(0o755) + monkeypatch.setattr(process_runtime, "os", _WindowsNameProxy(os)) + + def interrupted(self, timeout=None, input=None): + raise KeyboardInterrupt + + monkeypatch.setattr(subprocess.Popen, "communicate", interrupted) + + started = time.monotonic() + with pytest.raises(KeyboardInterrupt): + scope.resolve_git_root(repo) + assert time.monotonic() - started < 3