diff --git a/tests/_worker_io.py b/tests/_worker_io.py new file mode 100644 index 0000000..5e27411 --- /dev/null +++ b/tests/_worker_io.py @@ -0,0 +1,31 @@ +"""Timeout-guarded read from a worker subprocess's stdout. + +A bare ``proc.stdout.readline()`` blocks forever if the worker never writes the +expected line (protocol mismatch, crash before flushing, deadlock) — the worker +stays alive waiting on stdin, so the pipe never hits EOF either. One such stuck +assertion turned a single flaky test into a CI job stuck for hours (until GitHub's +default 6h job timeout, or a human, killed it). Every worker-subprocess test should +read through this instead of calling ``proc.stdout.readline()`` directly. + +Reads run in a helper thread rather than gating on ``select()`` on the raw fd: +``proc.stdout`` is a buffered ``TextIOWrapper``, so a burst of several JSON lines +delivered in one OS-level read can sit fully in its internal buffer — ``select()`` +on the fd then reports "not ready" (nothing new at the OS level) even though +``readline()`` would return one of those lines instantly. +""" +from __future__ import annotations + +import queue +import subprocess +import threading + +DEFAULT_TIMEOUT = 15.0 + + +def readline_timeout(proc: subprocess.Popen, timeout: float = DEFAULT_TIMEOUT) -> str: + result: "queue.Queue[str]" = queue.Queue(maxsize=1) + threading.Thread(target=lambda: result.put(proc.stdout.readline()), daemon=True).start() + try: + return result.get(timeout=timeout) + except queue.Empty: + raise TimeoutError(f"worker produced no output within {timeout}s (pid={proc.pid})") from None diff --git a/tests/test_debugging.py b/tests/test_debugging.py index 0d78598..fed41d4 100644 --- a/tests/test_debugging.py +++ b/tests/test_debugging.py @@ -13,6 +13,7 @@ from pathlib import Path import protocol as P +from tests._worker_io import readline_timeout APP_DIR = Path(__file__).resolve().parent.parent @@ -25,7 +26,7 @@ def _worker(entry: str) -> Iterator[subprocess.Popen]: env={**os.environ}, ) try: - proc.stdout.readline() # startup graph topology + readline_timeout(proc) # startup graph topology yield proc finally: proc.kill() @@ -39,7 +40,7 @@ def _send(proc: subprocess.Popen, obj: dict) -> None: def _read_until(proc: subprocess.Popen, type_: str): for _ in range(80): - line = proc.stdout.readline() + line = readline_timeout(proc) if not line: break ev = P.ServerEventAdapter.validate_python(json.loads(line)) @@ -87,8 +88,8 @@ def test_ac03_fork_time_travel_with_override() -> None: _read_until(proc, "state_snapshot") # fork from there with an override; re-runs b, c on a new runId _send(proc, {"cmd": "fork", "checkpointId": ckpt, "stateOverride": {"log": ["X"]}}) + final = _read_until(proc, "state_snapshot") # final state (emitted before run_finished) _read_until(proc, "run_finished") # fork run finished - final = _read_until(proc, "state_snapshot") assert "X" in final.snapshot.values["log"] # override is in the forked state diff --git a/tests/test_entry_wizard.py b/tests/test_entry_wizard.py index 7f6ca6e..977b9dd 100644 --- a/tests/test_entry_wizard.py +++ b/tests/test_entry_wizard.py @@ -15,6 +15,7 @@ import protocol as P from graphloupe_sidecar import discover +from tests._worker_io import readline_timeout APP_DIR = Path(__file__).resolve().parent.parent @@ -119,7 +120,7 @@ def _load_via_adapter(tmp_path: Path, symbol: str) -> P.ServerEvent: gl.mkdir() (gl / "entry.py").write_text(ADAPTER.format(symbol=symbol), encoding="utf-8") with _worker("entry:build_graph", tmp_path) as proc: - return P.ServerEventAdapter.validate_python(json.loads(proc.stdout.readline())) + return P.ServerEventAdapter.validate_python(json.loads(readline_timeout(proc))) def test_adapter_loads_compiled_graph_variable(tmp_path: Path) -> None: diff --git a/tests/test_graph_loading.py b/tests/test_graph_loading.py index 835f9f0..406a064 100644 --- a/tests/test_graph_loading.py +++ b/tests/test_graph_loading.py @@ -18,6 +18,7 @@ import protocol as P from graphloupe_sidecar.server import app +from tests._worker_io import readline_timeout APP_DIR = Path(__file__).resolve().parent.parent @@ -37,7 +38,7 @@ def _worker(entry: str, env_extra: dict[str, str] | None = None) -> Iterator[sub def _first(proc: subprocess.Popen): - return P.ServerEventAdapter.validate_python(json.loads(proc.stdout.readline())) + return P.ServerEventAdapter.validate_python(json.loads(readline_timeout(proc))) def test_ac01_custom_entry_topology() -> None: diff --git a/tests/test_health_check.py b/tests/test_health_check.py index 0567c82..2105420 100644 --- a/tests/test_health_check.py +++ b/tests/test_health_check.py @@ -11,6 +11,7 @@ from pathlib import Path import protocol as P +from tests._worker_io import readline_timeout APP_DIR = Path(__file__).resolve().parent.parent @@ -53,7 +54,7 @@ def _worker(entry: str, root: Path) -> Iterator[subprocess.Popen]: def _topology(src: str, tmp_path: Path) -> P.GraphTopology: (tmp_path / "g.py").write_text(src, encoding="utf-8") with _worker("g:build_graph", tmp_path) as proc: - msg = P.ServerEventAdapter.validate_python(json.loads(proc.stdout.readline())) + msg = P.ServerEventAdapter.validate_python(json.loads(readline_timeout(proc))) assert isinstance(msg, P.GraphTopology) return msg diff --git a/tests/test_manual_inference.py b/tests/test_manual_inference.py index 9fda1b9..31b26ae 100644 --- a/tests/test_manual_inference.py +++ b/tests/test_manual_inference.py @@ -18,6 +18,7 @@ import protocol as P from graphloupe_sidecar.graph import build_manual_graph +from tests._worker_io import readline_timeout APP_DIR = Path(__file__).resolve().parent.parent @@ -46,7 +47,7 @@ def _worker(entry: str) -> Iterator[subprocess.Popen]: env={**os.environ}, ) try: - proc.stdout.readline() # consume the startup graph topology line + readline_timeout(proc) # consume the startup graph topology line yield proc finally: proc.kill() @@ -60,7 +61,7 @@ def _send(proc: subprocess.Popen, obj: dict) -> None: def _read_until(proc: subprocess.Popen, type_: str): for _ in range(50): - line = proc.stdout.readline() + line = readline_timeout(proc) if not line: break ev = P.ServerEventAdapter.validate_python(json.loads(line)) diff --git a/tests/test_workspace_load.py b/tests/test_workspace_load.py index 0dbc7ae..2526038 100644 --- a/tests/test_workspace_load.py +++ b/tests/test_workspace_load.py @@ -15,6 +15,7 @@ from pathlib import Path import protocol as P +from tests._worker_io import readline_timeout APP_DIR = Path(__file__).resolve().parent.parent @@ -54,7 +55,7 @@ def _worker(entry: str, root: Path) -> Iterator[subprocess.Popen]: def _first(proc: subprocess.Popen): - return P.ServerEventAdapter.validate_python(json.loads(proc.stdout.readline())) + return P.ServerEventAdapter.validate_python(json.loads(readline_timeout(proc))) def test_ac01_loads_graph_from_project_root(tmp_path: Path) -> None: