diff --git a/scripts/nvx_tools/adversarial.py b/scripts/nvx_tools/adversarial.py index 9439e8822..45f178a1a 100644 --- a/scripts/nvx_tools/adversarial.py +++ b/scripts/nvx_tools/adversarial.py @@ -485,6 +485,8 @@ def __init__( self._stdin = self._process.stdin self._stdout = self._process.stdout self._responses: queue.Queue[bytes | None] = queue.Queue() + # Set when close gives up on a reader that a surviving process keeps blocked. + self._abandoned = False self._reader = threading.Thread( target=self._read_responses, name=f"nvx-executor-reader-{label}", @@ -563,13 +565,22 @@ def close(self) -> None: cleanup_error = error finally: self._closed = True - for stream in (self._stdin, self._stdout, self._stderr_stream): + for stream in (self._stdin, self._stderr_stream): try: stream.close() except OSError as error: if cleanup_error is None: cleanup_error = error self._reader.join(timeout=2.0) + if self._reader.is_alive(): + # A process that outlived the wrapper still holds stdout. Closing the buffered + # stream would wait for the blocked reader, so the reader closes it when it + # ends, and it retains nothing further. + self._abandoned = True + if cleanup_error is None: + cleanup_error = ScriptError( + "executor output is still held open by a surviving process" + ) if cleanup_error is not None: raise cleanup_error @@ -577,12 +588,19 @@ def abort(self) -> None: self._terminate() def _read_responses(self) -> None: - while line := self._stdout.readline(MAX_PROTOCOL_LINE_BYTES + 1): - if len(line) > MAX_PROTOCOL_LINE_BYTES: - self._responses.put(None) - return - self._responses.put(line) - self._responses.put(None) + try: + while line := self._stdout.readline(MAX_PROTOCOL_LINE_BYTES + 1): + if self._abandoned: + # Keep draining so a surviving process is not blocked on a full pipe, + # but retain nothing once the client has been closed. + continue + if len(line) > MAX_PROTOCOL_LINE_BYTES: + self._responses.put(None) + return + self._responses.put(line) + self._responses.put(None) + finally: + self._stdout.close() def _terminate(self) -> None: if self._process.poll() is None: diff --git a/scripts/test_adversarial.py b/scripts/test_adversarial.py index 996eba46b..9af8734e0 100644 --- a/scripts/test_adversarial.py +++ b/scripts/test_adversarial.py @@ -8,6 +8,7 @@ import hashlib import json import os +import queue import signal import socket import subprocess @@ -23,6 +24,7 @@ from nvx_tools.adversarial import ( LOCAL_EXECUTOR_STATE_ROOT, + SCHEMA_VERSION, CampaignConfig, CopilotController, ExecutorClient, @@ -2764,5 +2766,82 @@ def fake_run( self.assertIn(case_id, repair_prompt) +class ExecutorCloseTests(unittest.TestCase): + def test_close_is_bounded_when_a_surviving_child_holds_executor_stdout(self): + # The wrapper exits after starting a child that inherits its stdout, so the protocol + # reader stays blocked. Closing must not wait for that child to exit. + with tempfile.TemporaryDirectory(prefix="nvx-executor-close-") as temporary: + root = Path(temporary) + run_dir = root / "run" + run_dir.mkdir() + pid_file = root / "child.pid" + child = ( + f'"{sys.executable}" -c "import os, sys, time; ' + "open(sys.argv[1], 'w').write(str(os.getpid())); time.sleep(20)\"" + f' "{pid_file}"' + ) + if os.name == "nt": + wrapper = root / "wrapper.cmd" + wrapper.write_bytes(f'@echo off\r\nstart "" /B {child}\r\n'.encode()) + else: + wrapper = root / "wrapper.sh" + wrapper.write_bytes(f"#!/bin/sh\n{child} &\n".encode()) + wrapper.chmod(0o755) + client = ExecutorClient( + run_dir=run_dir, + command=str(wrapper), + label="surviving-child", + ) + try: + with self.assertRaises(ExecutorUnavailableError): + client.request( + {"schema_version": SCHEMA_VERSION, "operation": "shutdown"}, + timeout=0.5, + ) + started = time.monotonic() + # If the wrapper exited before request() flushed stdin, close() can see the + # buffered stdin data fail with BrokenPipeError instead of the stdout ScriptError. + with self.assertRaises((ScriptError, BrokenPipeError)): + client.close() + self.assertLess(time.monotonic() - started, 8.0) + finally: + # The surviving child holds the log and must not outlive the test. + deadline = time.monotonic() + 5.0 + while not pid_file.exists() and time.monotonic() < deadline: + time.sleep(0.05) + if pid_file.exists(): + pid = pid_file.read_text(encoding="utf-8") + if os.name == "nt": + subprocess.run( + ["taskkill.exe", "/PID", pid, "/T", "/F"], + check=False, + capture_output=True, + ) + else: + os.kill(int(pid), signal.SIGKILL) + + def test_reader_closes_its_stream_when_the_output_ends(self): + client = ExecutorClient.__new__(ExecutorClient) + client._stdout = MagicMock() + client._stdout.readline.side_effect = [b'{"ok": true}\n', b""] + client._responses = queue.Queue() + client._abandoned = False + client._read_responses() + client._stdout.close.assert_called_once_with() + self.assertEqual(client._responses.get_nowait(), b'{"ok": true}\n') + self.assertIsNone(client._responses.get_nowait()) + + def test_abandoned_reader_drains_output_without_retaining_it(self): + client = ExecutorClient.__new__(ExecutorClient) + client._stdout = MagicMock() + client._stdout.readline.side_effect = [b"late\n", b"later\n", b""] + client._responses = queue.Queue() + client._abandoned = True + client._read_responses() + client._stdout.close.assert_called_once_with() + self.assertIsNone(client._responses.get_nowait()) + self.assertTrue(client._responses.empty()) + + if __name__ == "__main__": unittest.main()