diff --git a/news/7328.bugfix.md b/news/7328.bugfix.md new file mode 100644 index 00000000000..8c5e419ee8c --- /dev/null +++ b/news/7328.bugfix.md @@ -0,0 +1 @@ +`reflex run` now stops its frontend on SIGTERM and SIGINT without a TTY, while keeping frontend workers in the CLI process group so a hard kill also stops them. diff --git a/reflex/reflex.py b/reflex/reflex.py index e060971d046..217aef27f7a 100644 --- a/reflex/reflex.py +++ b/reflex/reflex.py @@ -3,11 +3,14 @@ from __future__ import annotations import logging +import signal import sys from collections.abc import Callable +from contextlib import contextmanager from importlib import import_module from importlib.util import find_spec from pathlib import Path +from types import FrameType from typing import TYPE_CHECKING, Any, NoReturn, cast import click @@ -397,6 +400,23 @@ def _compile_app_worker( otel.flush() +@contextmanager +def _frontend_sigterm_handler(enabled: bool): + """Exit cleanly on SIGTERM for a frontend-only run.""" + if not enabled: + yield + return + + def stop_frontend(signum: int, frame: FrameType | None) -> None: + raise SystemExit(0) + + previous = signal.signal(signal.SIGTERM, stop_frontend) + try: + yield + finally: + signal.signal(signal.SIGTERM, previous) + + def _run_dev( running_mode: constants.RunningMode, frontend_port: int | None, @@ -437,21 +457,45 @@ def _run_dev( running_mode.has_backend(), )) - # Start the frontend and backend. - with processes.run_concurrently_context(*commands): - # In dev mode, run the backend on the main thread. - if running_mode.has_backend() and backend_port: - exec.run_backend( - backend_host, - int(backend_port), - config.loglevel.subprocess_level(), - running_mode.has_frontend(), - ) - # The windows uvicorn bug workaround - # https://github.com/reflex-dev/reflex/issues/2335 - if constants.IS_WINDOWS and exec.frontend_process: - # Sends SIGTERM in windows - exec.kill(exec.frontend_process.pid) + with exec._frontend_process_lock: + exec._frontend_shutting_down = False + exec.frontend_process = None + + frontend_only = running_mode.has_frontend() and not running_mode.has_backend() + + def stop_frontend() -> None: + with exec._frontend_process_lock: + exec._frontend_shutting_down = True + if (process := exec.frontend_process) is not None: + if process.poll() is None: + process.terminate() + exec.frontend_process = None + + try: + with ( + _frontend_sigterm_handler(frontend_only and sys.platform != "win32"), + processes.run_concurrently_context( + *commands, interrupt_on_failure=not frontend_only + ) as tasks, + ): + try: + if frontend_only and tasks: + tasks[0].result() + elif running_mode.has_backend() and backend_port: + exec.run_backend( + backend_host, + int(backend_port), + config.loglevel.subprocess_level(), + running_mode.has_frontend(), + ) + # The windows uvicorn bug workaround + # https://github.com/reflex-dev/reflex/issues/2335 + if constants.IS_WINDOWS and exec.frontend_process: + exec.kill(exec.frontend_process.pid) + finally: + stop_frontend() + finally: + stop_frontend() def _run_preview(running_mode: constants.RunningMode, port: int, host: str): diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index 1d50fcff424..5843b03ac04 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -35,6 +35,8 @@ # For uvicorn windows bug fix (#2335) frontend_process = None +_frontend_process_lock = threading.Lock() +_frontend_shutting_down = False DEV_BACKEND_RELOAD_MARKER = ".reflex_dev_backend_started" @@ -296,7 +298,10 @@ def run_process_and_launch_url( **kwargs, ) global frontend_process - frontend_process = process + with _frontend_process_lock: + frontend_process = process + if _frontend_shutting_down and process.poll() is None: + process.terminate() if process.stdout: for line in processes.stream_logs("Starting frontend", process): new_content, new_hash = get_package_json_and_hash(json_file_path) diff --git a/reflex/utils/processes.py b/reflex/utils/processes.py index e41303a2fad..934bc37b774 100644 --- a/reflex/utils/processes.py +++ b/reflex/utils/processes.py @@ -283,6 +283,7 @@ def _interrupt_main_thread(): @contextlib.contextmanager def run_concurrently_context( *fns: Callable[..., Any] | tuple[Callable[..., Any], ...], + interrupt_on_failure: bool = True, ) -> Generator[list[futures.Future], None, None]: """Run functions concurrently in a thread pool. @@ -293,6 +294,7 @@ def run_concurrently_context( Args: *fns: The functions to run. + interrupt_on_failure: Whether to wake the main thread if a task fails. Yields: The futures for the functions. @@ -346,9 +348,10 @@ def raise_first_failure(tasks: list[futures.Future]): try: executor = futures.ThreadPoolExecutor(max_workers=len(fns)) # Submit the tasks. - tasks = [executor.submit(*fn) for fn in fns] - for task in tasks: - task.add_done_callback(wake_main_thread) + tasks = [executor.submit(fn[0], *fn[1:]) for fn in fns] + if interrupt_on_failure: + for task in tasks: + task.add_done_callback(wake_main_thread) try: try: diff --git a/tests/units/test_reflex.py b/tests/units/test_reflex.py index f2f5215c2d1..b1d850a9700 100644 --- a/tests/units/test_reflex.py +++ b/tests/units/test_reflex.py @@ -2,17 +2,22 @@ from __future__ import annotations +import contextlib import json import os +import signal import subprocess import sys +import time import click import click.testing +import psutil import pytest from pytest_mock import MockerFixture from reflex import reflex +from reflex.testing import DEFAULT_TIMEOUT _CLI_STARTUP_DENIED_MODULES = frozenset({ "PIL", @@ -434,6 +439,427 @@ def test_init_records_version_check_after_frontend_setup( assert events == ["frontend", "version"] +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +def test_frontend_preflight_failure_cleans_process(tmp_path): + ready = tmp_path / "frontend.pid" + driver = tmp_path / "driver.py" + driver.write_text( + """import subprocess, sys, types +from reflex.utils import build, exec as exec_mod, telemetry + +def frontend(*args): + p = subprocess.Popen([sys.executable, "-c", "import time;time.sleep(60)"]) + exec_mod.frontend_process = p + with open(PIDS, "w") as handshake: + handshake.write(str(p.pid)) + sys.stderr.write("x" * 131072) + raise SystemExit(3) + +exec_mod.run_frontend = frontend +exec_mod.run_backend = lambda *args: None +telemetry.send = lambda *args, **kwargs: None +build.setup_frontend = lambda *args, **kwargs: None +import reflex.reflex as rx +rx._compile_app = lambda: None +rx.get_config = lambda: types.SimpleNamespace( + _set_persistent=lambda **kwargs: None, + loglevel=types.SimpleNamespace(subprocess_level=lambda: None), +) +from reflex_base import constants +rx._run_dev(constants.RunningMode.FULLSTACK, 3000, 8000, "127.0.0.1") +""".replace("PIDS", repr(str(ready))) + ) + launcher = subprocess.Popen( + [sys.executable, str(driver)], + cwd=tmp_path, + start_new_session=True, + stderr=subprocess.PIPE, + ) + child = None + try: + _, stderr = launcher.communicate(timeout=DEFAULT_TIMEOUT) + assert ready.exists(), ( + "frontend did not start: " + stderr.decode(errors="replace")[-1000:] + ) + child = int(ready.read_text()) + assert launcher.returncode == 3, stderr.decode(errors="replace")[-1000:] + with contextlib.suppress(psutil.NoSuchProcess): + assert psutil.Process(child).status() in ( + psutil.STATUS_ZOMBIE, + psutil.STATUS_DEAD, + ) + finally: + if launcher.poll() is None: + os.killpg(launcher.pid, signal.SIGKILL) + launcher.communicate(timeout=DEFAULT_TIMEOUT) + if child is not None: + with contextlib.suppress(ProcessLookupError): + os.kill(child, signal.SIGKILL) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +def test_late_frontend_start_stops_after_backend_returns(tmp_path): + """A frontend started after backend teardown must exit without hanging.""" + pids = tmp_path / "frontend.pid" + driver = tmp_path / "driver.py" + driver.write_text( + """import subprocess, sys, time, types +from pathlib import Path +from reflex.utils import build, exec as exec_mod, processes, telemetry + +original_new_process = processes.new_process +def record_process(*args, **kwargs): + deadline = time.monotonic() + 5 + while not exec_mod._frontend_shutting_down and time.monotonic() < deadline: + time.sleep(0.01) + assert exec_mod._frontend_shutting_down + p = original_new_process(*args, **kwargs) + with open(PIDS, "w") as handshake: + handshake.write(str(p.pid)) + return p +processes.new_process = record_process +exec_mod.get_web_dir = lambda: Path.cwd() +exec_mod.get_package_json_and_hash = lambda *args: ({}, "unchanged") +exec_mod.frontend_env = lambda *args: {} +exec_mod.path_ops.get_node_bin_path = lambda: None + +def frontend(*args): + exec_mod.run_process_and_launch_url( + [sys.executable, "-c", "import time;time.sleep(60)"], True + ) + +def backend(*args): + return None + +exec_mod.run_frontend = frontend +exec_mod.run_backend = backend +telemetry.send = lambda *args, **kwargs: None +build.setup_frontend = lambda *args, **kwargs: None +import reflex.reflex as rx +rx._compile_app = lambda: None +rx.get_config = lambda: types.SimpleNamespace( + _set_persistent=lambda **kwargs: None, + loglevel=types.SimpleNamespace(subprocess_level=lambda: None), +) +from reflex_base import constants +rx._run_dev(constants.RunningMode.FULLSTACK, 3000, 8000, "127.0.0.1") +""".replace("PIDS", repr(str(pids))) + ) + launcher = subprocess.Popen( + [sys.executable, str(driver)], + cwd=tmp_path, + start_new_session=True, + stderr=subprocess.PIPE, + ) + child = None + try: + try: + launcher.wait(timeout=3) + except subprocess.TimeoutExpired: + pytest.fail("late frontend trapped the run after backend returned") + assert launcher.returncode == 0 + assert pids.exists(), "late frontend never launched" + child = int(pids.read_text().strip()) + with contextlib.suppress(psutil.NoSuchProcess): + assert psutil.Process(child).status() in ( + psutil.STATUS_ZOMBIE, + psutil.STATUS_DEAD, + ) + finally: + if launcher.poll() is None: + os.killpg(launcher.pid, signal.SIGKILL) + launcher.wait(timeout=DEFAULT_TIMEOUT) + if child is not None: + with contextlib.suppress(ProcessLookupError): + os.kill(child, signal.SIGKILL) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +def test_fullstack_backend_exit_stops_frontend(tmp_path): + """Frontend must not trap the run after a normal backend return.""" + pids = tmp_path / "frontend.pid" + driver = tmp_path / "driver.py" + driver.write_text( + """import subprocess, sys, time, types +from pathlib import Path +from reflex.utils import build, exec as exec_mod, processes, telemetry + +exec_mod.get_web_dir = lambda: Path.cwd() +exec_mod.get_package_json_and_hash = lambda *args: ({}, "unchanged") +exec_mod.frontend_env = lambda *args: {} +exec_mod.path_ops.get_node_bin_path = lambda: None +original_new_process = processes.new_process +def record_process(*args, **kwargs): + p = original_new_process(*args, **kwargs) + with open(PIDS, "w") as handshake: + handshake.write(str(p.pid)) + return p +processes.new_process = record_process +exec_mod.run_frontend = lambda *args: exec_mod.run_process_and_launch_url( + [sys.executable, "-c", "import time;time.sleep(60)"], True +) +def backend(*args): + from pathlib import Path + deadline = time.monotonic() + 5 + while not Path(PIDS).exists() and time.monotonic() < deadline: + time.sleep(0.01) +exec_mod.run_backend = backend +telemetry.send = lambda *args, **kwargs: None +build.setup_frontend = lambda *args, **kwargs: None +import reflex.reflex as rx +rx._compile_app = lambda: None +rx.get_config = lambda: types.SimpleNamespace( + _set_persistent=lambda **kwargs: None, + loglevel=types.SimpleNamespace(subprocess_level=lambda: None), +) +from reflex_base import constants +rx._run_dev(constants.RunningMode.FULLSTACK, 3000, 8000, "127.0.0.1") +""".replace("PIDS", repr(str(pids))) + ) + launcher = subprocess.Popen( + [sys.executable, str(driver)], + cwd=tmp_path, + start_new_session=True, + stderr=subprocess.PIPE, + ) + child = None + try: + deadline = time.monotonic() + DEFAULT_TIMEOUT + while time.monotonic() < deadline: + if launcher.poll() is not None: + pytest.fail(f"launcher exited early: {launcher.returncode}") + if pids.exists() and (pid := pids.read_text().strip()): + child = int(pid) + break + time.sleep(0.01) + assert child is not None + try: + launcher.wait(timeout=2) + except subprocess.TimeoutExpired: + pytest.fail("full-stack run hung after backend returned") + assert launcher.returncode == 0 + with contextlib.suppress(psutil.NoSuchProcess): + assert psutil.Process(child).status() in ( + psutil.STATUS_ZOMBIE, + psutil.STATUS_DEAD, + ) + finally: + if launcher.poll() is None: + os.killpg(launcher.pid, signal.SIGKILL) + launcher.wait(timeout=DEFAULT_TIMEOUT) + if child is not None: + with contextlib.suppress(ProcessLookupError): + os.kill(child, signal.SIGKILL) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups") +@pytest.mark.parametrize("mode", ["frontend", "fullstack"]) +def test_sigkill_run_group_stops_frontend(tmp_path, mode): + """A hard kill of the CLI job group must not leave frontend workers alive.""" + pids = tmp_path / "children" + driver = tmp_path / "driver.py" + driver.write_text( + """import subprocess, sys, time, types +from pathlib import Path +from reflex.utils import build, exec as exec_mod, processes, telemetry + +code = "import subprocess,sys,time\\n" + \ + "p=subprocess.Popen([sys.executable,'-c','import time;time.sleep(60)'])\\n" + \ + "print(p.pid,flush=True);time.sleep(60)\\n" +exec_mod.get_web_dir=lambda:Path.cwd() +exec_mod.get_package_json_and_hash=lambda *args:({}, "unchanged") +exec_mod.frontend_env=lambda *args:{} +exec_mod.path_ops.get_node_bin_path=lambda:None +original_new_process=processes.new_process +def record_process(*args,**kwargs): + p=original_new_process(*args,**kwargs) + with open(PIDS,"w") as handshake: + handshake.write(str(p.pid)) + return p +processes.new_process=record_process +exec_mod.run_frontend=lambda *args:exec_mod.run_process_and_launch_url( + [sys.executable,"-c",code],True +) +exec_mod.run_backend=lambda *args: time.sleep(60) +telemetry.send=lambda *a,**k:None +build.setup_frontend=lambda *a,**k:None +import reflex.reflex as rx +rx._compile_app=lambda:None +rx.get_config=lambda:types.SimpleNamespace(_set_persistent=lambda **k:None, + loglevel=types.SimpleNamespace(subprocess_level=lambda:None)) +from reflex_base import constants +rx._run_dev(MODE,3000,PORT,"127.0.0.1") +""" + .replace("PIDS", repr(str(pids))) + .replace( + "MODE", + "constants.RunningMode.FRONTEND_ONLY" + if mode == "frontend" + else "constants.RunningMode.FULLSTACK", + ) + .replace("PORT", "None" if mode == "frontend" else "8000") + ) + launcher = subprocess.Popen( + [sys.executable, str(driver)], + cwd=tmp_path, + start_new_session=True, + stderr=subprocess.PIPE, + text=True, + ) + child = grandchild = None + try: + deadline = time.monotonic() + DEFAULT_TIMEOUT + while time.monotonic() < deadline: + if launcher.poll() is not None: + pytest.fail(f"launcher exited early: {launcher.returncode}") + if pids.exists(): + parts = pids.read_text().split() + if len(parts) == 1: + child = int(parts[0]) + children = psutil.Process(child).children() + if children: + grandchild = children[0].pid + break + time.sleep(0.01) + assert child is not None + assert grandchild is not None + assert os.getpgid(child) == launcher.pid + os.killpg(launcher.pid, signal.SIGKILL) + launcher.wait(timeout=DEFAULT_TIMEOUT) + for pid in (child, grandchild): + deadline = time.monotonic() + DEFAULT_TIMEOUT + while time.monotonic() < deadline: + try: + if psutil.Process(pid).status() in ( + psutil.STATUS_ZOMBIE, + psutil.STATUS_DEAD, + ): + break + except psutil.NoSuchProcess: + break + time.sleep(0.05) + else: + pytest.fail(f"frontend descendant {pid} survived SIGKILL") + finally: + if launcher.poll() is None: + os.killpg(launcher.pid, signal.SIGKILL) + launcher.wait(timeout=DEFAULT_TIMEOUT) + if child is not None: + with contextlib.suppress(ProcessLookupError): + os.killpg(child, signal.SIGKILL) + for pid in (child, grandchild): + if pid is not None: + with contextlib.suppress(ProcessLookupError): + os.kill(pid, signal.SIGKILL) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +@pytest.mark.parametrize("mode", ["frontend", "fullstack"]) +@pytest.mark.parametrize("sig", ["SIGTERM", "SIGINT"]) +def test_no_tty_run_stops_frontend(tmp_path, mode, sig): + """Headless runs stop the frontend on SIGTERM or SIGINT.""" + pids = tmp_path / "children" + driver = tmp_path / "driver.py" + driver.write_text( + """import signal, subprocess, sys, time, types +from reflex.utils import build, exec as exec_mod, telemetry + +def frontend(*args): + code = "import signal,subprocess,sys,time\\n" + \ + "p=subprocess.Popen([sys.executable,'-c','import time;time.sleep(60)'])\\n" + \ + "signal.signal(signal.SIGTERM, lambda *a: (p.terminate(), sys.exit(0)))\\n" + \ + "print(p.pid,flush=True);time.sleep(60)\\n" + p = subprocess.Popen([sys.executable,"-c",code],stdout=subprocess.PIPE,text=True) + grandchild = int(p.stdout.readline()) + exec_mod.frontend_process = p + with open(PIDS,"w") as handshake: + handshake.write(f"{p.pid} {grandchild}") + p.wait() + +def backend(*args): + def stop(sig,frame): raise SystemExit(0) + signal.signal(signal.SIGTERM,stop) + signal.signal(signal.SIGINT,stop) + while True: time.sleep(1) + +exec_mod.run_frontend=frontend +exec_mod.run_backend=backend +telemetry.send=lambda *a,**k:None +build.setup_frontend=lambda *a,**k:None +import reflex.reflex as rx +rx._compile_app=lambda:None +rx.get_config=lambda:types.SimpleNamespace(_set_persistent=lambda **k:None, + loglevel=types.SimpleNamespace(subprocess_level=lambda:None)) +from reflex_base import constants +rx._run_dev(MODE,3000,PORT,"127.0.0.1") +""" + .replace("PIDS", repr(str(pids))) + .replace( + "MODE", + "constants.RunningMode.FRONTEND_ONLY" + if mode == "frontend" + else "constants.RunningMode.FULLSTACK", + ) + .replace("PORT", "None" if mode == "frontend" else "8000") + ) + launcher = subprocess.Popen( + [sys.executable, str(driver)], + cwd=tmp_path, + start_new_session=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + child = grandchild = None + try: + deadline = time.monotonic() + DEFAULT_TIMEOUT + while time.monotonic() < deadline: + if launcher.poll() is not None: + pytest.fail(f"launcher exited early: {launcher.returncode}") + if pids.exists(): + parts = pids.read_text().split() + if len(parts) == 2: + child, grandchild = map(int, parts) + break + time.sleep(0.01) + assert child is not None, "frontend did not start" + assert grandchild is not None, "frontend grandchild did not start" + os.kill(launcher.pid, getattr(signal, sig)) + try: + returncode = launcher.wait(timeout=DEFAULT_TIMEOUT) + except subprocess.TimeoutExpired: + pytest.fail("no-TTY launcher hung after signal") + if sig == "SIGINT": + assert returncode in (0, -signal.SIGINT) + else: + assert returncode == 0 + for pid in (child, grandchild): + deadline = time.monotonic() + DEFAULT_TIMEOUT + while time.monotonic() < deadline: + try: + if psutil.Process(pid).status() in ( + psutil.STATUS_ZOMBIE, + psutil.STATUS_DEAD, + ): + break + except psutil.NoSuchProcess: + break + time.sleep(0.05) + else: + pytest.fail(f"frontend descendant {pid} survived signal") + finally: + if launcher.poll() is None: + os.killpg(launcher.pid, signal.SIGKILL) + launcher.wait(timeout=DEFAULT_TIMEOUT) + if child is not None: + with contextlib.suppress(ProcessLookupError): + os.killpg(child, signal.SIGKILL) + for pid in (child, grandchild): + if pid is not None: + with contextlib.suppress(ProcessLookupError): + os.kill(pid, signal.SIGKILL) + + @pytest.mark.parametrize( ("argv", "supervised", "expected"), [(["--json"], False, True), (["--json"], True, False), ([], False, False)],