🐛 fix(query): respect timeout with child pipes

docs/changelog/+probe-descendant-pipes.bugfix.rst

@@ -0,0 +1 @@
+Return after an interpreter query times out without leaving pipe readers blocked by a child process.

src/python_discovery/_cached_py_info.py

@@ -10,11 +10,14 @@
 import secrets
 import sys
 import tempfile
+from codecs import getincrementaldecoder
 from collections import OrderedDict
 from contextlib import contextmanager, suppress
+from dataclasses import dataclass
 from functools import lru_cache
-from io import DEFAULT_BUFFER_SIZE
+from io import DEFAULT_BUFFER_SIZE, IncrementalNewlineDecoder
 from pathlib import Path
+from select import select
 from shlex import quote
 from subprocess import DEVNULL, PIPE, Popen, TimeoutExpired, list2cmdline  # ruff:ignore[suspicious-subprocess-import]
 from threading import Event, Thread
@@ -25,6 +28,9 @@
 from ._py_info import PythonInfo
 from ._py_info_collect import MIN_INTERROGATE_VERSION, UNSUPPORTED_EXIT_CODE, UNSUPPORTED_MARKER
 
+if sys.platform == "win32":  # pragma: win32 cover
+    from ._windows._cancel_io import stop_query_reader
+
 if TYPE_CHECKING:
     from collections.abc import Generator, Mapping
     from typing import IO, TypedDict
