From 6bfa992f55a7ea2b3821f2e1872dbccfcfe97297 Mon Sep 17 00:00:00 2001 From: Chao Wang <26245345+ChaoWao@users.noreply.github.com> Date: Tue, 21 Jul 2026 06:04:01 -0700 Subject: [PATCH] Fix: isolate Remote L3 daemon per-connection send failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The shared Remote L3 control daemon's accept loop treated a send failure on one parent connection as fatal to the whole daemon. When a timed-out parent closed its socket before the daemon replied, the success-reply send raised BrokenPipe, the except handler issued a second bare send on the same dead connection which also raised, and that exception escaped the accept loop into `finally: server.close()`. The daemon is the control plane for *starting* sessions; already-ready sessions talk to their parent over their own direct command/health sockets, so its death does not tear those down. What it does kill is the current, still-undelivered runner and every subsequent session-start request to that host — a shared-fate failure of the host's control plane. - `_start_session` now returns `(reply, proc)`: the live ready runner is handed back to the caller instead of being reaped internally, so the caller can reclaim it if the reply is never delivered. A failed handshake is still killed + reaped in place and returns `proc=None`. A successful send only means the bytes were queued locally, not that the parent read them (unobserved receipt would need an ACK / lease). - New `_serve_connection` owns one connection with full isolation and a closed ownership transaction: any `Exception` affects only that session; `KeyboardInterrupt`/`SystemExit` (BaseException) propagate for clean shutdown instead of becoming a spurious error reply; and a live runner is disposed of exactly once on every exit path — handed to the reaper only once its send returns, otherwise reclaimed by a `finally`. This covers the send failing, the reaper thread failing to launch, and a control exception unwinding mid-connection. - `serve` delegates to a `_serve_loop` that stops when the listener closes, so the loop can be driven and torn down deterministically. Tests reproduce each gap first (all fail against the pre-fix source and against interim versions lacking the finally/Exception narrowing): deterministic `_serve_connection` unit tests for hand-off, reclaim on send failure, reclaim on reaper-launch failure, error-reply isolation, truncated-frame/EOF, real-error reporting, and control-exception propagation (with exactly-once reclaim); a `_serve_loop` test for multi-connection survival and clean stop; and two bounded real-socket tests (dead parent, truncated frame) that serve the next connection and join their server thread with no listener leak or port race. --- python/simpler/remote_l3_worker.py | 71 +++- tests/ut/py/test_remote_l3_lifecycle.py | 410 +++++++++++++++++++++++- 2 files changed, 456 insertions(+), 25 deletions(-) diff --git a/python/simpler/remote_l3_worker.py b/python/simpler/remote_l3_worker.py index fc0acf0dea..a53834d9ed 100644 --- a/python/simpler/remote_l3_worker.py +++ b/python/simpler/remote_l3_worker.py @@ -141,7 +141,14 @@ def _reap_session_runner(proc: subprocess.Popen[Any]) -> None: pass -def _start_session(manifest: dict[str, Any]) -> dict[str, Any]: +def _start_session(manifest: dict[str, Any]) -> tuple[dict[str, Any], subprocess.Popen[Any] | None]: + # Returns (reply, proc). proc is the live, ready runner handle when the reply + # is ok — the caller owns it: hand it to a background reaper once the reply + # send to the parent succeeds, or reclaim it if the send raises. proc is None + # when no runner survives (a failed handshake has already been killed and + # reaped here), so a failed send then leaves nothing to reclaim. A successful + # send only means the bytes were queued locally, not that the parent read + # them; unobserved receipt would need an ACK / lease, which sim does not have. _validate_manifest(manifest) # Both numeric timeouts are validated before any spawn resource (ready pipe, # manifest tempfile, runner Popen) exists: the runner is never launched only @@ -185,9 +192,8 @@ def _start_session(manifest: dict[str, Any]) -> dict[str, Any]: ready["pid"] = int(proc.pid) if not ready.get("ok", False): _wait_or_kill_runner(proc) - else: - threading.Thread(target=_reap_session_runner, args=(proc,), daemon=True).start() - return ready + return ready, None + return ready, proc finally: if ready_w >= 0: try: @@ -205,22 +211,63 @@ def _start_session(manifest: dict[str, Any]) -> dict[str, Any]: pass +def _serve_connection(conn: socket.socket) -> None: + # One parent connection, fully isolated: any Exception on this socket affects + # only this session, never the shared accept loop. KeyboardInterrupt / + # SystemExit are BaseException, not Exception, so they propagate out for a + # clean daemon shutdown instead of turning into a spurious error reply. A live + # runner is disposed of exactly once on every exit path — handed to the reaper + # only once its reply send returns, otherwise reclaimed by the finally — so no + # session runner is orphaned (until its own timeout) or double-reaped, even + # when the send fails, the reaper cannot be launched, or a control exception + # unwinds mid-connection. + proc: subprocess.Popen[Any] | None = None + try: + try: + manifest = _read_json(conn) + reply, proc = _start_session(manifest) + except Exception as exc: # noqa: BLE001 + # Handshake failed (bad manifest, runner start error); _start_session + # already reaped any runner it started. Best-effort error reply, + # swallowing only a send that itself fails on a dead parent. + with contextlib.suppress(OSError): + _send_json(conn, {"ok": False, "error": f"{type(exc).__name__}: {exc}"}) + return + try: + _send_json(conn, reply) + if proc is not None: + threading.Thread(target=_reap_session_runner, args=(proc,), daemon=True).start() + proc = None + except Exception: # noqa: BLE001 + # Local send raised (parent gone) or the reaper thread could not be + # started; the finally reclaims the still-owned runner. + return + finally: + if proc is not None: + _wait_or_kill_runner(proc) + + +def _serve_loop(server: socket.socket) -> None: + while True: + try: + conn, _addr = server.accept() + except OSError: + # Listener closed (shutdown); stop accepting. + return + with conn: + _serve_connection(conn) + + def serve(host: str, port: int) -> int: server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((host, port)) server.listen() try: - while True: - conn, _addr = server.accept() - with conn: - try: - manifest = _read_json(conn) - _send_json(conn, _start_session(manifest)) - except BaseException as exc: # noqa: BLE001 - _send_json(conn, {"ok": False, "error": f"{type(exc).__name__}: {exc}"}) + _serve_loop(server) finally: server.close() + return 0 def main(argv: list[str] | None = None) -> int: diff --git a/tests/ut/py/test_remote_l3_lifecycle.py b/tests/ut/py/test_remote_l3_lifecycle.py index 1c272f7a55..b7d694fd1b 100644 --- a/tests/ut/py/test_remote_l3_lifecycle.py +++ b/tests/ut/py/test_remote_l3_lifecycle.py @@ -7,8 +7,11 @@ # See LICENSE in the root of the software repository for the full text of the License. # ----------------------------------------------------------------------------------------------------------- +import contextlib +import json import os import socket +import struct import threading import time from typing import cast @@ -89,7 +92,7 @@ def fake_read_ready(fd, timeout_s): assert fake_proc.wait_calls >= 1 -def test_start_session_reaps_successful_runner(monkeypatch): +def test_start_session_returns_live_runner_without_reaping(monkeypatch): class FakePopen: pid = 12345 @@ -100,26 +103,42 @@ def wait(self, timeout=None): self.wait_calls += 1 return 0 - class FakeThread: - def __init__(self, *, target, args, daemon): - self.target = target - self.args = args - self.daemon = daemon - - def start(self): - self.target(*self.args) - fake_proc = FakePopen() monkeypatch.setattr(remote_l3_worker.subprocess, "Popen", lambda *args, **kwargs: fake_proc) monkeypatch.setattr(remote_l3_worker, "_read_runner_ready", lambda fd, timeout_s: {"ok": True}) - monkeypatch.setattr(remote_l3_worker.threading, "Thread", FakeThread) - reply = remote_l3_worker._start_session(_manifest()) + reply, proc = remote_l3_worker._start_session(_manifest()) + # A ready runner is handed back live for the caller to hand off or reclaim; + # _start_session itself neither reaps nor kills it. assert reply["ok"] is True assert reply["pid"] == fake_proc.pid - assert fake_proc.wait_calls == 1 + assert proc is fake_proc + assert fake_proc.wait_calls == 0 + + +def test_start_session_returns_none_proc_when_runner_reports_not_ok(monkeypatch): + class FakePopen: + pid = 777 + + def wait(self, timeout=None): + return 0 + + fake_proc = FakePopen() + reclaimed: list = [] + + monkeypatch.setattr(remote_l3_worker.subprocess, "Popen", lambda *args, **kwargs: fake_proc) + monkeypatch.setattr(remote_l3_worker, "_read_runner_ready", lambda fd, timeout_s: {"ok": False}) + monkeypatch.setattr(remote_l3_worker, "_wait_or_kill_runner", lambda p, **kw: reclaimed.append(p)) + + reply, proc = remote_l3_worker._start_session(_manifest()) + + # A failed handshake is killed+reaped exactly once inside _start_session, so + # the caller gets no runner to reclaim. + assert reply["ok"] is False + assert proc is None + assert reclaimed == [fake_proc] def test_run_session_bounds_post_ready_command_accept(monkeypatch): @@ -270,3 +289,368 @@ def close(self): assert sock.closed assert conn.closed + + +class _FakeProc: + """A runner that never exits on its own — models a ready, live session.""" + + pid = 4242 + + def __init__(self): + self.terminated = False + self.killed = False + + def wait(self, timeout=None): + raise remote_l3_worker.subprocess.TimeoutExpired(cmd="runner", timeout=timeout if timeout is not None else 0.0) + + def terminate(self): + self.terminated = True + + def kill(self): + self.killed = True + + +class _SyncThread: + """Runs the reaper target inline so hand-off is deterministically observable + (no scheduler race): construction+start executes the target immediately.""" + + def __init__(self, *, target, args, daemon): + self._target = target + self._args = args + + def start(self): + self._target(*self._args) + + +def _start_must_not_run(manifest): + raise AssertionError("_start_session must not run when the handshake read fails") + + +def _serve(monkeypatch, *, read=None, start=None, send=None, thread: type = _SyncThread): + """Install the common _serve_connection collaborators and return the spy + lists (sent, reaped, reclaimed).""" + sent: list = [] + reaped: list = [] + reclaimed: list = [] + if read is not None: + monkeypatch.setattr(remote_l3_worker, "_read_json", read) + if start is not None: + monkeypatch.setattr(remote_l3_worker, "_start_session", start) + if send is None: + send = lambda conn, payload: sent.append(payload) # noqa: E731 + monkeypatch.setattr(remote_l3_worker, "_send_json", send) + monkeypatch.setattr(remote_l3_worker, "_reap_session_runner", lambda p: reaped.append(p)) + monkeypatch.setattr(remote_l3_worker, "_wait_or_kill_runner", lambda p, **kw: reclaimed.append(p)) + monkeypatch.setattr(remote_l3_worker.threading, "Thread", thread) + return sent, reaped, reclaimed + + +def test_serve_connection_hands_off_runner_after_send_returns(monkeypatch): + proc = _FakeProc() + sent, reaped, reclaimed = _serve( + monkeypatch, + read=lambda conn: {"manifest": True}, + start=lambda manifest: ({"ok": True, "pid": proc.pid}, proc), + ) + + remote_l3_worker._serve_connection(cast(socket.socket, object())) + + assert sent == [{"ok": True, "pid": proc.pid}] + assert reaped == [proc] # send returned → runner handed to the reaper exactly once + assert reclaimed == [] # ownership transferred, so the finally does not reclaim + + +def test_serve_connection_reclaims_runner_when_send_raises(monkeypatch): + proc = _FakeProc() + + def broken_send(conn, payload): + raise BrokenPipeError("parent disconnected before reply") + + _sent, reaped, reclaimed = _serve( + monkeypatch, + read=lambda conn: {"manifest": True}, + start=lambda manifest: ({"ok": True, "pid": proc.pid}, proc), + send=broken_send, + ) + + # A dead parent on the reply must not escape this connection. + remote_l3_worker._serve_connection(cast(socket.socket, object())) + + assert reclaimed == [proc] # undelivered runner reclaimed exactly once + assert reaped == [] # never handed off + + +def test_serve_connection_reclaims_runner_when_reaper_launch_fails(monkeypatch): + proc = _FakeProc() + + class _FailingThread: + def __init__(self, *, target, args, daemon): + pass + + def start(self): + raise RuntimeError("can't start new thread") + + sent, reaped, reclaimed = _serve( + monkeypatch, + read=lambda conn: {"manifest": True}, + start=lambda manifest: ({"ok": True, "pid": proc.pid}, proc), + thread=_FailingThread, + ) + + # Thread exhaustion at reaper launch must neither escape the connection nor + # orphan the runner. + remote_l3_worker._serve_connection(cast(socket.socket, object())) + + assert sent == [{"ok": True, "pid": proc.pid}] # reply was delivered + assert reaped == [] + assert reclaimed == [proc] # runner reclaimed exactly once + + +def test_serve_connection_swallows_error_reply_send_failure(monkeypatch): + def bad_start(manifest): + raise ValueError("bad manifest") + + def broken_send(conn, payload): + raise ConnectionResetError("parent disconnected before error reply") + + _sent, _reaped, reclaimed = _serve( + monkeypatch, + read=lambda conn: {"manifest": True}, + start=bad_start, + send=broken_send, + ) + + # No runner was ever created and the error reply cannot land — nothing to + # reclaim, nothing escapes. + remote_l3_worker._serve_connection(cast(socket.socket, object())) + + assert reclaimed == [] + + +def test_serve_connection_survives_truncated_frame(monkeypatch): + def eof(conn): + raise EOFError("remote daemon socket closed") + + _sent, _reaped, reclaimed = _serve( + monkeypatch, + read=eof, + start=_start_must_not_run, + ) + + # A truncated / closed frame is an ordinary handshake failure: no runner, no + # escape. + remote_l3_worker._serve_connection(cast(socket.socket, object())) + + assert reclaimed == [] + + +def test_serve_connection_reports_real_error_to_live_parent(monkeypatch): + def bad_start(manifest): + raise ValueError("bad manifest") + + sent, _reaped, _reclaimed = _serve( + monkeypatch, + read=lambda conn: {"manifest": True}, + start=bad_start, + ) + + remote_l3_worker._serve_connection(cast(socket.socket, object())) + + assert len(sent) == 1 + assert sent[0]["ok"] is False + assert "ValueError" in sent[0]["error"] + + +def test_serve_connection_propagates_control_exception_from_handshake(monkeypatch): + def interrupt(conn): + raise KeyboardInterrupt + + sent, _reaped, reclaimed = _serve( + monkeypatch, + read=interrupt, + start=_start_must_not_run, + ) + + # KeyboardInterrupt is BaseException: it propagates for clean shutdown rather + # than being caught into a spurious error reply. + with pytest.raises(KeyboardInterrupt): + remote_l3_worker._serve_connection(cast(socket.socket, object())) + + assert sent == [] + assert reclaimed == [] + + +def test_serve_connection_reclaims_runner_and_propagates_control_exception_on_send(monkeypatch): + proc = _FakeProc() + + def interrupt_send(conn, payload): + raise KeyboardInterrupt + + _sent, reaped, reclaimed = _serve( + monkeypatch, + read=lambda conn: {"manifest": True}, + start=lambda manifest: ({"ok": True, "pid": proc.pid}, proc), + send=interrupt_send, + ) + + # The control exception unwinds, but the live runner is reclaimed exactly + # once before it propagates. + with pytest.raises(KeyboardInterrupt): + remote_l3_worker._serve_connection(cast(socket.socket, object())) + + assert reaped == [] + assert reclaimed == [proc] + + +class _FakeConn: + def __init__(self): + self.closed = False + + def __enter__(self): + return self + + def __exit__(self, *exc): + self.closed = True + return False + + +class _FakeServer: + def __init__(self, conns): + self._conns = list(conns) + + def accept(self): + if self._conns: + return self._conns.pop(0), ("127.0.0.1", 0) + raise OSError("listener closed") + + +def test_serve_loop_serves_every_connection_and_stops_on_listener_close(monkeypatch): + served: list = [] + monkeypatch.setattr(remote_l3_worker, "_serve_connection", lambda conn: served.append(conn)) + + conns = [_FakeConn(), _FakeConn(), _FakeConn()] + remote_l3_worker._serve_loop(cast(socket.socket, _FakeServer(conns))) + + # Every connection was handled (loop survived each) and the loop exited + # cleanly when accept() reported the listener closed; each conn was closed. + assert served == conns + assert all(c.closed for c in conns) + + +def _serve_bounded(listener, n): + for _ in range(n): + conn, _addr = listener.accept() + with conn: + remote_l3_worker._serve_connection(conn) + + +def _bounded_daemon(listener, n): + thread = threading.Thread(target=_serve_bounded, args=(listener, n), daemon=True) + thread.start() + return thread + + +def _new_listener(): + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(("127.0.0.1", 0)) + listener.listen() + return listener, listener.getsockname()[1] + + +def _send_manifest(sock): + data = json.dumps({"any": "manifest"}).encode("utf-8") + sock.sendall(struct.pack("