diff --git a/loopx/extensions/process_runtime.py b/loopx/extensions/process_runtime.py index cbb4a28d56..bb7875e279 100644 --- a/loopx/extensions/process_runtime.py +++ b/loopx/extensions/process_runtime.py @@ -12,7 +12,7 @@ from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from pathlib import Path -from typing import Any, BinaryIO +from typing import Any, BinaryIO, IO, cast _PROCESS_IO_CHUNK_BYTES = 64 * 1024 _PROCESS_TERMINATE_GRACE_SECONDS = 1.0 @@ -588,24 +588,11 @@ def run_capped_process( ) -> CappedProcessResult: """Run a provider while bounding both output streams during execution.""" - process = subprocess.Popen( - list(argv), - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - bufsize=0, - env=dict(env) if env is not None else None, - cwd=cwd, - start_new_session=os.name == "posix", - creationflags=( - int(getattr(subprocess, "CREATE_NEW_PROCESS_GROUP")) - if os.name == "nt" else 0 - ), - ) - stdin_stream = process.stdin - assert stdin_stream is not None - assert process.stdout is not None - assert process.stderr is not None + # Keep the object before initialization can publish a PID and then unwind. + # Pre-PID failures remain the native constructor's boundary. + process = cast(subprocess.Popen[bytes], subprocess.Popen.__new__(subprocess.Popen)) + threads: list[threading.Thread] = [] + stdin_stream: IO[bytes] stdout = bytearray() stderr = bytearray() @@ -648,33 +635,53 @@ def write_stdin() -> None: except (OSError, ValueError): pass - threads = [ - threading.Thread( - target=read_stream, - args=(process.stdout, stdout), - kwargs={"overflow_kind": "response_too_large"}, - name="loopx-extension-stdout-reader", - daemon=True, - ), - threading.Thread( - target=read_stream, - args=(process.stderr, stderr), - kwargs={"overflow_kind": "stderr_too_large"}, - name="loopx-extension-stderr-reader", - daemon=True, - ), - threading.Thread( - target=write_stdin, - name="loopx-extension-stdin-writer", - daemon=True, - ), - ] - for thread in threads: - thread.start() - - deadline = None if timeout_seconds is None else time.monotonic() + timeout_seconds timed_out = False try: + subprocess.Popen.__init__( + process, + list(argv), + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=0, + env=dict(env) if env is not None else None, + cwd=cwd, + start_new_session=os.name == "posix", + creationflags=( + int(getattr(subprocess, "CREATE_NEW_PROCESS_GROUP")) + if os.name == "nt" else 0 + ), + ) + assert process.stdin is not None + stdin_stream = process.stdin + assert process.stdout is not None + assert process.stderr is not None + + threads = [ + threading.Thread( + target=read_stream, + args=(process.stdout, stdout), + kwargs={"overflow_kind": "response_too_large"}, + name="loopx-extension-stdout-reader", + daemon=True, + ), + threading.Thread( + target=read_stream, + args=(process.stderr, stderr), + kwargs={"overflow_kind": "stderr_too_large"}, + name="loopx-extension-stderr-reader", + daemon=True, + ), + threading.Thread( + target=write_stdin, + name="loopx-extension-stdin-writer", + daemon=True, + ), + ] + for thread in threads: + thread.start() + + deadline = None if timeout_seconds is None else time.monotonic() + timeout_seconds while process.poll() is None: remaining = None if deadline is None else deadline - time.monotonic() if remaining is not None and remaining <= 0: @@ -685,12 +692,17 @@ def write_stdin() -> None: terminate_process_tree(process, termination_grace_seconds) break except BaseException: - terminate_process_tree(process, termination_grace_seconds) + if getattr(process, "pid", None) is not None: + terminate_process_tree(process, termination_grace_seconds) raise finally: for thread in threads: - thread.join(timeout=_PROCESS_TERMINATE_GRACE_SECONDS) - for stream in (process.stdout, process.stderr): + if thread.ident is not None: + thread.join(timeout=_PROCESS_TERMINATE_GRACE_SECONDS) + for stream in (getattr(process, "stdin", None), getattr(process, "stdout", None), + getattr(process, "stderr", None)): + if stream is None: + continue try: stream.close() except (OSError, ValueError): diff --git a/tests/extensions/test_process_runtime_startup.py b/tests/extensions/test_process_runtime_startup.py new file mode 100644 index 0000000000..64e221fd0b --- /dev/null +++ b/tests/extensions/test_process_runtime_startup.py @@ -0,0 +1,192 @@ +"""A launched child stays owned when capped-process setup does not finish.""" +from __future__ import annotations + +import os +from pathlib import Path +import signal +import subprocess +import sys +import threading +from unittest.mock import Mock + +import pytest + +from loopx.extensions import process_runtime + + +@pytest.fixture +def owned_children(monkeypatch): + children = [] + yield children + # The historical implementation leaks these children. Keep the negative + # case safe too, without treating test cleanup as product cleanup. + monkeypatch.undo() + for child in children: + process_runtime.terminate_process_tree(child, grace_seconds=.1) + for stream in (child.stdin, child.stdout, child.stderr): + if stream is not None: + stream.close() + + +def _capture_child(monkeypatch, children, command, *, after_start=None): + initialize = subprocess.Popen.__init__ + + def capture(self, *args, **kwargs): + initialize(self, *args, **kwargs) + if self.args == command: + children.append(self) + if after_start is not None: + after_start(self) + + monkeypatch.setattr(subprocess.Popen, "__init__", capture) + + +def _assert_retired(children): + assert len(children) == 1 + child = children[0] + assert child.poll() is not None, "capped-process setup left its child running" + assert all(stream.closed for stream in (child.stdin, child.stdout, child.stderr)) + + +@pytest.mark.parametrize("error_type", [KeyboardInterrupt, OSError]) +@pytest.mark.parametrize("boundary", ["constructor_return", "native_start"]) +def test_constructor_unwind_after_launch_retires_acquired_child( + monkeypatch, owned_children, error_type, boundary, +): + command = [sys.executable, "-c", "import time; time.sleep(30)"] + failure = error_type("cancelled after the native launch returned a PID") + + def interrupt(_child): + raise failure + + if boundary == "constructor_return": + _capture_child(monkeypatch, owned_children, command, after_start=interrupt) + else: + execute = subprocess.Popen._execute_child + + def start_then_unwind(self, *args, **kwargs): + execute(self, *args, **kwargs) + if self.args == command: + # The native constructor now has a real PID, but its outer + # initialization (and exception cleanup) has not returned. + owned_children.append(self) + raise failure + + monkeypatch.setattr(subprocess.Popen, "_execute_child", start_then_unwind) + with pytest.raises(error_type) as raised: + process_runtime.run_capped_process( + command, stdin=b"", timeout_seconds=5, output_limit_bytes=1024, + termination_grace_seconds=.1, + ) + assert raised.value is failure + _assert_retired(owned_children) + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX cancellation signal") +def test_signal_after_launch_retires_child_before_cancel_returns(monkeypatch, owned_children): + command = [sys.executable, "-c", "import time; time.sleep(30)"] + failure = KeyboardInterrupt("caller cancelled") + + def cancelled(_number, _frame): + raise failure + + previous = signal.signal(signal.SIGTERM, cancelled) + try: + _capture_child(monkeypatch, owned_children, command, + after_start=lambda _child: os.kill(os.getpid(), signal.SIGTERM)) + with pytest.raises(KeyboardInterrupt) as raised: + process_runtime.run_capped_process( + command, stdin=b"", timeout_seconds=5, output_limit_bytes=1024, + termination_grace_seconds=.1, + ) + assert raised.value is failure + _assert_retired(owned_children) + finally: + signal.signal(signal.SIGTERM, previous) + + +@pytest.mark.parametrize("boundary", ["construct", "before_start", "after_start"]) +def test_partial_stream_setup_retires_child_and_closes_all_pipes( + monkeypatch, owned_children, boundary, +): + command = [sys.executable, "-c", "import time; time.sleep(30)"] + failure = RuntimeError("stream setup failed") + _capture_child(monkeypatch, owned_children, command) + initialize, start = threading.Thread.__init__, threading.Thread.start + threads = [] + + def construct(self, *args, **kwargs): + selected = kwargs.get("name", "").startswith("loopx-extension-") + if selected and boundary == "construct": + raise failure + initialize(self, *args, **kwargs) + if selected: + threads.append(self) + + def start_or_fail(self): + if self not in threads: + return start(self) + if boundary == "after_start": + start(self) + raise failure + + monkeypatch.setattr(threading.Thread, "__init__", construct) + monkeypatch.setattr(threading.Thread, "start", start_or_fail) + with pytest.raises(RuntimeError) as raised: + process_runtime.run_capped_process( + command, stdin=b"", timeout_seconds=5, output_limit_bytes=1024, + termination_grace_seconds=.1, + ) + assert raised.value is failure + _assert_retired(owned_children) + assert not any(thread.is_alive() for thread in threads) + + +def test_failure_before_any_pid_does_not_terminate_an_unowned_process(monkeypatch): + failure = OSError("launch was not attempted") + + def reject(self, *args, **kwargs): + raise failure + + retire = Mock() + monkeypatch.setattr(subprocess.Popen, "__init__", reject) + monkeypatch.setattr(process_runtime, "terminate_process_tree", retire) + with pytest.raises(OSError) as raised: + process_runtime.run_capped_process( + ["unused"], stdin=b"", timeout_seconds=5, output_limit_bytes=1024, + ) + assert raised.value is failure + retire.assert_not_called() + + +def test_failed_startup_cleanup_remains_an_error(monkeypatch, owned_children): + command = [sys.executable, "-c", "import time; time.sleep(30)"] + cancelled = KeyboardInterrupt("cancelled during setup") + unconfirmed = RuntimeError("owned child cleanup unconfirmed") + + def interrupt(_child): + raise cancelled + + _capture_child(monkeypatch, owned_children, command, after_start=interrupt) + monkeypatch.setattr(process_runtime, "terminate_process_tree", Mock(side_effect=unconfirmed)) + with pytest.raises(BaseException) as raised: + process_runtime.run_capped_process( + command, stdin=b"", timeout_seconds=5, output_limit_bytes=1024, + ) + assert raised.value is unconfirmed + assert raised.value.__context__ is cancelled + + +@pytest.mark.parametrize("timeout", [None, 5]) +def test_ordinary_launch_preserves_inputs_and_result(tmp_path: Path, timeout): + text = "complete input ☃\n".encode() + result = process_runtime.run_capped_process( + [sys.executable, "-c", + "import os,sys; assert os.getcwd() == sys.argv[1]; " + "assert os.environ['PROCESS_FIXTURE'] == 'present'; " + "sys.stdout.buffer.write(sys.stdin.buffer.read()); " + "sys.stderr.write('diagnostic'); sys.exit(3)", str(tmp_path)], + stdin=text, timeout_seconds=timeout, output_limit_bytes=1024, + cwd=tmp_path, env={**os.environ, "PROCESS_FIXTURE": "present"}, + ) + assert (result.returncode, result.stdout, result.failure_kind) == (3, text, None)