@@ -292,29 +298,45 @@ def _collect_query_output(process: Popen[str], timeout: float | None) -> tuple[s
         raise ValueError(msg)
     out_chunks: list[str] = []
     err_chunks: list[str] = []
-    overflow = Event()
-    read_errors: list[OSError] = []
+    state = _QueryReadState(process=process, overflow=Event(), stop=Event(), read_errors=[])
 
     readers = (
-        Thread(target=_drain_query_stream, args=(process, stdout, out_chunks, overflow, read_errors), daemon=True),
-        Thread(target=_drain_query_stream, args=(process, stderr, err_chunks, overflow, read_errors), daemon=True),
+        Thread(target=_drain_query_stream, args=(state, stdout, out_chunks), daemon=True),
+        Thread(target=_drain_query_stream, args=(state, stderr, err_chunks), daemon=True),
     )
     for reader in readers:
         reader.start()
     if not _query_completed(process, readers, timeout):
+        state.stop.set()
         with suppress(OSError):
             process.kill()
-        for reader in readers:
-            reader.join()
+        # A descendant may keep the pipes open after the candidate exits.
         process.wait()
+        _stop_query_readers(readers)
         return "", "timed out", -1
-    if read_errors:
-        raise read_errors[0]
-    if overflow.is_set():
+    if state.read_errors:
+        raise state.read_errors[0]
+    if state.overflow.is_set():
         return "", f"produced more than {_MAX_QUERY_OUTPUT} characters of output", -1
     return "".join(out_chunks), "".join(err_chunks), process.wait()
 
 
+@dataclass(frozen=True)
+class _QueryReadState:
+    process: Popen[str]
+    overflow: Event
+    stop: Event
+    read_errors: list[OSError]
+
+
+def _stop_query_readers(readers: tuple[Thread, Thread]) -> None:
+    for reader in readers:
+        if sys.platform == "win32":  # pragma: win32 cover
+            stop_query_reader(reader)
+        else:
+            reader.join()
+
+
 def _query_completed(process: Popen[str], readers: tuple[Thread, Thread], timeout: float | None) -> bool:
     deadline = None if timeout is None else monotonic() + timeout
     for reader in readers:
@@ -328,30 +350,34 @@ def _query_completed(process: Popen[str], readers: tuple[Thread, Thread], timeou
     return True
 
 
-def _drain_query_stream(
-    process: Popen[str], stream: IO[str], chunks: list[str], overflow: Event, read_errors: list[OSError]
-) -> None:
+def _drain_query_stream(state: _QueryReadState, stream: IO[str], chunks: list[str]) -> None:
     size = 0
     failed = False
+    decoder = IncrementalNewlineDecoder(getincrementaldecoder("utf-8")("backslashreplace"), translate=True)
     with stream:
-        while True:
+        while not state.stop.is_set():
             try:
-                chunk = stream.read(DEFAULT_BUFFER_SIZE)
+                if sys.platform != "win32" and not select([stream], [], [], 0.05)[0]:  # pragma: win32 cover
+                    continue
+                data = os.read(stream.fileno(), DEFAULT_BUFFER_SIZE)
             except OSError as error:
-                read_errors.append(error)
-                failed = True
+                if not state.stop.is_set():  # pragma: win32 cover  # canceled ReadFile raises on Windows
+                    state.read_errors.append(error)
+                    failed = True
                 break
-            if not chunk:
-                break
-            chunks.append(chunk[: _MAX_QUERY_OUTPUT + 1 - size])
-            size += len(chunk)
-            if size > _MAX_QUERY_OUTPUT:
-                overflow.set()
-                failed = True
+            chunk = decoder.decode(data, final=not data)
+            if chunk:
+                chunks.append(chunk[: _MAX_QUERY_OUTPUT + 1 - size])
+                size += len(chunk)
+                if size > _MAX_QUERY_OUTPUT:
+                    state.overflow.set()
+                    failed = True
+                    break
+            if not data:
                 break
     if failed:
         with suppress(OSError):
-            process.kill()
+            state.process.kill()
 
 
 @lru_cache(maxsize=128)

src/python_discovery/_windows/_cancel_io.py

@@ -0,0 +1,51 @@
+"""Cancel a blocked query pipe read on Windows."""
+
+from __future__ import annotations
+
+from ctypes import WinDLL, WinError, get_last_error, wintypes  # ty: ignore[unresolved-import]  # Windows-only APIs
+from time import monotonic
+from typing import TYPE_CHECKING, Final
+
+if TYPE_CHECKING:
+    from threading import Thread
+
+
+def stop_query_reader(reader: Thread) -> None:
+    deadline = monotonic() + 1
+    while reader.is_alive() and monotonic() < deadline:
+        if (thread_id := reader.native_id) is not None:
+            try:
+                _cancel_synchronous_io(thread_id)
+            except OSError:
+                if reader.is_alive():
+                    raise
+        reader.join(0.05)
+    if reader.is_alive():
+        msg = "query output reader did not stop"
+        raise TimeoutError(msg)
+
+
+def _cancel_synchronous_io(thread_id: int) -> None:
+    error_not_found: Final[int] = 1168
+    kernel = WinDLL("kernel32", use_last_error=True)
+    open_thread = kernel.OpenThread
+    open_thread.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD)
+    open_thread.restype = wintypes.HANDLE
+    cancel = kernel.CancelSynchronousIo
+    cancel.argtypes = (wintypes.HANDLE,)
+    cancel.restype = wintypes.BOOL
+    close = kernel.CloseHandle
+    close.argtypes = (wintypes.HANDLE,)
+    close.restype = wintypes.BOOL
+
+    if not (handle := open_thread(0x0001, wintypes.BOOL(0), thread_id)):
+        raise WinError(get_last_error())
+    try:
+        if not cancel(handle) and (error := get_last_error()) != error_not_found:
+            raise WinError(error)
+    finally:
+        if not close(handle):
+            raise WinError(get_last_error())
+
+
+__all__ = ["stop_query_reader"]

tests/test_cached_py_info.py

@@ -6,8 +6,11 @@
 import sys
 import tempfile
 from io import StringIO
+from os import read
 from pathlib import Path
 from subprocess import DEVNULL, PIPE, Popen, TimeoutExpired, list2cmdline
+from threading import enumerate as active_threads
+from time import monotonic
 from typing import TYPE_CHECKING, Final
 from unittest.mock import MagicMock, patch
 
@@ -28,6 +31,8 @@
 )
 
 if TYPE_CHECKING:
+    from typing import IO
+
     from pytest_mock import MockerFixture
 
 
@@ -97,7 +102,13 @@ def test_run_subprocess_invalid_json(mocker: MockerFixture) -> None:
 
 
 def _mock_process(out: str, err: str, code: int) -> MagicMock:
-    return MagicMock(stdout=StringIO(out), stderr=StringIO(err), returncode=code, **{"wait.return_value": code})
+    streams: list[IO[str]] = []
+    for value in (out, err):
+        read_fd, write_fd = os.pipe()
+        with os.fdopen(write_fd, "wb") as writer:
+            writer.write(value.encode())
+        streams.append(os.fdopen(read_fd, "r", encoding="utf-8"))
+    return MagicMock(stdout=streams[0], stderr=streams[1], returncode=code, **{"wait.return_value": code})
 
 
 def test_run_subprocess_with_cookies(mocker: MockerFixture) -> None:
