diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index ad389fd4cf..b4ffef0eab 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -431,6 +431,18 @@ it differs from the live server before dispatching expensive work. Stop only that isolated runtime after its requests settle, retaining the same temporary environment for shutdown; never restart a live server as test cleanup. +For namespace disposal, call `restart_effect_runtime(wait_for_exit=True)` and +require `stopped` before removing files. A closed listener or a zombie process +group leader does not prove that all runtime threads have exited. A PID that is +still present returns `shutdown_pending`; retain the namespace and retry after +the process exits and its parent reaps it. The ordinary restart action still +uses serving-locator retirement. + +删除隔离目录前，调用 `restart_effect_runtime(wait_for_exit=True)` 并确认返回 +`stopped`。端口关闭或进程组长成为僵尸进程，都不能证明其他线程已经退出。 +PID 仍存在时返回 `shutdown_pending`；保留目录，等进程退出并由父进程回收后再试。 +普通重启操作仍按服务 locator 的退役状态判断。 + Automatic test cleanup must verify namespace ownership as well as PID/token identity. An ordinary copy of a valid locator inside a test directory is not permission to stop its original server. The private authenticated `runtime.ping` diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index b5c78ed674..a5be09479d 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -771,22 +771,7 @@ def _stop_effect_runtime( if not _pid_is_alive(pid): stopped = True break - if os.name == "posix": - # A non-child zombie cannot write, but kill(pid, 0) still sees - # it. Managed runtimes start in their own session/process group. - from ..extensions.process_runtime import _posix_owned_group_has_exited - try: - stopped = _posix_owned_group_has_exited( - pid, max(.001, deadline - time.monotonic()), - ) - except (OSError, subprocess.TimeoutExpired, UnicodeError, RuntimeError): - stopped = False - if stopped: - # The child may have exited during the group snapshot, - # after the first nonblocking reap. Retire its wait status - # before returning successful namespace cleanup. - _reap_exited_runtime_child(info) - break + # A zombie leader can still have threads writing into the namespace. time.sleep(.05) continue published, published_token = _serving_token(info_path) diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index 681d147346..0d46550bf3 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -33,8 +33,9 @@ @pytest.mark.parametrize("replacement", [False, True]) +@pytest.mark.parametrize("owns_group", [True, False] if os.name == "posix" else [False]) def test_fixture_shutdown_waits_for_process_exit_after_locator_turnover( - tmp_path, monkeypatch, replacement, + tmp_path, monkeypatch, replacement, owns_group, ): locator = tmp_path / "runtime.json" if replacement: @@ -42,7 +43,7 @@ def test_fixture_shutdown_waits_for_process_exit_after_locator_turnover( monkeypatch.setattr(effect_runtime, "_request_with_info", lambda *a, **kw: {}) with subprocess.Popen( [sys.executable, "-c", "import time; print('ready',flush=True); time.sleep(60)"], - stdout=subprocess.PIPE, text=True, start_new_session=os.name == "posix", + stdout=subprocess.PIPE, text=True, start_new_session=owns_group, ) as process: try: assert process.stdout.readline().strip() == "ready" @@ -69,6 +70,50 @@ def test_fixture_shutdown_waits_for_process_exit_after_locator_turnover( process.wait() +@pytest.mark.skipif(sys.platform != "linux", reason="Linux thread-group leader exit") +def test_fixture_shutdown_waits_for_writer_after_thread_group_leader_exits(tmp_path, monkeypatch): + witness = tmp_path / "written-after-stop" + monkeypatch.setattr(effect_runtime, "_request_with_info", lambda *a, **kw: {}) + source = """ +import ctypes, os, pathlib, sys, threading +def write_after_release(): + print('ready', flush=True) + sys.stdin.readline() + pathlib.Path(sys.argv[1]).write_text('worker can still write') + os._exit(0) +threading.Thread(target=write_after_release).start() +ctypes.CDLL(None).pthread_exit(None) +""" + with subprocess.Popen( + [sys.executable, "-c", source, str(witness)], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, start_new_session=True, + ) as process: + try: + assert process.stdout.readline().strip() == "ready" + state = Path(f"/proc/{process.pid}/stat") + deadline = time.monotonic() + 5 + while state.read_text().rsplit(")", 1)[1].split()[0] != "Z": + assert time.monotonic() < deadline, "thread-group leader did not exit" + time.sleep(.01) + info = {"pid": process.pid, "token": "original"} + stopped = effect_runtime._stop_effect_runtime( + tmp_path / "retired-locator", info, timeout=.1, wait_for_process_exit=True, + ) + assert stopped["status"] == "shutdown_pending" + assert effect_runtime._pid_is_alive(process.pid) + process.stdin.write("release\n") + process.stdin.flush() + process.wait(timeout=5) + assert witness.read_text() == "worker can still write" + assert effect_runtime._stop_effect_runtime( + tmp_path / "retired-locator", info, timeout=.1, wait_for_process_exit=True, + )["status"] == "stopped" + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + def _effect_id(label: str) -> str: return f"fixture-goal:fixture-agent:{_TODO_ID}:{label}" diff --git a/tests/control_plane/test_effect_runtime_restart.py b/tests/control_plane/test_effect_runtime_restart.py index ce45daa315..a8af8b05e6 100644 --- a/tests/control_plane/test_effect_runtime_restart.py +++ b/tests/control_plane/test_effect_runtime_restart.py @@ -354,42 +354,30 @@ def unavailable(pid, timeout): @pytest.mark.skipif(os.name != "posix" or not hasattr(os, "WNOWAIT"), reason="POSIX child exit observation") -def test_namespace_shutdown_reaps_child_exiting_during_group_observation(tmp_path, monkeypatch): - from loopx.extensions import process_runtime - +def test_namespace_shutdown_reaps_exited_child_before_disposal(tmp_path, monkeypatch): child = subprocess.Popen( [sys.executable, "-c", "import sys; sys.stdin.read()"], stdin=subprocess.PIPE, start_new_session=True, ) info = {"pid": child.pid, "token": "serving"} - observe_group = process_runtime._posix_owned_group_has_exited - requests, observations = [], [] + requests = [] def shutdown(_info, **kwargs): requests.append(kwargs["method"]) - # Keep the child live through the first nonblocking reap. It exits - # during the following real process-group observation instead. - return {} - - def exited_during_observation(pid, timeout): - assert pid == child.pid and effect_runtime._pid_is_alive(pid) + assert effect_runtime._pid_is_alive(child.pid) child.stdin.close() - # Observe the real exit without reaping it. kill(pid, 0) still sees - # this zombie, while the group observer correctly sees no writer. - os.waitid(os.P_PID, pid, os.WEXITED | os.WNOWAIT) - exited = observe_group(pid, timeout) - observations.append(exited) - assert exited and effect_runtime._pid_is_alive(pid) - return exited + # Wait for the shutdown response's real exit without reaping the child. + os.waitid(os.P_PID, child.pid, os.WEXITED | os.WNOWAIT) + assert effect_runtime._pid_is_alive(child.pid) + return {} monkeypatch.setattr(effect_runtime, "_request_with_info", shutdown) - monkeypatch.setattr(process_runtime, "_posix_owned_group_has_exited", exited_during_observation) try: result = effect_runtime._stop_effect_runtime( tmp_path / "runtime.json", info, timeout=5, wait_for_process_exit=True, ) assert result["stopped"] and result["status"] == "stopped" - assert requests == ["runtime.shutdown"] and observations == [True] + assert requests == ["runtime.shutdown"] assert not effect_runtime._pid_is_alive(child.pid), "Exited child must be reaped before cleanup returns" with pytest.raises(ChildProcessError): os.waitpid(child.pid, os.WNOHANG)