From 0a3b0efcc6af2e9b107d106ce4c78260bf620024 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 16:40:05 -0400 Subject: [PATCH] Reap timed-out docs example trees Always drain docs-example pipes, even when the process exited before the first status check, and decode malformed output with replacement so diagnostics preserve the original failure. Run POSIX examples in dedicated sessions and kill the full process group on timeout. Reap the leader in every exit path, with bounded Windows cleanup that cannot wait forever on descendant-held pipes. Cover fast non-zero exits, invalid output bytes, process-group termination, and post-timeout reaping. Review: PR #480 (goodboy,copilot-pull-request-reviewer[bot]) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/test_docs_examples.py | 248 +++++++++++++++++++++++++++--------- 1 file changed, 189 insertions(+), 59 deletions(-) diff --git a/tests/test_docs_examples.py b/tests/test_docs_examples.py index 90081a49..f23fc926 100644 --- a/tests/test_docs_examples.py +++ b/tests/test_docs_examples.py @@ -5,11 +5,13 @@ Let's make sure them docs work yah? from contextlib import contextmanager import itertools import os +import signal import sys import subprocess import platform import shutil from typing import Callable +from unittest.mock import Mock import pytest import tractor @@ -21,6 +23,181 @@ _non_linux: bool = platform.system() != 'Linux' _friggin_macos: bool = platform.system() == 'Darwin' +def _kill_proc_tree(proc: subprocess.Popen) -> None: + ''' + Terminate an example process and its POSIX descendants. + + ''' + try: + if platform.system() == 'Windows': + proc.kill() + else: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + +def _reap_killed_proc( + proc: subprocess.Popen, +) -> tuple[bytes, bytes]: + ''' + Reap a killed process without waiting on Windows descendants. + + ''' + if platform.system() != 'Windows': + return proc.communicate() + + proc.wait(timeout=5) + if proc.stdout: + proc.stdout.close() + if proc.stderr: + proc.stderr.close() + return b'', b'' + + +def _wait_for_proc( + proc: subprocess.Popen, + timeout: float, + test_log: tractor.log.StackLevelAdapter, +) -> None: + ''' + Wait for an example process and surface its captured output. + + ''' + try: + out, err = proc.communicate(timeout=timeout) + + except subprocess.TimeoutExpired as timeout_exc: + test_log.exception( + f'Example failed to finish within {timeout}s ??\n' + ) + _kill_proc_tree(proc) + out, err = _reap_killed_proc(proc) + if platform.system() == 'Windows': + out = timeout_exc.output or b'' + err = timeout_exc.stderr or b'' + + errmsg: str = err.decode(errors='replace') + + # XXX, ALWAYS surface the subproc's full stderr + # whenever it exits non-zero! + # + # The prior impl only raised when the LAST stderr + # line contained 'Error', swallowing any crash whose + # traceback ends in a non-`XxxError:` line; in + # particular EVERY `tractor` root-actor crash ends + # with the strict-EG collapse note, + # '( ^^^ this exc was collapsed from a group ^^^ )', + # so ALL such failures were reduced to a bare + # `assert 1 == 0` in CI logs.. see GH #473. + rc: int|None = proc.returncode + if rc: + outmsg: str = out.decode(errors='replace') + raise Exception( + f'Example script exited with rc={rc} !?\n' + f'\n' + f'stdout:\n' + f'{outmsg}\n' + f'\n' + f'stderr:\n' + f'{errmsg}\n' + ) + + # if we get some gnarly output let's aggregate and raise + if errmsg: + errlines = errmsg.splitlines() + last_error = errlines[-1] + if ( + 'Error' in last_error + + # XXX: currently we print this to console, but maybe + # shouldn't eventually once we figure out what's + # a better way to be explicit about aio side + # cancels? + and + 'asyncio.exceptions.CancelledError' not in last_error + ): + raise Exception(errmsg) + + assert proc.returncode == 0 + + +def test_wait_for_failed_example_captures_output(): + ''' + Preserve diagnostics from a subprocess which already exited. + + The previous `poll()` guard skipped `communicate()` when a fast + failure returned a non-zero status before the parent checked it. + Its stdout and stderr were therefore reported as empty. This + fake process begins with `returncode=1` and returns non-UTF-8 + output, proving the helper always drains both pipes and replaces + undecodable bytes without hiding the original process failure. + + ''' + proc = Mock() + proc.returncode = 1 + proc.communicate.return_value = ( + b'stdout\xff', + b'stderr\xff', + ) + + with pytest.raises(Exception) as exc_info: + _wait_for_proc( + proc=proc, + timeout=1, + test_log=Mock(), + ) + + proc.communicate.assert_called_once_with(timeout=1) + errmsg: str = str(exc_info.value) + assert 'stdout\ufffd' in errmsg + assert 'stderr\ufffd' in errmsg + + +@pytest.mark.skipif( + platform.system() == 'Windows', + reason='POSIX process groups are unavailable on Windows', +) +def test_wait_for_timed_out_example_reaps_group( + monkeypatch: pytest.MonkeyPatch, +): + ''' + Kill the example process group and reap its leader on timeout. + + The old timeout branch killed only the immediate process and + never drained it. Actor descendants could retain the capture + pipes while the leader remained unreaped, hanging CI until its + job timeout. This fake process raises `TimeoutExpired` on the + timed wait and completes on the second `communicate()` call; + the assertions prove group-directed `SIGKILL` precedes that + final drain and leaves a concrete non-zero return code. + + ''' + proc = Mock() + proc.pid = 1234 + + def communicate(timeout=None): + if timeout is not None: + raise subprocess.TimeoutExpired('example', timeout) + proc.returncode = -signal.SIGKILL + return b'', b'timed out' + + proc.communicate.side_effect = communicate + killpg = Mock() + monkeypatch.setattr(os, 'killpg', killpg) + + with pytest.raises(Exception, match='timed out'): + _wait_for_proc( + proc=proc, + timeout=.01, + test_log=Mock(), + ) + + killpg.assert_called_once_with(1234, signal.SIGKILL) + assert proc.communicate.call_count == 2 + assert proc.returncode == -signal.SIGKILL + + @pytest.fixture def run_example_in_subproc( loglevel: str, @@ -61,6 +238,7 @@ def run_example_in_subproc( ] else: script_file = testdir.makefile('.py', script_code) + kwargs['start_new_session'] = True cmdargs = [ sys.executable, str(script_file), @@ -77,9 +255,12 @@ def run_example_in_subproc( **kwargs, ) assert not proc.returncode - yield proc - proc.wait() - assert proc.returncode == 0 + try: + yield proc + finally: + if proc.poll() is None: + _kill_proc_tree(proc) + _reap_killed_proc(proc) yield run @@ -163,59 +344,8 @@ def test_example( code = ex.read() with run_example_in_subproc(code) as proc: - out = None - err = None - try: - if not proc.poll(): - out, err = proc.communicate(timeout=timeout) - - except subprocess.TimeoutExpired as e: - test_log.exception( - f'Example failed to finish within {timeout}s ??\n' - ) - proc.kill() - err = e.stderr - - errmsg: str = err.decode() if err else '' - - # XXX, ALWAYS surface the subproc's full stderr - # whenever it exits non-zero! - # - # The prior impl only raised when the LAST stderr - # line contained 'Error', swallowing any crash whose - # traceback ends in a non-`XxxError:` line; in - # particular EVERY `tractor` root-actor crash ends - # with the strict-EG collapse note, - # '( ^^^ this exc was collapsed from a group ^^^ )', - # so ALL such failures were reduced to a bare - # `assert 1 == 0` in CI logs.. see GH #473. - rc: int|None = proc.returncode - if rc: - outmsg: str = out.decode() if out else '' - raise Exception( - f'Example script exited with rc={rc} !?\n' - f'\n' - f'stdout:\n' - f'{outmsg}\n' - f'\n' - f'stderr:\n' - f'{errmsg}\n' - ) - - # if we get some gnarly output let's aggregate and raise - if errmsg: - errlines = errmsg.splitlines() - last_error = errlines[-1] - if ( - 'Error' in last_error - - # XXX: currently we print this to console, but maybe - # shouldn't eventually once we figure out what's - # a better way to be explicit about aio side - # cancels? - and - 'asyncio.exceptions.CancelledError' not in last_error - ): - raise Exception(errmsg) - - assert proc.returncode == 0 + _wait_for_proc( + proc=proc, + timeout=timeout, + test_log=test_log, + )