@@ -161,6 +172,15 @@ def test_collect_query_output_limit(stream: str, timeout: float | None) -> None:
     )
 
 
+def test_collect_query_output_normalizes_trailing_newline() -> None:
+    assert _collect_query_output(_mock_process("\r", "", 0), 1) == ("\n", "", 0)
+
+
+def test_collect_query_output_decodes_split_utf8() -> None:
+    with patch("python_discovery._cached_py_info.os.read", side_effect=lambda fd, _: read(fd, 1)):
+        assert _collect_query_output(_mock_process("é", "", 0), 1) == ("é", "", 0)
+
+
 @pytest.mark.parametrize(
     "code",
     [
@@ -173,25 +193,41 @@ def test_collect_query_output_timeout(code: str) -> None:
     assert _collect_query_output(process, 0.1) == ("", "timed out", -1)
 
 
-def test_collect_query_output_timeout_after_eof(mocker: MockerFixture) -> None:
-    process = mocker.MagicMock(
-        stdout=StringIO(""),
-        stderr=StringIO(""),
-        **{"wait.side_effect": [TimeoutExpired("python", 0), 0]},
+def test_collect_query_output_timeout_reaps_readers() -> None:
+    code = (
+        "import subprocess, sys\n"
+        "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(10)'], "
+        "stdin=subprocess.DEVNULL, stdout=sys.stdout, stderr=sys.stderr)\n"
     )
+    process = Popen([sys.executable, "-c", code], stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True)
+    process.wait(timeout=5)
+    before = set(active_threads())
+    started = monotonic()
+
+    assert _collect_query_output(process, 0.1) == ("", "timed out", -1)
+    assert monotonic() - started < 2.5
+    assert not set(active_threads()) - before
+
+
+def test_collect_query_output_timeout_after_eof(mocker: MockerFixture) -> None:
+    process = Popen([sys.executable, "-c", ""], stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True)
+    process.wait(timeout=5)
+    mocker.patch.object(process, "wait", side_effect=[TimeoutExpired("python", 0), 0])
+    mocker.patch.object(process, "kill")
 
     assert _collect_query_output(process, 1) == ("", "timed out", -1)
     process.kill.assert_called_once()
 
 
 def test_collect_query_output_read_error(mocker: MockerFixture) -> None:
-    stream = mocker.MagicMock()
-    stream.read.side_effect = OSError("broken pipe")
-    process = mocker.MagicMock(stdout=stream, stderr=StringIO(""), **{"wait.return_value": -9})
+    process = Popen([sys.executable, "-c", ""], stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True)
+    process.wait(timeout=5)
+    mocker.patch("python_discovery._cached_py_info.os.read", side_effect=OSError("broken pipe"))
+    mocker.patch.object(process, "kill")
 
     with pytest.raises(OSError, match="broken pipe"):
         _collect_query_output(process, 1)
-    process.kill.assert_called_once()
+    process.kill.assert_called()
 
 
 @pytest.mark.parametrize(

tests/test_query_timeout.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import sys
+from time import monotonic
 from typing import TYPE_CHECKING
 
 import pytest
@@ -11,6 +12,8 @@
 from python_discovery._cached_py_info import _collect_query_output
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
     from pytest_mock import MockerFixture
 
 
@@ -86,3 +89,22 @@ def test_from_exe_timeout_warning_cache_is_bounded(caplog: pytest.LogCaptureFixt
     assert [record.getMessage() for record in caplog.records] == [
         "ignoring PY_DISCOVERY_TIMEOUT='invalid-timeout-0', not a positive number of seconds; using 15.0"
     ]
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell launcher")
+def test_from_exe_timeout_with_inherited_pipes(tmp_path: Path) -> None:
+    executable = tmp_path / "python3.11"
+    executable.write_text("#!/bin/sh\n/bin/sleep 3 &\nexit 0\n")
+    executable.chmod(0o755)
+
+    started = monotonic()
+    result = PythonInfo.from_exe(
+        str(executable),
+        env={**os.environ, "PY_DISCOVERY_TIMEOUT": "0.1"},
+        ignore_cache=True,
+        raise_on_error=False,
+        resolve_to_host=False,
+    )
+
+    assert result is None
+    assert monotonic() - started < 2.5