From 22a1ac65a29893352b6967f1bae3599d002371c2 Mon Sep 17 00:00:00 2001 From: Farhan Date: Fri, 11 Sep 2026 01:46:52 +0500 Subject: [PATCH 01/18] perf: preload the app and fork production granian workers On Linux the production backend now forces the "fork" start method, imports the app in the supervisor, and freezes the GC before granian spawns workers, so every worker shares the framework and app pages copy-on-write instead of re-importing ~1,800 modules privately. A forked child inherits the telemetry ThreadPoolExecutor without its worker thread, so events submitted in a worker would never be sent; an at-fork hook drops the inherited pool so the child creates its own. REFLEX_BACKEND_START_METHOD overrides the start method for apps that are not fork-safe. Blank app, GRANIAN_WORKERS=4, median of 3 runs (PSS over the process tree from /proc//smaps_rollup after serving requests): before: 721 MB PSS, 692 MB private dirty, port up in 1.82 s after: 223 MB PSS, 70 MB private dirty, port up in 1.07 s Claude-Session: https://claude.ai/code/session_01CEb3ocfFLeHkjkAKKH8YCy --- news/+prod-fork-preload.performance.md | 1 + .../news/+prod-fork-preload.performance.md | 1 + .../src/reflex_base/environment.py | 7 ++ reflex/utils/exec.py | 44 +++++++++ reflex/utils/telemetry.py | 10 ++ tests/units/test_telemetry.py | 11 +++ tests/units/utils/test_exec.py | 97 +++++++++++++++++++ 7 files changed, 171 insertions(+) create mode 100644 news/+prod-fork-preload.performance.md create mode 100644 packages/reflex-base/news/+prod-fork-preload.performance.md diff --git a/news/+prod-fork-preload.performance.md b/news/+prod-fork-preload.performance.md new file mode 100644 index 00000000000..28aa7e7d14f --- /dev/null +++ b/news/+prod-fork-preload.performance.md @@ -0,0 +1 @@ +Production backend workers on Linux are now forked from a supervisor that has already imported the app, so framework and app code are shared copy-on-write instead of re-imported per worker. A 4-worker blank app drops from about 720 MB to about 220 MB of proportional set size and starts faster. Set `REFLEX_BACKEND_START_METHOD=spawn` for apps that are not fork-safe. diff --git a/packages/reflex-base/news/+prod-fork-preload.performance.md b/packages/reflex-base/news/+prod-fork-preload.performance.md new file mode 100644 index 00000000000..53d78489de8 --- /dev/null +++ b/packages/reflex-base/news/+prod-fork-preload.performance.md @@ -0,0 +1 @@ +Add the `REFLEX_BACKEND_START_METHOD` environment variable to choose how production backend workers are started (`fork`, `spawn`, or `forkserver`). diff --git a/packages/reflex-base/src/reflex_base/environment.py b/packages/reflex-base/src/reflex_base/environment.py index 3bb80d6970b..aacb2741fc2 100644 --- a/packages/reflex-base/src/reflex_base/environment.py +++ b/packages/reflex-base/src/reflex_base/environment.py @@ -708,6 +708,13 @@ class EnvironmentVariables: # Whether to run Granian in a spawn process. This enables Reflex to pick up on environment variable changes between hot reloads. REFLEX_STRICT_HOT_RELOAD: EnvVar[bool] = env_var(False) + # The multiprocessing start method for production backend workers. Unset means "fork" wherever + # the interpreter itself defaults to a fork-based method (Linux), so workers share the app + # preloaded by the supervisor. Set to "spawn" for apps that are not fork-safe. + REFLEX_BACKEND_START_METHOD: EnvVar[ + Literal["fork", "spawn", "forkserver"] | None + ] = env_var(None) + # The path to the reflex log file. If not set, the log file will be stored in the reflex user directory. REFLEX_LOG_FILE: EnvVar[Path | None] = env_var(None) diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index c93b11949c3..16786d8c254 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -785,6 +785,43 @@ def run_uvicorn_backend_prod( ) +def _backend_start_method() -> str | None: + """Resolve the multiprocessing start method for production backend workers. + + Returns: + The start method to force, or None to keep the interpreter default. + """ + if (method := environment.REFLEX_BACKEND_START_METHOD.get()) is not None: + return method + import multiprocessing + + # Python defaults to fork (<3.14) or forkserver (3.14+) on Linux and to + # spawn elsewhere; only where fork is already the platform norm do we rely + # on it so workers can share the supervisor's pages. + if multiprocessing.get_start_method() in ("fork", "forkserver"): + return "fork" + return None + + +def _preload_for_fork(app_target: str | None) -> None: + """Import the app in the supervisor so forked workers share its pages. + + Args: + app_target: The ASGI app target; None means the reflex app, which is + imported here. Any other target lives in an already-loaded module. + """ + import gc + + from reflex.utils import prerequisites + + if app_target is None: + prerequisites.get_app() + # Freezing keeps worker GC passes from writing to the preloaded objects' + # headers, which would copy-on-write the shared pages private again. + gc.collect() + gc.freeze() + + def run_granian_backend_prod( host: str, port: int, loglevel: LogLevel, app_target: str | None = None ): @@ -796,12 +833,19 @@ def run_granian_backend_prod( loglevel: The log level. app_target: The ASGI app target to run. Defaults to the reflex app instance. """ + import multiprocessing + from granian.constants import Interfaces from granian.log import LogLevels from granian.server import Server as Granian logger.debug("Using Granian for backend") + if (start_method := _backend_start_method()) is not None: + multiprocessing.set_start_method(start_method, force=True) + if start_method == "fork": + _preload_for_fork(app_target) + granian_app = Granian( target=app_target or get_app_instance_from_file(), factory=True, diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index f2f8b794301..1d6cdd6db54 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -496,6 +496,16 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: return _executor +def _reset_executor_after_fork() -> None: + """Drop the inherited executor; its worker thread does not exist in the child.""" + global _executor + _executor = None + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_reset_executor_after_fork) + + def _current_registration_context() -> RegistrationContext | None: """Return the caller's RegistrationContext, or None if none is attached. diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 113d01c4d05..4f408ef390a 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -801,3 +801,14 @@ def test_flush_returns_false_when_worker_does_not_drain_in_time(): finally: release.set() blocker.result(timeout=5) + + +def test_executor_is_recreated_after_fork(): + """A forked child drops the inherited pool, whose thread it does not own.""" + inherited = telemetry._get_telemetry_executor() + + telemetry._reset_executor_after_fork() + + fresh = telemetry._get_telemetry_executor() + assert fresh is not inherited + assert fresh.submit(lambda: 1).result(timeout=5) == 1 diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index 5dfc677c094..ae85908c965 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -1,5 +1,7 @@ """Tests for development backend launchers in ``reflex.utils.exec``.""" +import gc +import multiprocessing import os from pathlib import Path @@ -8,6 +10,7 @@ from reflex_base.environment import environment from reflex.utils import exec as exec_utils +from reflex.utils import prerequisites DEV_BACKEND_RELOAD_ENV_NAME = environment.REFLEX_DEV_BACKEND_RELOAD_ACTIVE.name @@ -116,3 +119,97 @@ def test_arbitrate_ssr_env_var_wins(monkeypatch: pytest.MonkeyPatch): monkeypatch.setenv(environment.REFLEX_SSR.name, "False") assert exec_utils.arbitrate_ssr(True) is False + + +def _fake_granian_prod(mocker: MockerFixture, calls: list[str]): + """Patch granian and the prod launcher's collaborators, recording call order.""" + granian_server = pytest.importorskip("granian.server") + + class FakeGranian: + def __init__(self, *_args, **_kwargs): + pass + + def serve(self): + calls.append("serve") + + mocker.patch.object(granian_server, "Server", FakeGranian) + mocker.patch.object( + exec_utils, "get_app_instance_from_file", return_value="app:app" + ) + mocker.patch.object(exec_utils, "_get_backend_workers", return_value=1) + mocker.patch.object( + multiprocessing, + "set_start_method", + side_effect=lambda method, force=False: calls.append(f"start:{method}"), + ) + mocker.patch.object( + prerequisites, "get_app", side_effect=lambda: calls.append("preload") + ) + mocker.patch.object(gc, "freeze", side_effect=lambda: calls.append("freeze")) + + +def test_run_granian_backend_prod_preloads_app_before_forking( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """With fork, the app is imported and the heap frozen before workers start.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") + calls: list[str] = [] + _fake_granian_prod(mocker, calls) + + exec_utils.run_granian_backend_prod( + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + ) + + assert calls == ["start:fork", "preload", "freeze", "serve"] + + +def test_run_granian_backend_prod_spawn_skips_preload( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """Spawned workers re-import the app, so the supervisor does not load it.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "spawn") + calls: list[str] = [] + _fake_granian_prod(mocker, calls) + + exec_utils.run_granian_backend_prod( + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + ) + + assert calls == ["start:spawn", "serve"] + + +def test_run_granian_backend_prod_custom_target_only_freezes( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """A non-reflex target lives in an already-imported module.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") + calls: list[str] = [] + _fake_granian_prod(mocker, calls) + + exec_utils.run_granian_backend_prod( + host="0.0.0.0", + port=8000, + loglevel=exec_utils.LogLevel.INFO, + app_target="reflex.utils.exec:_frontend_prod_app", + ) + + assert calls == ["start:fork", "freeze", "serve"] + + +@pytest.mark.parametrize( + ("default_method", "expected"), + [("fork", "fork"), ("forkserver", "fork"), ("spawn", None)], +) +def test_backend_start_method_follows_platform_default( + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, + default_method: str, + expected: str | None, +): + """Fork is forced only where the interpreter already defaults to forking.""" + monkeypatch.delenv(environment.REFLEX_BACKEND_START_METHOD.name, raising=False) + mocker.patch.object( + multiprocessing, "get_start_method", return_value=default_method + ) + + assert exec_utils._backend_start_method() == expected From 579c420e39bacac65bff76ffe3cc006d84eed30a Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 30 Sep 2026 23:10:36 +0000 Subject: [PATCH 02/18] Defer run telemetry past the fork and drain the executor first Address the review asking not to fork granian workers while a telemetry thread is alive: - run_backend_prod, run_frontend_prod, run_granian_backend_prod and run_uvicorn_backend_prod take an on_started callback; `reflex run --env prod` sends the "run-prod" event from it instead of before the server starts. Granian calls it from Server.startup() once the workers are spawned; uvicorn/gunicorn runs in a separate interpreter, so it is called right before that process is launched. - Before forking, the supervisor finishes Plotly serializer setup (_prepare_serializers_for_fork) and drains and shuts down the telemetry executor, so no telemetry thread exists at the fork. - The at-fork child hook also recreates the executor creation lock, so a fork taken while another thread held it cannot block the child's first telemetry send forever. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/reflex.py | 15 ++++-- reflex/utils/exec.py | 67 +++++++++++++++++++---- reflex/utils/telemetry.py | 18 ++++++- tests/units/test_reflex.py | 35 ++++++++++++ tests/units/test_telemetry.py | 59 +++++++++++++++++++++ tests/units/utils/test_exec.py | 97 +++++++++++++++++++++++++++++++--- 6 files changed, 269 insertions(+), 22 deletions(-) diff --git a/reflex/reflex.py b/reflex/reflex.py index 00da577abbc..a874ef8df9d 100644 --- a/reflex/reflex.py +++ b/reflex/reflex.py @@ -515,8 +515,11 @@ def _run_prod(running_mode: constants.RunningMode, port: int, host: str): _skip_compile() - # Post a telemetry event. - telemetry.send("run-prod") + # Post the telemetry event once the workers are running: sending it here + # would start the telemetry thread before the server forks its workers. + def on_started(): + """Send the run telemetry from the supervisor.""" + telemetry.send("run-prod") # Display custom message when there is a keyboard interrupt. atexit.register(processes.atexit_handler) @@ -528,10 +531,14 @@ def _run_prod(running_mode: constants.RunningMode, port: int, host: str): ) if running_mode.has_backend(): exec.run_backend_prod( - host, port, config.loglevel.subprocess_level(), running_mode.has_frontend() + host, + port, + config.loglevel.subprocess_level(), + running_mode.has_frontend(), + on_started=on_started, ) else: - exec.run_frontend_prod(host, port) + exec.run_frontend_prod(host, port, on_started=on_started) def _run( diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index 69b4cb234b4..fc37b7db621 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -14,7 +14,7 @@ import subprocess import sys import threading -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from pathlib import Path from typing import Any, NamedTuple, TypedDict @@ -398,7 +398,9 @@ def _frontend_prod_app(): return Starlette(routes=[get_frontend_mount()]) -def run_frontend_prod(host: str, port: int): +def run_frontend_prod( + host: str, port: int, on_started: Callable[[], Any] | None = None +): """Run the frontend in production mode by serving compiled static files. Uses the same granian/uvicorn infrastructure as the backend. @@ -406,16 +408,25 @@ def run_frontend_prod(host: str, port: int): Args: host: The host to serve on. port: The port to serve on. + on_started: Called once the server's workers have been started. """ loglevel = get_config().loglevel.subprocess_level() if should_use_granian(): run_granian_backend_prod( - host, port, loglevel, app_target=f"{__name__}:_frontend_prod_app" + host, + port, + loglevel, + app_target=f"{__name__}:_frontend_prod_app", + on_started=on_started, ) else: run_uvicorn_backend_prod( - host, port, loglevel, app_target=f"{__name__}:_frontend_prod_app" + host, + port, + loglevel, + app_target=f"{__name__}:_frontend_prod_app", + on_started=on_started, ) @@ -869,6 +880,7 @@ def run_backend_prod( port: int, loglevel: constants.LogLevel = constants.LogLevel.ERROR, mount_frontend_compiled_app: bool = False, + on_started: Callable[[], Any] | None = None, ): """Run the backend. @@ -877,13 +889,14 @@ def run_backend_prod( port: The app port loglevel: The log level. mount_frontend_compiled_app: Whether to mount the compiled frontend app with the backend. + on_started: Called once the server's workers have been started. """ environment.REFLEX_MOUNT_FRONTEND_COMPILED_APP.set(mount_frontend_compiled_app) if should_use_granian(): - run_granian_backend_prod(host, port, loglevel) + run_granian_backend_prod(host, port, loglevel, on_started=on_started) else: - run_uvicorn_backend_prod(host, port, loglevel) + run_uvicorn_backend_prod(host, port, loglevel, on_started=on_started) def _get_backend_workers(): @@ -893,7 +906,11 @@ def _get_backend_workers(): def run_uvicorn_backend_prod( - host: str, port: int, loglevel: LogLevel, app_target: str | None = None + host: str, + port: int, + loglevel: LogLevel, + app_target: str | None = None, + on_started: Callable[[], Any] | None = None, ): """Run the backend in production mode using Uvicorn. @@ -902,6 +919,8 @@ def run_uvicorn_backend_prod( port: The app port loglevel: The log level. app_target: The ASGI app target to run. Defaults to the reflex app instance. + on_started: Called right before the server process is launched; it + runs in a separate interpreter, so nothing here is forked. """ import os import shlex @@ -944,6 +963,9 @@ def run_uvicorn_backend_prod( *("--log-level", loglevel.value), ] + if on_started is not None: + on_started() + processes.new_process( command, run=True, @@ -981,10 +1003,15 @@ def _preload_for_fork(app_target: str | None) -> None: """ import gc - from reflex.utils import prerequisites + from reflex_base.utils import serializers + + from reflex.utils import prerequisites, telemetry if app_target is None: prerequisites.get_app() + serializers._prepare_serializers_for_fork() + # Forking while the telemetry thread may hold a lock can deadlock the child. + telemetry._shutdown_executor() # Freezing keeps worker GC passes from writing to the preloaded objects' # headers, which would copy-on-write the shared pages private again. gc.collect() @@ -992,7 +1019,11 @@ def _preload_for_fork(app_target: str | None) -> None: def run_granian_backend_prod( - host: str, port: int, loglevel: LogLevel, app_target: str | None = None + host: str, + port: int, + loglevel: LogLevel, + app_target: str | None = None, + on_started: Callable[[], Any] | None = None, ): """Run the backend in production mode using Granian. @@ -1001,6 +1032,8 @@ def run_granian_backend_prod( port: The app port loglevel: The log level. app_target: The ASGI app target to run. Defaults to the reflex app instance. + on_started: Called in the supervisor once the workers have been started, + so work it does (e.g. a telemetry thread) is not forked into them. """ import multiprocessing @@ -1015,7 +1048,21 @@ def run_granian_backend_prod( if start_method == "fork": _preload_for_fork(app_target) - granian_app = Granian( + class NotifyingGranian(Granian): # pyright: ignore[reportGeneralTypeIssues] + """Granian server that reports when its workers have been started.""" + + def startup(self, *args, **kwargs): + """Start the supervisor and its workers, then notify the caller. + + Args: + args: Positional arguments for the Granian startup. + kwargs: Keyword arguments for the Granian startup. + """ + super().startup(*args, **kwargs) + if on_started is not None: + on_started() + + granian_app = NotifyingGranian( target=app_target or get_app_instance_from_file(), factory=True, address=host, diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index 8cac84f5494..abf36f2cea1 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -503,9 +503,23 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: return _executor -def _reset_executor_after_fork() -> None: - """Drop the inherited executor; its worker thread does not exist in the child.""" +def _shutdown_executor() -> None: + """Deliver queued telemetry and stop the worker thread. + + Called before forking so no telemetry thread is alive at the fork; the + next send lazily starts a new worker. + """ global _executor + with _executor_lock: + executor, _executor = _executor, None + if executor is not None: + executor.shutdown(wait=True) + + +def _reset_executor_after_fork() -> None: + """Drop the inherited executor and lock; the child owns neither's thread.""" + global _executor, _executor_lock + _executor_lock = threading.Lock() _executor = None diff --git a/tests/units/test_reflex.py b/tests/units/test_reflex.py index bd6d8c80382..af94fdf484f 100644 --- a/tests/units/test_reflex.py +++ b/tests/units/test_reflex.py @@ -10,6 +10,7 @@ import click import click.testing import pytest +from reflex_base import constants from reflex import reflex @@ -169,6 +170,40 @@ def test_compile_app_worker_flushes_telemetry_on_failure(mocker): flush.assert_called_once_with() +@pytest.mark.parametrize( + ("running_mode", "launcher"), + [ + (constants.RunningMode.BACKEND_ONLY, "run_backend_prod"), + (constants.RunningMode.FRONTEND_ONLY, "run_frontend_prod"), + ], +) +def test_run_prod_sends_telemetry_once_the_server_started( + mocker, running_mode: constants.RunningMode, launcher: str +): + """The run-prod event waits for the server so no telemetry thread is forked.""" + from reflex.utils import build, exec, processes, telemetry + + send = mocker.patch.object(telemetry, "send") + mocker.patch.object(reflex, "get_config") + mocker.patch.object(reflex, "_compile_app") + mocker.patch.object(reflex, "_skip_compile") + mocker.patch.object(build, "setup_frontend_prod") + mocker.patch.object(processes, "atexit_handler") + mocker.patch("atexit.register") + mocker.patch.object(exec, "notify_app_running") + mocker.patch.object(exec, "notify_frontend") + + def serve(*_args, on_started): + send.assert_not_called() + on_started() + + mocker.patch.object(exec, launcher, side_effect=serve) + + reflex._run_prod(running_mode, 8000, "127.0.0.1") + + send.assert_called_once_with("run-prod") + + def test_cloud_commands_registered(): """The hosting CLI commands import, resolve, and dispatch only on demand.""" probe = """ diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index fa5c55dae67..1f64f2d3e48 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -1,9 +1,11 @@ import asyncio import importlib.metadata import json +import os import sys import threading import uuid +import warnings from types import SimpleNamespace import pytest @@ -828,6 +830,63 @@ def test_executor_is_recreated_after_fork(): assert fresh.submit(lambda: 1).result(timeout=5) == 1 +def test_executor_lock_is_recreated_after_fork(): + """A lock held by another thread at fork time does not block the child.""" + held = telemetry._executor_lock + held.acquire() + try: + telemetry._reset_executor_after_fork() + finally: + held.release() + + assert telemetry._executor_lock is not held + assert not telemetry._executor_lock.locked() + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") +def test_forked_child_starts_with_fresh_executor_state(): + """The at-fork hook is registered and runs in a real forked child.""" + telemetry._get_telemetry_executor() + with warnings.catch_warnings(): + # The telemetry worker thread is alive, which is the point of the test. + warnings.simplefilter("ignore", DeprecationWarning) + with telemetry._executor_lock: + pid = os.fork() + if pid == 0: + clean = ( + telemetry._executor is None + and not telemetry._executor_lock.locked() + ) + os._exit(0 if clean else 1) + _, status = os.waitpid(pid, 0) + assert os.waitstatus_to_exitcode(status) == 0 + + +def test_shutdown_executor_drains_queued_work_and_stops_the_thread(): + """Queued events are delivered and no telemetry thread survives the call.""" + done = threading.Event() + executor = telemetry._get_telemetry_executor() + executor.submit(done.set) + + telemetry._shutdown_executor() + + assert done.is_set() + assert telemetry._executor is None + assert not any( + thread.name.startswith("reflex-telemetry") for thread in threading.enumerate() + ) + # The next send lazily starts a new worker. + assert telemetry._get_telemetry_executor().submit(lambda: 1).result(5) == 1 + + +def test_shutdown_executor_without_executor_is_a_noop(): + """Nothing to drain when no telemetry was ever sent.""" + telemetry._shutdown_executor() + telemetry._shutdown_executor() + + assert telemetry._executor is None + + def test_send_event_posts_json_without_httpx(mocker: MockerFixture): """Delivery goes through urllib, so backend workers never import httpx.""" urlopen = mocker.patch("reflex.utils.telemetry.urllib.request.urlopen") diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index abf4472b656..cfce7260843 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -19,7 +19,7 @@ from reflex_base.utils import serializers from reflex.utils import exec as exec_utils -from reflex.utils import prerequisites +from reflex.utils import prerequisites, telemetry DEV_BACKEND_RELOAD_ENV_NAME = environment.REFLEX_DEV_BACKEND_RELOAD_ACTIVE.name @@ -404,8 +404,12 @@ class FakeGranian: def __init__(self, *_args, **_kwargs): pass + def startup(self, *_args): + calls.append("workers") + def serve(self): calls.append("serve") + self.startup(None, None) mocker.patch.object(granian_server, "Server", FakeGranian) mocker.patch.object( @@ -420,22 +424,45 @@ def serve(self): mocker.patch.object( prerequisites, "get_app", side_effect=lambda: calls.append("preload") ) + mocker.patch.object( + serializers, + "_prepare_serializers_for_fork", + side_effect=lambda: calls.append("serializers"), + ) + mocker.patch.object( + telemetry, "_shutdown_executor", side_effect=lambda: calls.append("drain") + ) mocker.patch.object(gc, "freeze", side_effect=lambda: calls.append("freeze")) def test_run_granian_backend_prod_preloads_app_before_forking( mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch ): - """With fork, the app is imported and the heap frozen before workers start.""" + """With fork, the app is imported, telemetry drained and the heap frozen first. + + ``on_started`` runs only once the workers have been started. + """ monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") calls: list[str] = [] _fake_granian_prod(mocker, calls) exec_utils.run_granian_backend_prod( - host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + host="0.0.0.0", + port=8000, + loglevel=exec_utils.LogLevel.INFO, + on_started=lambda: calls.append("started"), ) - assert calls == ["start:fork", "preload", "freeze", "serve"] + assert calls == [ + "start:fork", + "preload", + "serializers", + "drain", + "freeze", + "serve", + "workers", + "started", + ] def test_run_granian_backend_prod_spawn_skips_preload( @@ -450,7 +477,7 @@ def test_run_granian_backend_prod_spawn_skips_preload( host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO ) - assert calls == ["start:spawn", "serve"] + assert calls == ["start:spawn", "serve", "workers"] def test_run_granian_backend_prod_custom_target_only_freezes( @@ -468,7 +495,65 @@ def test_run_granian_backend_prod_custom_target_only_freezes( app_target="reflex.utils.exec:_frontend_prod_app", ) - assert calls == ["start:fork", "freeze", "serve"] + assert calls == ["start:fork", "serializers", "drain", "freeze", "serve", "workers"] + + +@pytest.mark.parametrize("use_granian", [True, False]) +def test_run_backend_prod_forwards_on_started(mocker: MockerFixture, use_granian: bool): + """The prod backend hands ``on_started`` to whichever server it runs.""" + mocker.patch.object(exec_utils, "should_use_granian", return_value=use_granian) + launcher = mocker.patch.object( + exec_utils, + "run_granian_backend_prod" if use_granian else "run_uvicorn_backend_prod", + ) + on_started = mocker.Mock() + + exec_utils.run_backend_prod( + "0.0.0.0", 8000, exec_utils.LogLevel.INFO, on_started=on_started + ) + + launcher.assert_called_once_with( + "0.0.0.0", 8000, exec_utils.LogLevel.INFO, on_started=on_started + ) + + +@pytest.mark.parametrize("use_granian", [True, False]) +def test_run_frontend_prod_forwards_on_started( + mocker: MockerFixture, use_granian: bool +): + """The prod frontend server hands ``on_started`` to its launcher.""" + mocker.patch.object(exec_utils, "should_use_granian", return_value=use_granian) + launcher = mocker.patch.object( + exec_utils, + "run_granian_backend_prod" if use_granian else "run_uvicorn_backend_prod", + ) + on_started = mocker.Mock() + + exec_utils.run_frontend_prod("0.0.0.0", 3000, on_started=on_started) + + assert launcher.call_args.kwargs["on_started"] is on_started + + +def test_run_uvicorn_backend_prod_calls_on_started_before_launch( + mocker: MockerFixture, +): + """The server runs in a separate interpreter, so the callback runs first.""" + calls: list[str] = [] + mocker.patch.object(exec_utils, "_get_backend_workers", return_value=1) + mocker.patch( + "reflex.utils.processes.new_process", + side_effect=lambda *_args, **_kwargs: calls.append("launch"), + ) + + exec_utils.run_uvicorn_backend_prod( + "0.0.0.0", + 8000, + exec_utils.LogLevel.INFO, + app_target="app:app", + on_started=lambda: calls.append("started"), + ) + + assert calls == ["started", "launch"] @pytest.mark.parametrize( From 8e4d870d1bd29340275befa07cdd06490793f777 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 30 Sep 2026 23:24:18 +0000 Subject: [PATCH 03/18] Bound the pre-fork telemetry drain _shutdown_executor() waited on the telemetry executor without a limit, so a send stalled on DNS (urlopen's timeout does not cover getaddrinfo) or a slow event collector could block production startup forever. It now waits up to two seconds through _flush(), then cancels what is still queued and stops the pool without waiting for a stalled send; the forked children reset the executor and its lock through the at-fork hook. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/telemetry.py | 13 ++++++++++--- tests/units/test_telemetry.py | 20 ++++++++++++++++++++ 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index abf36f2cea1..45c7f0a3856 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -503,17 +503,24 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: return _executor -def _shutdown_executor() -> None: +def _shutdown_executor(timeout: float = 2) -> None: """Deliver queued telemetry and stop the worker thread. Called before forking so no telemetry thread is alive at the fork; the - next send lazily starts a new worker. + next send lazily starts a new worker. A send that stalls past the timeout + (e.g. on a DNS lookup) is abandoned so it cannot hold up startup. + + Args: + timeout: Maximum number of seconds to wait for queued telemetry. """ global _executor + drained = _flush(timeout) + if not drained: + logger.debug(f"Telemetry did not drain within {timeout}s before forking.") with _executor_lock: executor, _executor = _executor, None if executor is not None: - executor.shutdown(wait=True) + executor.shutdown(wait=drained, cancel_futures=True) def _reset_executor_after_fork() -> None: diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 1f64f2d3e48..0b03f6d3412 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -4,6 +4,7 @@ import os import sys import threading +import time import uuid import warnings from types import SimpleNamespace @@ -879,6 +880,25 @@ def test_shutdown_executor_drains_queued_work_and_stops_the_thread(): assert telemetry._get_telemetry_executor().submit(lambda: 1).result(5) == 1 +def test_shutdown_executor_gives_up_on_a_stalled_task(): + """A send that hangs (e.g. on DNS) cannot block the fork past the timeout.""" + release = threading.Event() + executor = telemetry._get_telemetry_executor() + stalled = executor.submit(release.wait, 10) + queued = executor.submit(lambda: None) + started = time.monotonic() + try: + telemetry._shutdown_executor(timeout=0.05) + elapsed = time.monotonic() - started + finally: + release.set() + stalled.result(timeout=5) + + assert elapsed < 2 + assert telemetry._executor is None + assert queued.cancelled() + + def test_shutdown_executor_without_executor_is_a_noop(): """Nothing to drain when no telemetry was ever sent.""" telemetry._shutdown_executor() From c7de30fb6fb0d4cb2fa3eac02ca615963fdea1f4 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 30 Sep 2026 23:28:59 +0000 Subject: [PATCH 04/18] Keep the telemetry pool published while it shuts down _shutdown_executor() cleared _executor before the old pool had stopped, so a send() racing the pre-fork shutdown could start a replacement pool whose thread was then alive at the fork. The pool now stays published until it has stopped (a racing submit fails and is suppressed) and is only then cleared. The shutdown test now checks the threads of the pool it stops instead of every "reflex-telemetry" thread in the process, and the fork-reset tests stop the pools they orphan, so the check no longer depends on test order. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/telemetry.py | 10 +++++++--- tests/units/test_telemetry.py | 31 ++++++++++++++++++++++++++++--- 2 files changed, 35 insertions(+), 6 deletions(-) diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index 45c7f0a3856..87326e757d9 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -514,13 +514,17 @@ def _shutdown_executor(timeout: float = 2) -> None: timeout: Maximum number of seconds to wait for queued telemetry. """ global _executor + if (executor := _executor) is None: + return drained = _flush(timeout) if not drained: logger.debug(f"Telemetry did not drain within {timeout}s before forking.") + # Stay published while stopping: a racing send() then fails to submit (and + # is suppressed) instead of starting a second pool before the fork. + executor.shutdown(wait=drained, cancel_futures=True) with _executor_lock: - executor, _executor = _executor, None - if executor is not None: - executor.shutdown(wait=drained, cancel_futures=True) + if _executor is executor: + _executor = None def _reset_executor_after_fork() -> None: diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 0b03f6d3412..d76a337ecdb 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -829,16 +829,20 @@ def test_executor_is_recreated_after_fork(): fresh = telemetry._get_telemetry_executor() assert fresh is not inherited assert fresh.submit(lambda: 1).result(timeout=5) == 1 + # In this process the orphaned pool's thread is real; stop it. + inherited.shutdown() def test_executor_lock_is_recreated_after_fork(): """A lock held by another thread at fork time does not block the child.""" + orphaned = telemetry._get_telemetry_executor() held = telemetry._executor_lock held.acquire() try: telemetry._reset_executor_after_fork() finally: held.release() + orphaned.shutdown() assert telemetry._executor_lock is not held assert not telemetry._executor_lock.locked() @@ -868,18 +872,39 @@ def test_shutdown_executor_drains_queued_work_and_stops_the_thread(): done = threading.Event() executor = telemetry._get_telemetry_executor() executor.submit(done.set) + workers = list(executor._threads) # pyright: ignore[reportAttributeAccessIssue] telemetry._shutdown_executor() assert done.is_set() assert telemetry._executor is None - assert not any( - thread.name.startswith("reflex-telemetry") for thread in threading.enumerate() - ) + assert workers + assert not any(worker.is_alive() for worker in workers) # The next send lazily starts a new worker. assert telemetry._get_telemetry_executor().submit(lambda: 1).result(5) == 1 +def test_shutdown_executor_keeps_the_pool_published_until_it_stops( + mocker: MockerFixture, +): + """A send racing the shutdown cannot start a second pool before the fork.""" + executor = telemetry._get_telemetry_executor() + seen = [] + stop = executor.shutdown + + def shutdown(*args, **kwargs): + # What a concurrent send() would submit to while the pool stops. + seen.append(telemetry._get_telemetry_executor()) + stop(*args, **kwargs) + + mocker.patch.object(executor, "shutdown", side_effect=shutdown) + + telemetry._shutdown_executor() + + assert seen == [executor] + assert telemetry._executor is None + + def test_shutdown_executor_gives_up_on_a_stalled_task(): """A send that hangs (e.g. on DNS) cannot block the fork past the timeout.""" release = threading.Event() From b8ef3cc2db35e1053b7493d353fc3d575dd8214f Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 30 Sep 2026 23:41:23 +0000 Subject: [PATCH 05/18] Spawn production workers when a telemetry send outlives the drain After the bounded drain, a stalled telemetry send (e.g. blocked in getaddrinfo) keeps its thread alive, and that thread can hold process-wide locks such as the resolver's across a fork. Forking then trades a startup hang for a worker hang. _shutdown_executor() now reports whether the pool's thread actually stopped, _preload_for_fork() passes that on, and run_granian_backend_prod() starts its workers with "spawn" instead of "fork" when it did not, so the supervisor never forks with a live telemetry thread and startup waits at most the drain timeout. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/exec.py | 14 ++++++++++---- reflex/utils/telemetry.py | 8 ++++++-- tests/units/test_telemetry.py | 11 ++++++----- tests/units/utils/test_exec.py | 34 ++++++++++++++++++++++++++++++---- 4 files changed, 52 insertions(+), 15 deletions(-) diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index fc37b7db621..ed5fa1a16b2 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -994,12 +994,15 @@ def _backend_start_method() -> str | None: return None -def _preload_for_fork(app_target: str | None) -> None: +def _preload_for_fork(app_target: str | None) -> bool: """Import the app in the supervisor so forked workers share its pages. Args: app_target: The ASGI app target; None means the reflex app, which is imported here. Any other target lives in an already-loaded module. + + Returns: + Whether forking is safe; False when a telemetry thread is still alive. """ import gc @@ -1011,11 +1014,13 @@ def _preload_for_fork(app_target: str | None) -> None: prerequisites.get_app() serializers._prepare_serializers_for_fork() # Forking while the telemetry thread may hold a lock can deadlock the child. - telemetry._shutdown_executor() + if not telemetry._shutdown_executor(): + return False # Freezing keeps worker GC passes from writing to the preloaded objects' # headers, which would copy-on-write the shared pages private again. gc.collect() gc.freeze() + return True def run_granian_backend_prod( @@ -1044,9 +1049,10 @@ def run_granian_backend_prod( logger.debug("Using Granian for backend") if (start_method := _backend_start_method()) is not None: + if start_method == "fork" and not _preload_for_fork(app_target): + logger.debug("A telemetry send is still running; spawning workers.") + start_method = "spawn" multiprocessing.set_start_method(start_method, force=True) - if start_method == "fork": - _preload_for_fork(app_target) class NotifyingGranian(Granian): # pyright: ignore[reportGeneralTypeIssues] """Granian server that reports when its workers have been started.""" diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index 87326e757d9..698106badb4 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -503,7 +503,7 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: return _executor -def _shutdown_executor(timeout: float = 2) -> None: +def _shutdown_executor(timeout: float = 2) -> bool: """Deliver queued telemetry and stop the worker thread. Called before forking so no telemetry thread is alive at the fork; the @@ -512,10 +512,13 @@ def _shutdown_executor(timeout: float = 2) -> None: Args: timeout: Maximum number of seconds to wait for queued telemetry. + + Returns: + Whether the worker thread has stopped, i.e. whether forking is safe. """ global _executor if (executor := _executor) is None: - return + return True drained = _flush(timeout) if not drained: logger.debug(f"Telemetry did not drain within {timeout}s before forking.") @@ -525,6 +528,7 @@ def _shutdown_executor(timeout: float = 2) -> None: with _executor_lock: if _executor is executor: _executor = None + return not any(thread.is_alive() for thread in executor._threads) def _reset_executor_after_fork() -> None: diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index d76a337ecdb..db192f0389c 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -872,9 +872,9 @@ def test_shutdown_executor_drains_queued_work_and_stops_the_thread(): done = threading.Event() executor = telemetry._get_telemetry_executor() executor.submit(done.set) - workers = list(executor._threads) # pyright: ignore[reportAttributeAccessIssue] + workers = list(executor._threads) - telemetry._shutdown_executor() + assert telemetry._shutdown_executor() assert done.is_set() assert telemetry._executor is None @@ -913,7 +913,8 @@ def test_shutdown_executor_gives_up_on_a_stalled_task(): queued = executor.submit(lambda: None) started = time.monotonic() try: - telemetry._shutdown_executor(timeout=0.05) + # The stalled send's thread is still alive, so forking is unsafe. + assert not telemetry._shutdown_executor(timeout=0.05) elapsed = time.monotonic() - started finally: release.set() @@ -926,8 +927,8 @@ def test_shutdown_executor_gives_up_on_a_stalled_task(): def test_shutdown_executor_without_executor_is_a_noop(): """Nothing to drain when no telemetry was ever sent.""" - telemetry._shutdown_executor() - telemetry._shutdown_executor() + assert telemetry._shutdown_executor() + assert telemetry._shutdown_executor() assert telemetry._executor is None diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index cfce7260843..24b77660523 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -396,7 +396,9 @@ def test_arbitrate_ssr_env_var_wins(monkeypatch: pytest.MonkeyPatch): assert exec_utils.arbitrate_ssr(True) is False -def _fake_granian_prod(mocker: MockerFixture, calls: list[str]): +def _fake_granian_prod( + mocker: MockerFixture, calls: list[str], telemetry_stopped: bool = True +): """Patch granian and the prod launcher's collaborators, recording call order.""" granian_server = pytest.importorskip("granian.server") @@ -430,7 +432,9 @@ def serve(self): side_effect=lambda: calls.append("serializers"), ) mocker.patch.object( - telemetry, "_shutdown_executor", side_effect=lambda: calls.append("drain") + telemetry, + "_shutdown_executor", + side_effect=lambda: calls.append("drain") or telemetry_stopped, ) mocker.patch.object(gc, "freeze", side_effect=lambda: calls.append("freeze")) @@ -454,17 +458,39 @@ def test_run_granian_backend_prod_preloads_app_before_forking( ) assert calls == [ - "start:fork", "preload", "serializers", "drain", "freeze", + "start:fork", "serve", "workers", "started", ] +def test_run_granian_backend_prod_spawns_when_telemetry_is_stuck( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """A telemetry thread that outlives the drain makes forking unsafe.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") + calls: list[str] = [] + _fake_granian_prod(mocker, calls, telemetry_stopped=False) + + exec_utils.run_granian_backend_prod( + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + ) + + assert calls == [ + "preload", + "serializers", + "drain", + "start:spawn", + "serve", + "workers", + ] + + def test_run_granian_backend_prod_spawn_skips_preload( mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch ): @@ -495,7 +521,7 @@ def test_run_granian_backend_prod_custom_target_only_freezes( app_target="reflex.utils.exec:_frontend_prod_app", ) - assert calls == ["start:fork", "serializers", "drain", "freeze", "serve", "workers"] + assert calls == ["serializers", "drain", "freeze", "start:fork", "serve", "workers"] @pytest.mark.parametrize("use_granian", [True, False]) From c3457a484c8f93dc3b817124fd49da94fcdbd495 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 30 Sep 2026 23:58:04 +0000 Subject: [PATCH 06/18] Set the start method before preloading and deflake the stalled drain test run_granian_backend_prod() imported the app before selecting the start method, so multiprocessing objects the app creates at import time used the interpreter default context (forkserver on Python 3.14 Linux) rather than the requested one. The requested method is now set first, and only switched to "spawn" when the telemetry drain leaves a live thread. The stalled-drain test now waits until the worker is running the stalled task before shutting down, and joins the worker instead of calling result() on a future that shutdown may have cancelled. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/exec.py | 6 ++++-- tests/units/test_telemetry.py | 15 +++++++++++++-- tests/units/utils/test_exec.py | 11 ++++++++--- 3 files changed, 25 insertions(+), 7 deletions(-) diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index ed5fa1a16b2..56d7f665df9 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -1049,10 +1049,12 @@ def run_granian_backend_prod( logger.debug("Using Granian for backend") if (start_method := _backend_start_method()) is not None: + # Set before the preload so the app's import-time multiprocessing + # objects use the requested context. + multiprocessing.set_start_method(start_method, force=True) if start_method == "fork" and not _preload_for_fork(app_target): logger.debug("A telemetry send is still running; spawning workers.") - start_method = "spawn" - multiprocessing.set_start_method(start_method, force=True) + multiprocessing.set_start_method("spawn", force=True) class NotifyingGranian(Granian): # pyright: ignore[reportGeneralTypeIssues] """Granian server that reports when its workers have been started.""" diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index db192f0389c..4642ee80094 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -907,10 +907,20 @@ def shutdown(*args, **kwargs): def test_shutdown_executor_gives_up_on_a_stalled_task(): """A send that hangs (e.g. on DNS) cannot block the fork past the timeout.""" + running = threading.Event() release = threading.Event() + + def stall(): + running.set() + release.wait(10) + executor = telemetry._get_telemetry_executor() - stalled = executor.submit(release.wait, 10) + workers = list(executor._threads) + executor.submit(stall) queued = executor.submit(lambda: None) + # Only a task the worker has picked up keeps it alive; a queued one would + # just be cancelled. + assert running.wait(5) started = time.monotonic() try: # The stalled send's thread is still alive, so forking is unsafe. @@ -918,7 +928,8 @@ def test_shutdown_executor_gives_up_on_a_stalled_task(): elapsed = time.monotonic() - started finally: release.set() - stalled.result(timeout=5) + for worker in workers: + worker.join(5) assert elapsed < 2 assert telemetry._executor is None diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index 24b77660523..a85b6ff999c 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -458,11 +458,11 @@ def test_run_granian_backend_prod_preloads_app_before_forking( ) assert calls == [ + "start:fork", "preload", "serializers", "drain", "freeze", - "start:fork", "serve", "workers", "started", @@ -472,7 +472,11 @@ def test_run_granian_backend_prod_preloads_app_before_forking( def test_run_granian_backend_prod_spawns_when_telemetry_is_stuck( mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch ): - """A telemetry thread that outlives the drain makes forking unsafe.""" + """A telemetry thread that outlives the drain makes forking unsafe. + + The app is still imported under the requested start method, so anything it + creates at import time uses the same context as before the fallback. + """ monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") calls: list[str] = [] _fake_granian_prod(mocker, calls, telemetry_stopped=False) @@ -482,6 +486,7 @@ def test_run_granian_backend_prod_spawns_when_telemetry_is_stuck( ) assert calls == [ + "start:fork", "preload", "serializers", "drain", @@ -521,7 +526,7 @@ def test_run_granian_backend_prod_custom_target_only_freezes( app_target="reflex.utils.exec:_frontend_prod_app", ) - assert calls == ["serializers", "drain", "freeze", "start:fork", "serve", "workers"] + assert calls == ["start:fork", "serializers", "drain", "freeze", "serve", "workers"] @pytest.mark.parametrize("use_granian", [True, False]) From 234ff44eaa0bc73853043acb52757d1d338ae91c Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 00:06:30 +0000 Subject: [PATCH 07/18] Keep telemetry from starting a thread between the drain and the fork Once _shutdown_executor() had stopped the pool, a send() racing the pending fork could create a fresh pool and thread that granian then forked. The shutdown now pauses pool creation (checked under the executor lock; such sends are dropped with a debug log) until the next fork, which lifts it in the parent through an after_in_parent hook and in the child through the existing after_in_child reset. A caller that keeps forking can pass resume_after_fork=False to stay paused. When the workers are forked, the supervisor stops the pool again after on_started has sent "run-prod", so granian respawns do not fork while a send is in flight. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/exec.py | 9 +++++- reflex/utils/telemetry.py | 54 +++++++++++++++++++++++++++------- tests/units/test_telemetry.py | 53 ++++++++++++++++++++++++++++++++- tests/units/utils/test_exec.py | 1 + 4 files changed, 105 insertions(+), 12 deletions(-) diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index 56d7f665df9..1d1948239bb 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -1046,6 +1046,8 @@ def run_granian_backend_prod( from granian.log import LogLevels from granian.server import Server as Granian + from reflex.utils import telemetry + logger.debug("Using Granian for backend") if (start_method := _backend_start_method()) is not None: @@ -1054,7 +1056,8 @@ def run_granian_backend_prod( multiprocessing.set_start_method(start_method, force=True) if start_method == "fork" and not _preload_for_fork(app_target): logger.debug("A telemetry send is still running; spawning workers.") - multiprocessing.set_start_method("spawn", force=True) + start_method = "spawn" + multiprocessing.set_start_method(start_method, force=True) class NotifyingGranian(Granian): # pyright: ignore[reportGeneralTypeIssues] """Granian server that reports when its workers have been started.""" @@ -1069,6 +1072,10 @@ def startup(self, *args, **kwargs): super().startup(*args, **kwargs) if on_started is not None: on_started() + if start_method == "fork": + # Respawned workers fork too: stop the thread the + # callback may have started. + telemetry._shutdown_executor() granian_app = NotifyingGranian( target=app_target or get_app_instance_from_file(), diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index 698106badb4..a7c0cb3d3af 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -478,6 +478,10 @@ def _send( _executor_lock = threading.Lock() _executor: ThreadPoolExecutor | None = None +# Set by _shutdown_executor() so no new pool (and thread) starts before a +# pending fork; _resume_after_fork says whether that fork lifts it again. +_paused = False +_resume_after_fork = True def _get_telemetry_executor() -> ThreadPoolExecutor: @@ -492,10 +496,16 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: Returns: The shared single-worker telemetry executor. + + Raises: + RuntimeError: When the pool is stopped for a pending fork. """ global _executor if _executor is None: with _executor_lock: + if _paused: + msg = "Telemetry is paused for a pending fork." + raise RuntimeError(msg) if _executor is None: _executor = ThreadPoolExecutor( max_workers=1, thread_name_prefix="reflex-telemetry" @@ -503,21 +513,28 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: return _executor -def _shutdown_executor(timeout: float = 2) -> bool: +def _shutdown_executor(timeout: float = 2, *, resume_after_fork: bool = True) -> bool: """Deliver queued telemetry and stop the worker thread. - Called before forking so no telemetry thread is alive at the fork; the - next send lazily starts a new worker. A send that stalls past the timeout - (e.g. on a DNS lookup) is abandoned so it cannot hold up startup. + Called before forking so no telemetry thread is alive at the fork. Until + then no new worker starts and sends are dropped. A send that stalls past + the timeout (e.g. on a DNS lookup) is abandoned so it cannot hold up + startup. Args: timeout: Maximum number of seconds to wait for queued telemetry. + resume_after_fork: Whether the parent sends again after the next fork, + or stays paused for a supervisor that keeps forking. Returns: Whether the worker thread has stopped, i.e. whether forking is safe. """ - global _executor - if (executor := _executor) is None: + global _executor, _paused, _resume_after_fork + with _executor_lock: + _paused = True + _resume_after_fork = resume_after_fork + executor = _executor + if executor is None: return True drained = _flush(timeout) if not drained: @@ -525,21 +542,36 @@ def _shutdown_executor(timeout: float = 2) -> bool: # Stay published while stopping: a racing send() then fails to submit (and # is suppressed) instead of starting a second pool before the fork. executor.shutdown(wait=drained, cancel_futures=True) + stopped = not any(thread.is_alive() for thread in executor._threads) with _executor_lock: if _executor is executor: _executor = None - return not any(thread.is_alive() for thread in executor._threads) + if not stopped: + # The caller will not fork with a live thread, so sends may resume. + _paused = False + return stopped + + +def _resume_in_parent_after_fork() -> None: + """Let the parent send again once the fork it paused for has happened.""" + global _paused + if _resume_after_fork: + _paused = False def _reset_executor_after_fork() -> None: """Drop the inherited executor and lock; the child owns neither's thread.""" - global _executor, _executor_lock + global _executor, _executor_lock, _paused _executor_lock = threading.Lock() _executor = None + _paused = False if hasattr(os, "register_at_fork"): - os.register_at_fork(after_in_child=_reset_executor_after_fork) + os.register_at_fork( + after_in_parent=_resume_in_parent_after_fork, + after_in_child=_reset_executor_after_fork, + ) def _current_registration_context() -> RegistrationContext | None: @@ -612,10 +644,12 @@ def _submit(fn: Callable[..., Any], /, *args, **kwargs) -> None: kwargs: Keyword arguments forwarded to ``fn``. """ registration_context = _current_registration_context() - with suppress(Exception): + try: _get_telemetry_executor().submit( _run_suppressed, registration_context, fn, *args, **kwargs ) + except Exception as err: + logger.debug(f"Dropped telemetry event: {err}") def _flush(timeout: float | None = None) -> bool: diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 4642ee80094..49182bbce52 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -33,6 +33,8 @@ def _drain_telemetry_executor(): """ yield telemetry._flush() + # A test that stops the pool for a fork must not silence the next test. + telemetry._paused = False @pytest.fixture @@ -880,7 +882,8 @@ def test_shutdown_executor_drains_queued_work_and_stops_the_thread(): assert telemetry._executor is None assert workers assert not any(worker.is_alive() for worker in workers) - # The next send lazily starts a new worker. + # Once the fork has happened, the next send lazily starts a new worker. + telemetry._resume_in_parent_after_fork() assert telemetry._get_telemetry_executor().submit(lambda: 1).result(5) == 1 @@ -936,6 +939,54 @@ def stall(): assert queued.cancelled() +def test_send_after_pre_fork_shutdown_starts_no_thread(mocker: MockerFixture): + """Until the pending fork happens, a send cannot start a telemetry thread.""" + job = mocker.Mock() + telemetry._get_telemetry_executor() + assert telemetry._shutdown_executor() + + telemetry._submit(job) + + assert telemetry._executor is None + job.assert_not_called() + + +def test_parent_resumes_sending_after_the_fork(): + """The parent sends again once the fork it paused for has happened.""" + assert telemetry._shutdown_executor() + + telemetry._resume_in_parent_after_fork() + + assert telemetry._get_telemetry_executor().submit(lambda: 1).result(5) == 1 + + +def test_supervisor_can_stay_paused_across_forks(): + """A supervisor that keeps forking can keep telemetry paused for good.""" + assert telemetry._shutdown_executor(resume_after_fork=False) + + telemetry._resume_in_parent_after_fork() + + with pytest.raises(RuntimeError, match="paused"): + telemetry._get_telemetry_executor() + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") +def test_real_fork_resumes_parent_and_child(): + """The at-fork hooks lift the pause in both the parent and the child.""" + assert telemetry._shutdown_executor() + pid = os.fork() + if pid == 0: + code = 1 + try: + code = int(telemetry._paused) + finally: + os._exit(code) + _, status = os.waitpid(pid, 0) + + assert os.waitstatus_to_exitcode(status) == 0 + assert not telemetry._paused + + def test_shutdown_executor_without_executor_is_a_noop(): """Nothing to drain when no telemetry was ever sent.""" assert telemetry._shutdown_executor() diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index a85b6ff999c..54924f95973 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -466,6 +466,7 @@ def test_run_granian_backend_prod_preloads_app_before_forking( "serve", "workers", "started", + "drain", ] From f3712b5053dec7d31afa347b083f9211ef022e3e Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 00:12:04 +0000 Subject: [PATCH 08/18] Silence the multithreaded-fork warning in the telemetry fork test Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- tests/units/test_telemetry.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 49182bbce52..a49f1e16577 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -974,7 +974,10 @@ def test_supervisor_can_stay_paused_across_forks(): def test_real_fork_resumes_parent_and_child(): """The at-fork hooks lift the pause in both the parent and the child.""" assert telemetry._shutdown_executor() - pid = os.fork() + with warnings.catch_warnings(): + # Other tests leave unrelated threads behind in this process. + warnings.simplefilter("ignore", DeprecationWarning) + pid = os.fork() if pid == 0: code = 1 try: From 91e321c20d52433593b2e6223e8219619e83809a Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 00:23:19 +0000 Subject: [PATCH 09/18] Send the deferred run event from a detached process The supervisor that forks granian workers also forks again to respawn them, so it must never hold a telemetry thread: a "run-prod" send that stalled (DNS, the lscpu collector) after on_started kept one alive across those respawns, and the earlier bounded re-shutdown could not help. Once the supervisor drains and stops its pool before the first fork, it now stays paused for good (in-process sends are dropped with a debug log; forked children reset it). on_started hands "run-prod" to telemetry._send_detached(), which runs the event collection and delivery in a fresh interpreter: Popen with close_fds (so the server sockets stay in the supervisor) runs a snippet that forks once and exits, so the supervisor reaps its direct child at once and the sender is reparented to init. No process starts when telemetry is disabled. The after-fork parent hook and the post-callback shutdown are gone. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/reflex.py | 9 ++-- reflex/utils/exec.py | 14 ++--- reflex/utils/telemetry.py | 77 +++++++++++++++++++--------- tests/units/test_reflex.py | 8 ++- tests/units/test_telemetry.py | 93 ++++++++++++++++++++++++++-------- tests/units/utils/test_exec.py | 1 - 6 files changed, 139 insertions(+), 63 deletions(-) diff --git a/reflex/reflex.py b/reflex/reflex.py index a874ef8df9d..106d58ca212 100644 --- a/reflex/reflex.py +++ b/reflex/reflex.py @@ -515,11 +515,12 @@ def _run_prod(running_mode: constants.RunningMode, port: int, host: str): _skip_compile() - # Post the telemetry event once the workers are running: sending it here - # would start the telemetry thread before the server forks its workers. + # Post the telemetry event once the workers are running, from a separate + # process: the server may fork its supervisor again to respawn workers, so + # it must never hold a telemetry thread. def on_started(): - """Send the run telemetry from the supervisor.""" - telemetry.send("run-prod") + """Send the run telemetry without a thread in the supervisor.""" + telemetry._send_detached("run-prod") # Display custom message when there is a keyboard interrupt. atexit.register(processes.atexit_handler) diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index 1d1948239bb..856cce6b400 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -1037,8 +1037,9 @@ def run_granian_backend_prod( port: The app port loglevel: The log level. app_target: The ASGI app target to run. Defaults to the reflex app instance. - on_started: Called in the supervisor once the workers have been started, - so work it does (e.g. a telemetry thread) is not forked into them. + on_started: Called in the supervisor once the workers have been started. + It must not start threads: granian forks the supervisor again to + respawn workers. """ import multiprocessing @@ -1046,8 +1047,6 @@ def run_granian_backend_prod( from granian.log import LogLevels from granian.server import Server as Granian - from reflex.utils import telemetry - logger.debug("Using Granian for backend") if (start_method := _backend_start_method()) is not None: @@ -1056,8 +1055,7 @@ def run_granian_backend_prod( multiprocessing.set_start_method(start_method, force=True) if start_method == "fork" and not _preload_for_fork(app_target): logger.debug("A telemetry send is still running; spawning workers.") - start_method = "spawn" - multiprocessing.set_start_method(start_method, force=True) + multiprocessing.set_start_method("spawn", force=True) class NotifyingGranian(Granian): # pyright: ignore[reportGeneralTypeIssues] """Granian server that reports when its workers have been started.""" @@ -1072,10 +1070,6 @@ def startup(self, *args, **kwargs): super().startup(*args, **kwargs) if on_started is not None: on_started() - if start_method == "fork": - # Respawned workers fork too: stop the thread the - # callback may have started. - telemetry._shutdown_executor() granian_app = NotifyingGranian( target=app_target or get_app_instance_from_file(), diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index a7c0cb3d3af..c2d5d44d01d 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -7,6 +7,7 @@ import multiprocessing import os import platform +import subprocess import sys import threading import urllib.request @@ -478,10 +479,9 @@ def _send( _executor_lock = threading.Lock() _executor: ThreadPoolExecutor | None = None -# Set by _shutdown_executor() so no new pool (and thread) starts before a -# pending fork; _resume_after_fork says whether that fork lifts it again. +# Set by _shutdown_executor(): a process that is about to fork, and may fork +# again later, never starts another pool (and thread). Forked children reset it. _paused = False -_resume_after_fork = True def _get_telemetry_executor() -> ThreadPoolExecutor: @@ -513,26 +513,24 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: return _executor -def _shutdown_executor(timeout: float = 2, *, resume_after_fork: bool = True) -> bool: - """Deliver queued telemetry and stop the worker thread. +def _shutdown_executor(timeout: float = 2) -> bool: + """Deliver queued telemetry and stop the worker thread for good. - Called before forking so no telemetry thread is alive at the fork. Until - then no new worker starts and sends are dropped. A send that stalls past - the timeout (e.g. on a DNS lookup) is abandoned so it cannot hold up - startup. + Called by a supervisor before it forks workers, so no telemetry thread is + alive at that fork or any later one (e.g. a worker respawn). From then on + in-process sends are dropped; use ``_send_detached`` instead. A send that + stalls past the timeout (e.g. on a DNS lookup) is abandoned so it cannot + hold up startup, and the caller must then not fork. Args: timeout: Maximum number of seconds to wait for queued telemetry. - resume_after_fork: Whether the parent sends again after the next fork, - or stays paused for a supervisor that keeps forking. Returns: Whether the worker thread has stopped, i.e. whether forking is safe. """ - global _executor, _paused, _resume_after_fork + global _executor, _paused with _executor_lock: _paused = True - _resume_after_fork = resume_after_fork executor = _executor if executor is None: return True @@ -552,13 +550,6 @@ def _shutdown_executor(timeout: float = 2, *, resume_after_fork: bool = True) -> return stopped -def _resume_in_parent_after_fork() -> None: - """Let the parent send again once the fork it paused for has happened.""" - global _paused - if _resume_after_fork: - _paused = False - - def _reset_executor_after_fork() -> None: """Drop the inherited executor and lock; the child owns neither's thread.""" global _executor, _executor_lock, _paused @@ -568,10 +559,48 @@ def _reset_executor_after_fork() -> None: if hasattr(os, "register_at_fork"): - os.register_at_fork( - after_in_parent=_resume_in_parent_after_fork, - after_in_child=_reset_executor_after_fork, - ) + os.register_at_fork(after_in_child=_reset_executor_after_fork) + + +# Run by _send_detached in a fresh interpreter. It forks once more and exits, +# so the supervisor reaps it at once and the sender is reparented to init. +_DETACHED_SEND = """\ +import os +import sys + +if os.fork(): + os._exit(0) +os.setsid() +from reflex.utils import telemetry + +telemetry._process_event(sys.argv[1], True) +telemetry._flush(30) +os._exit(0) +""" + + +def _send_detached(event: str) -> None: + """Send an event from a short-lived detached process. + + For a supervisor that has stopped its telemetry thread to fork workers + safely: the event is collected and delivered by a fresh interpreter, so + the supervisor never gains a thread or waits on the network. The direct + child exits right after detaching the sender, so no zombie is left, and + ``close_fds`` keeps the server's sockets out of it. + + Args: + event: The event name. + """ + with suppress(Exception): + if not get_config().telemetry_enabled: + return + subprocess.Popen( + [sys.executable, "-c", _DETACHED_SEND, event], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + close_fds=True, + ).wait(timeout=10) def _current_registration_context() -> RegistrationContext | None: diff --git a/tests/units/test_reflex.py b/tests/units/test_reflex.py index af94fdf484f..f1b7a07d392 100644 --- a/tests/units/test_reflex.py +++ b/tests/units/test_reflex.py @@ -180,10 +180,14 @@ def test_compile_app_worker_flushes_telemetry_on_failure(mocker): def test_run_prod_sends_telemetry_once_the_server_started( mocker, running_mode: constants.RunningMode, launcher: str ): - """The run-prod event waits for the server so no telemetry thread is forked.""" + """The run-prod event waits for the server and leaves the supervisor. + + Sending it from another process keeps the supervisor, which granian may + fork again to respawn workers, free of telemetry threads. + """ from reflex.utils import build, exec, processes, telemetry - send = mocker.patch.object(telemetry, "send") + send = mocker.patch.object(telemetry, "_send_detached") mocker.patch.object(reflex, "get_config") mocker.patch.object(reflex, "_compile_app") mocker.patch.object(reflex, "_skip_compile") diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index a49f1e16577..e5eb1cb182c 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -882,9 +882,6 @@ def test_shutdown_executor_drains_queued_work_and_stops_the_thread(): assert telemetry._executor is None assert workers assert not any(worker.is_alive() for worker in workers) - # Once the fork has happened, the next send lazily starts a new worker. - telemetry._resume_in_parent_after_fork() - assert telemetry._get_telemetry_executor().submit(lambda: 1).result(5) == 1 def test_shutdown_executor_keeps_the_pool_published_until_it_stops( @@ -951,28 +948,27 @@ def test_send_after_pre_fork_shutdown_starts_no_thread(mocker: MockerFixture): job.assert_not_called() -def test_parent_resumes_sending_after_the_fork(): - """The parent sends again once the fork it paused for has happened.""" - assert telemetry._shutdown_executor() - - telemetry._resume_in_parent_after_fork() - - assert telemetry._get_telemetry_executor().submit(lambda: 1).result(5) == 1 - - -def test_supervisor_can_stay_paused_across_forks(): - """A supervisor that keeps forking can keep telemetry paused for good.""" - assert telemetry._shutdown_executor(resume_after_fork=False) - - telemetry._resume_in_parent_after_fork() +def test_stalled_shutdown_lifts_the_pause(): + """A caller that will not fork may send in-process again.""" + release = threading.Event() + running = threading.Event() + executor = telemetry._get_telemetry_executor() + workers = list(executor._threads) + executor.submit(lambda: running.set() or release.wait(10)) + assert running.wait(5) + try: + assert not telemetry._shutdown_executor(timeout=0.05) + finally: + release.set() + for worker in workers: + worker.join(5) - with pytest.raises(RuntimeError, match="paused"): - telemetry._get_telemetry_executor() + assert not telemetry._paused @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") -def test_real_fork_resumes_parent_and_child(): - """The at-fork hooks lift the pause in both the parent and the child.""" +def test_real_fork_keeps_the_parent_paused(): + """A supervisor that may fork again never sends in-process; its child may.""" assert telemetry._shutdown_executor() with warnings.catch_warnings(): # Other tests leave unrelated threads behind in this process. @@ -987,7 +983,60 @@ def test_real_fork_resumes_parent_and_child(): _, status = os.waitpid(pid, 0) assert os.waitstatus_to_exitcode(status) == 0 - assert not telemetry._paused + assert telemetry._paused + with pytest.raises(RuntimeError, match="paused"): + telemetry._get_telemetry_executor() + + +def test_send_detached_hands_the_event_to_a_detached_process( + mocker: MockerFixture, +): + """The supervisor sends through a short-lived process, never a thread.""" + mocker.patch.object( + telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) + ) + popen = mocker.patch.object(telemetry.subprocess, "Popen") + + telemetry._send_detached("run-prod") + + popen.assert_called_once_with( + [sys.executable, "-c", telemetry._DETACHED_SEND, "run-prod"], + stdin=telemetry.subprocess.DEVNULL, + stdout=telemetry.subprocess.DEVNULL, + stderr=telemetry.subprocess.DEVNULL, + close_fds=True, + ) + # The direct child exits right after detaching, so it is reaped at once. + popen.return_value.wait.assert_called_once() + assert telemetry._executor is None + + +def test_send_detached_starts_nothing_when_disabled(mocker: MockerFixture): + """With telemetry disabled no process is started.""" + mocker.patch.object( + telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=False) + ) + popen = mocker.patch.object(telemetry.subprocess, "Popen") + + telemetry._send_detached("run-prod") + + popen.assert_not_called() + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") +def test_detached_send_leaves_no_child_behind(mocker: MockerFixture): + """The direct child exits after detaching, so no zombie is left to reap.""" + mocker.patch.object( + telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) + ) + # Same detaching prologue, but the grandchild exits instead of sending. + prologue = telemetry._DETACHED_SEND.split("from reflex")[0] + mocker.patch.object(telemetry, "_DETACHED_SEND", prologue + "os._exit(0)\n") + popen = mocker.spy(telemetry.subprocess, "Popen") + + telemetry._send_detached("run-prod") + + assert popen.spy_return.returncode == 0 def test_shutdown_executor_without_executor_is_a_noop(): diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index 54924f95973..a85b6ff999c 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -466,7 +466,6 @@ def test_run_granian_backend_prod_preloads_app_before_forking( "serve", "workers", "started", - "drain", ] From e919fb76781371ce56eb3cc052bf409b9f1e4ba4 Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 00:23:57 +0000 Subject: [PATCH 10/18] Send the detached run event in-process where nothing forks Windows has no os.fork, so the detaching snippet cannot run there; the supervisor never forks on that platform either, so send in-process. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/telemetry.py | 4 ++++ tests/units/test_telemetry.py | 15 +++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index c2d5d44d01d..a5df9e8e0ab 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -591,6 +591,10 @@ def _send_detached(event: str) -> None: Args: event: The event name. """ + if not hasattr(os, "fork"): + # Nothing forks here (Windows), so an in-process send is safe. + send(event) + return with suppress(Exception): if not get_config().telemetry_enabled: return diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index e5eb1cb182c..12014e6500f 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -988,6 +988,7 @@ def test_real_fork_keeps_the_parent_paused(): telemetry._get_telemetry_executor() +@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") def test_send_detached_hands_the_event_to_a_detached_process( mocker: MockerFixture, ): @@ -1011,6 +1012,20 @@ def test_send_detached_hands_the_event_to_a_detached_process( assert telemetry._executor is None +def test_send_detached_sends_in_process_without_fork( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """Where nothing forks (Windows) the event is sent in-process.""" + monkeypatch.delattr(os, "fork", raising=False) + send = mocker.patch.object(telemetry, "send") + popen = mocker.patch.object(telemetry.subprocess, "Popen") + + telemetry._send_detached("run-prod") + + send.assert_called_once_with("run-prod") + popen.assert_not_called() + + def test_send_detached_starts_nothing_when_disabled(mocker: MockerFixture): """With telemetry disabled no process is started.""" mocker.patch.object( From 687db6cb8ed157508d3dcc90f352702b0e5ada42 Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 00:36:33 +0000 Subject: [PATCH 11/18] Use the detached sender only in a process paused to fork _send_detached() started a fresh interpreter and waited for it even when nothing forks the process: the uvicorn/gunicorn path calls on_started right before launching the server, so the launch waited on that child. Only a process that paused telemetry before forking its workers needs the detached sender; every other process (uvicorn, granian with spawn or forkserver, a failed pre-fork drain that fell back to spawn) sends in-process as before. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/telemetry.py | 18 ++++++++++-------- tests/units/test_telemetry.py | 28 ++++++++++++++++++++++++---- 2 files changed, 34 insertions(+), 12 deletions(-) diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index a5df9e8e0ab..5a0a1c9dd08 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -580,19 +580,21 @@ def _reset_executor_after_fork() -> None: def _send_detached(event: str) -> None: - """Send an event from a short-lived detached process. + """Send an event without a telemetry thread in a process that forks. - For a supervisor that has stopped its telemetry thread to fork workers - safely: the event is collected and delivered by a fresh interpreter, so - the supervisor never gains a thread or waits on the network. The direct - child exits right after detaching the sender, so no zombie is left, and - ``close_fds`` keeps the server's sockets out of it. + A supervisor that paused telemetry to fork workers safely hands the event + to a fresh interpreter, so it never gains a thread or waits on the + network. The direct child exits right after detaching the sender, so no + zombie is left, and ``close_fds`` keeps the server's sockets out of it. + Any other process sends in-process without waiting on a child. Args: event: The event name. """ - if not hasattr(os, "fork"): - # Nothing forks here (Windows), so an in-process send is safe. + with _executor_lock: + paused = _paused + if not paused or not hasattr(os, "fork"): + # Nothing forks this process with a telemetry thread alive. send(event) return with suppress(Exception): diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 12014e6500f..1759cbe3411 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -990,9 +990,10 @@ def test_real_fork_keeps_the_parent_paused(): @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") def test_send_detached_hands_the_event_to_a_detached_process( - mocker: MockerFixture, + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch ): - """The supervisor sends through a short-lived process, never a thread.""" + """A supervisor paused to fork sends through a short-lived process.""" + monkeypatch.setattr(telemetry, "_paused", True) mocker.patch.object( telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) ) @@ -1012,6 +1013,19 @@ def test_send_detached_hands_the_event_to_a_detached_process( assert telemetry._executor is None +def test_send_detached_sends_in_process_when_not_paused(mocker: MockerFixture): + """A process that does not fork (uvicorn, spawned granian) does not wait + on a child process: it sends in-process, as before. + """ + send = mocker.patch.object(telemetry, "send") + popen = mocker.patch.object(telemetry.subprocess, "Popen") + + telemetry._send_detached("run-prod") + + send.assert_called_once_with("run-prod") + popen.assert_not_called() + + def test_send_detached_sends_in_process_without_fork( mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch ): @@ -1026,8 +1040,11 @@ def test_send_detached_sends_in_process_without_fork( popen.assert_not_called() -def test_send_detached_starts_nothing_when_disabled(mocker: MockerFixture): +def test_send_detached_starts_nothing_when_disabled( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): """With telemetry disabled no process is started.""" + monkeypatch.setattr(telemetry, "_paused", True) mocker.patch.object( telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=False) ) @@ -1039,8 +1056,11 @@ def test_send_detached_starts_nothing_when_disabled(mocker: MockerFixture): @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") -def test_detached_send_leaves_no_child_behind(mocker: MockerFixture): +def test_detached_send_leaves_no_child_behind( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): """The direct child exits after detaching, so no zombie is left to reap.""" + monkeypatch.setattr(telemetry, "_paused", True) mocker.patch.object( telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) ) From b5f800a6dfebaeb6ef25493a692394e48cd8a298 Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 00:48:52 +0000 Subject: [PATCH 12/18] Reap the detached sender reliably and keep its tests offline - A direct child that outlives the wait is now killed and reaped instead of being left behind when TimeoutExpired was suppressed. - As PID 1 (a container without an init) or a Linux child subreaper, the double-forked sender would be reparented to the supervisor, which never reaps it. There the single child sends itself and is waited for (bounded, then killed and reaped), which only delays the supervisor's management loop, never the workers. - The snippet is split into its detaching prologue and its send body, so the tests swap in a self-contained body instead of slicing the real snippet, which could have posted real telemetry if it were reworded. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/telemetry.py | 60 +++++++++++++++++++++----- tests/units/test_telemetry.py | 80 +++++++++++++++++++++++++++++++++-- 2 files changed, 126 insertions(+), 14 deletions(-) diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index 5a0a1c9dd08..1144ef0cbb4 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -562,21 +562,52 @@ def _reset_executor_after_fork() -> None: os.register_at_fork(after_in_child=_reset_executor_after_fork) -# Run by _send_detached in a fresh interpreter. It forks once more and exits, -# so the supervisor reaps it at once and the sender is reparented to init. -_DETACHED_SEND = """\ +# Run by _send_detached in a fresh interpreter (argv: event, mode). In +# "detach" mode it forks once more and exits, so the supervisor reaps it at +# once and the sender is reparented to whoever reaps orphans. +_DETACH = """\ import os import sys -if os.fork(): - os._exit(0) -os.setsid() +if sys.argv[2] == "detach": + if os.fork(): + os._exit(0) + os.setsid() +""" +_SEND_EVENT = """\ from reflex.utils import telemetry telemetry._process_event(sys.argv[1], True) telemetry._flush(30) os._exit(0) """ +# Seconds the supervisor waits for its direct child before killing it. +_DETACHED_TIMEOUT = 10 +_PR_GET_CHILD_SUBREAPER = 37 + + +def _reaps_orphans() -> bool: + """Report whether orphaned descendants are reparented to this process. + + True as PID 1 (a container without an init) or as a Linux child + subreaper. Such a process would inherit a detached sender, and a server + supervisor never reaps it. + + Returns: + Whether this process inherits orphans. + """ + if os.getpid() == 1: + return True + if sys.platform != "linux": + return False + import ctypes + + flag = ctypes.c_int() + with suppress(Exception): + prctl = ctypes.CDLL(None, use_errno=True).prctl + if prctl(_PR_GET_CHILD_SUBREAPER, ctypes.byref(flag), 0, 0, 0) == 0: + return bool(flag.value) + return False def _send_detached(event: str) -> None: @@ -586,7 +617,10 @@ def _send_detached(event: str) -> None: to a fresh interpreter, so it never gains a thread or waits on the network. The direct child exits right after detaching the sender, so no zombie is left, and ``close_fds`` keeps the server's sockets out of it. - Any other process sends in-process without waiting on a child. + Where orphans would come back to this process (PID 1, a subreaper), the + direct child sends itself and is waited for instead. A child that + outlives ``_DETACHED_TIMEOUT`` is killed and reaped. Any other process + sends in-process without waiting on a child. Args: event: The event name. @@ -600,13 +634,19 @@ def _send_detached(event: str) -> None: with suppress(Exception): if not get_config().telemetry_enabled: return - subprocess.Popen( - [sys.executable, "-c", _DETACHED_SEND, event], + mode = "attached" if _reaps_orphans() else "detach" + child = subprocess.Popen( + [sys.executable, "-c", _DETACH + _SEND_EVENT, event, mode], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, close_fds=True, - ).wait(timeout=10) + ) + try: + child.wait(timeout=_DETACHED_TIMEOUT) + except subprocess.TimeoutExpired: + child.kill() + child.wait() def _current_registration_context() -> RegistrationContext | None: diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 1759cbe3411..1f842bfad3c 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -7,6 +7,7 @@ import time import uuid import warnings +from pathlib import Path from types import SimpleNamespace import pytest @@ -998,11 +999,18 @@ def test_send_detached_hands_the_event_to_a_detached_process( telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) ) popen = mocker.patch.object(telemetry.subprocess, "Popen") + mocker.patch.object(telemetry, "_reaps_orphans", return_value=False) telemetry._send_detached("run-prod") popen.assert_called_once_with( - [sys.executable, "-c", telemetry._DETACHED_SEND, "run-prod"], + [ + sys.executable, + "-c", + telemetry._DETACH + telemetry._SEND_EVENT, + "run-prod", + "detach", + ], stdin=telemetry.subprocess.DEVNULL, stdout=telemetry.subprocess.DEVNULL, stderr=telemetry.subprocess.DEVNULL, @@ -1064,9 +1072,9 @@ def test_detached_send_leaves_no_child_behind( mocker.patch.object( telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) ) - # Same detaching prologue, but the grandchild exits instead of sending. - prologue = telemetry._DETACHED_SEND.split("from reflex")[0] - mocker.patch.object(telemetry, "_DETACHED_SEND", prologue + "os._exit(0)\n") + mocker.patch.object(telemetry, "_reaps_orphans", return_value=False) + # The real detaching prologue, but the sender exits instead of sending. + mocker.patch.object(telemetry, "_SEND_EVENT", "os._exit(0)\n") popen = mocker.spy(telemetry.subprocess, "Popen") telemetry._send_detached("run-prod") @@ -1074,6 +1082,70 @@ def test_detached_send_leaves_no_child_behind( assert popen.spy_return.returncode == 0 +@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") +def test_detached_send_kills_and_reaps_a_child_that_hangs( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """A child that does not exit in time is killed and reaped, not leaked.""" + monkeypatch.setattr(telemetry, "_paused", True) + monkeypatch.setattr(telemetry, "_DETACHED_TIMEOUT", 0.2) + mocker.patch.object( + telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) + ) + mocker.patch.object(telemetry, "_reaps_orphans", return_value=True) + mocker.patch.object(telemetry, "_SEND_EVENT", "import time\ntime.sleep(30)\n") + popen = mocker.spy(telemetry.subprocess, "Popen") + + telemetry._send_detached("run-prod") + + assert popen.spy_return.returncode == -9 + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") +def test_detached_send_stays_attached_where_orphans_come_back( + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + """As PID 1 or a subreaper the sender is not orphaned to this process. + + Orphans would be reparented to the supervisor, which never reaps them, + so the single child sends and is waited for instead. + """ + monkeypatch.setattr(telemetry, "_paused", True) + mocker.patch.object( + telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) + ) + mocker.patch.object(telemetry, "_reaps_orphans", return_value=True) + sent = tmp_path / "sent" + mocker.patch.object( + telemetry, + "_SEND_EVENT", + f"open({str(sent)!r}, 'w').write(str(os.getppid()))\n", + ) + popen = mocker.spy(telemetry.subprocess, "Popen") + + telemetry._send_detached("run-prod") + + assert popen.spy_return.args[-1] == "attached" + assert popen.spy_return.returncode == 0 + # The sender ran in the direct child, which was reaped before returning. + assert sent.read_text() == str(os.getpid()) + + +def test_reaps_orphans_as_pid_1(mocker: MockerFixture): + """PID 1 inherits every orphan.""" + mocker.patch.object(telemetry.os, "getpid", return_value=1) + + assert telemetry._reaps_orphans() + + +def test_reaps_orphans_is_false_for_an_ordinary_process(): + """The test runner is neither PID 1 nor a child subreaper.""" + assert os.getpid() != 1 + assert not telemetry._reaps_orphans() + + def test_shutdown_executor_without_executor_is_a_noop(): """Nothing to drain when no telemetry was ever sent.""" assert telemetry._shutdown_executor() From aebe6ec55314c7014e53d98a01221a82b8fadc6e Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 06:16:43 +0000 Subject: [PATCH 13/18] Keep the run event's sender attached under a PID 1 parent `reflex run --json` re-runs itself below an output supervisor; as a container's PID 1 (e.g. `CMD exec reflex run --json`) that supervisor inherits the double-forked sender but only waits for its own child, so the sender stayed a zombie for the server's lifetime. Orphans always go to PID 1 (absent a subreaper), so whenever PID 1 is our own parent the sender may linger: _reaps_orphans() now also returns True for os.getppid() == 1, which makes the server send attached and reap the child itself. This covers reflex's output supervisor and any other non-reaping PID 1 parent; a false positive (e.g. a service under systemd) only costs a bounded wait in the supervisor. The ordinary-process test no longer depends on the runner's PIDs or subreaper flag: it fakes getpid/getppid and prctl. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/telemetry.py | 12 ++++--- tests/units/test_telemetry.py | 64 +++++++++++++++++++++++++++++++++-- 2 files changed, 68 insertions(+), 8 deletions(-) diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index 1144ef0cbb4..36371e01176 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -587,16 +587,18 @@ def _reset_executor_after_fork() -> None: def _reaps_orphans() -> bool: - """Report whether orphaned descendants are reparented to this process. + """Report whether a detached sender could be left unreaped. True as PID 1 (a container without an init) or as a Linux child - subreaper. Such a process would inherit a detached sender, and a server - supervisor never reaps it. + subreaper, which inherit the sender but never reap it as a server + supervisor, and under a PID 1 parent, which inherits it too: e.g. the + output supervisor of ``reflex run --json`` as a container's PID 1 only + waits for its own child. A false positive only costs a bounded wait. Returns: - Whether this process inherits orphans. + Whether orphans of this process may be left as zombies. """ - if os.getpid() == 1: + if os.getpid() == 1 or os.getppid() == 1: return True if sys.platform != "linux": return False diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 1f842bfad3c..e4dcd2e7ef9 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -1140,12 +1140,70 @@ def test_reaps_orphans_as_pid_1(mocker: MockerFixture): assert telemetry._reaps_orphans() -def test_reaps_orphans_is_false_for_an_ordinary_process(): - """The test runner is neither PID 1 nor a child subreaper.""" - assert os.getpid() != 1 +def _fake_prctl(mocker: MockerFixture, subreaper: int): + """Make PR_GET_CHILD_SUBREAPER report the given flag. + + Args: + mocker: The mocker fixture. + subreaper: The flag value prctl writes back. + """ + import ctypes + + def prctl(_option, flag, *_args): + flag._obj.value = subreaper + return 0 + + mocker.patch.object(ctypes, "CDLL", return_value=SimpleNamespace(prctl=prctl)) + + +def test_reaps_orphans_under_a_pid_1_parent(mocker: MockerFixture): + """Orphans go to PID 1, which may not reap them when it is our parent. + + E.g. ``reflex run --json`` as a container's PID 1: the output supervisor + only waits for its own child, the server. + """ + mocker.patch.object(telemetry.os, "getpid", return_value=7) + mocker.patch.object(telemetry.os, "getppid", return_value=1) + + assert telemetry._reaps_orphans() + + +def test_reaps_orphans_is_false_for_an_ordinary_process(mocker: MockerFixture): + """Neither PID 1, under PID 1, nor a child subreaper.""" + mocker.patch.object(telemetry.os, "getpid", return_value=7) + mocker.patch.object(telemetry.os, "getppid", return_value=6) + _fake_prctl(mocker, 0) + assert not telemetry._reaps_orphans() +@pytest.mark.skipif(sys.platform != "linux", reason="Linux-only prctl") +def test_reaps_orphans_as_a_child_subreaper(mocker: MockerFixture): + """A child subreaper inherits its descendants' orphans.""" + mocker.patch.object(telemetry.os, "getpid", return_value=7) + mocker.patch.object(telemetry.os, "getppid", return_value=6) + _fake_prctl(mocker, 1) + + assert telemetry._reaps_orphans() + + +def test_send_detached_stays_attached_under_a_pid_1_parent( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """Under a PID 1 parent the sender is not orphaned to that parent.""" + monkeypatch.setattr(telemetry, "_paused", True) + mocker.patch.object( + telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) + ) + mocker.patch.object(telemetry.os, "getpid", return_value=7) + mocker.patch.object(telemetry.os, "getppid", return_value=1) + popen = mocker.patch.object(telemetry.subprocess, "Popen") + + telemetry._send_detached("run-prod") + + assert popen.call_args.args[0][-1] == "attached" + + def test_shutdown_executor_without_executor_is_a_noop(): """Nothing to drain when no telemetry was ever sent.""" assert telemetry._shutdown_executor() From 437b1c77cdf1dc75b58415d426ab2ae928cbe742 Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 06:28:36 +0000 Subject: [PATCH 14/18] Only stay attached under reflex's own output supervisor as PID 1 aebe6ec55 sent attached under any PID 1 parent, which includes a working init such as systemd, where the detached sender is reaped fine and the attached send only made the supervisor wait. The PID 1 parent case now requires the REFLEX_OUTPUT_SUPERVISED marker, i.e. the parent is the `reflex run --json` output supervisor, which only waits for its own child. PID 1 itself and child subreapers are unchanged. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- reflex/utils/telemetry.py | 9 +++--- tests/units/test_telemetry.py | 52 ++++++++++++++++++++++------------- 2 files changed, 38 insertions(+), 23 deletions(-) diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index 36371e01176..461a103c806 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -23,6 +23,7 @@ from reflex_base.config import get_config from reflex_base.environment import environment from reflex_base.registry import RegistrationContext +from reflex_base.utils import log from reflex_base.utils.decorator import once, once_unless_none from reflex_base.utils.exceptions import ReflexError from typing_extensions import NotRequired @@ -591,14 +592,14 @@ def _reaps_orphans() -> bool: True as PID 1 (a container without an init) or as a Linux child subreaper, which inherit the sender but never reap it as a server - supervisor, and under a PID 1 parent, which inherits it too: e.g. the - output supervisor of ``reflex run --json`` as a container's PID 1 only - waits for its own child. A false positive only costs a bounded wait. + supervisor, and under reflex's own ``reflex run --json`` output + supervisor running as PID 1, which inherits it but only waits for its own + child. Any other PID 1 parent is taken to be an init that reaps orphans. Returns: Whether orphans of this process may be left as zombies. """ - if os.getpid() == 1 or os.getppid() == 1: + if os.getpid() == 1 or (os.getppid() == 1 and log.is_output_supervised()): return True if sys.platform != "linux": return False diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index e4dcd2e7ef9..184e9de7496 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -1156,30 +1156,43 @@ def prctl(_option, flag, *_args): mocker.patch.object(ctypes, "CDLL", return_value=SimpleNamespace(prctl=prctl)) -def test_reaps_orphans_under_a_pid_1_parent(mocker: MockerFixture): - """Orphans go to PID 1, which may not reap them when it is our parent. - - E.g. ``reflex run --json`` as a container's PID 1: the output supervisor - only waits for its own child, the server. - """ - mocker.patch.object(telemetry.os, "getpid", return_value=7) - mocker.patch.object(telemetry.os, "getppid", return_value=1) - - assert telemetry._reaps_orphans() - - -def test_reaps_orphans_is_false_for_an_ordinary_process(mocker: MockerFixture): - """Neither PID 1, under PID 1, nor a child subreaper.""" +@pytest.mark.parametrize( + ("supervised", "ppid", "expected"), + [ + # `reflex run --json` as a container's PID 1: the output supervisor + # inherits the sender but only waits for its own child, the server. + (True, 1, True), + # A real init (e.g. systemd) as the parent reaps orphans. + (False, 1, False), + (True, 6, False), + (False, 6, False), + ], +) +def test_reaps_orphans_under_reflex_output_supervisor_as_pid_1( + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, + supervised: bool, + ppid: int, + expected: bool, +): + """Only reflex's own output supervisor as PID 1 leaves orphans unreaped.""" + if supervised: + monkeypatch.setenv("REFLEX_OUTPUT_SUPERVISED", "1") + else: + monkeypatch.delenv("REFLEX_OUTPUT_SUPERVISED", raising=False) mocker.patch.object(telemetry.os, "getpid", return_value=7) - mocker.patch.object(telemetry.os, "getppid", return_value=6) + mocker.patch.object(telemetry.os, "getppid", return_value=ppid) _fake_prctl(mocker, 0) - assert not telemetry._reaps_orphans() + assert telemetry._reaps_orphans() is expected @pytest.mark.skipif(sys.platform != "linux", reason="Linux-only prctl") -def test_reaps_orphans_as_a_child_subreaper(mocker: MockerFixture): +def test_reaps_orphans_as_a_child_subreaper( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): """A child subreaper inherits its descendants' orphans.""" + monkeypatch.delenv("REFLEX_OUTPUT_SUPERVISED", raising=False) mocker.patch.object(telemetry.os, "getpid", return_value=7) mocker.patch.object(telemetry.os, "getppid", return_value=6) _fake_prctl(mocker, 1) @@ -1187,11 +1200,12 @@ def test_reaps_orphans_as_a_child_subreaper(mocker: MockerFixture): assert telemetry._reaps_orphans() -def test_send_detached_stays_attached_under_a_pid_1_parent( +def test_send_detached_stays_attached_under_a_pid_1_output_supervisor( mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch ): - """Under a PID 1 parent the sender is not orphaned to that parent.""" + """Under reflex's output supervisor as PID 1 the sender stays attached.""" monkeypatch.setattr(telemetry, "_paused", True) + monkeypatch.setenv("REFLEX_OUTPUT_SUPERVISED", "1") mocker.patch.object( telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) ) From 1e45ee16322366a0f4c64d1133d3cdefb5b28698 Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 06:34:44 +0000 Subject: [PATCH 15/18] Skip the attached-sender test where os.fork is missing On Windows _send_detached sends in-process and never calls Popen, so the test's popen.call_args lookup raised. It now carries the same os.fork guard as the other detached-sender tests; running the module with os.fork deleted shows every test passing or skipping. Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- tests/units/test_telemetry.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 184e9de7496..a6f8d1c5dc9 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -1200,6 +1200,7 @@ def test_reaps_orphans_as_a_child_subreaper( assert telemetry._reaps_orphans() +@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") def test_send_detached_stays_attached_under_a_pid_1_output_supervisor( mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch ): From 048718e5d2c85d0895538cbe28b5d4aa342efb72 Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 1 Oct 2026 17:08:10 +0000 Subject: [PATCH 16/18] Shorten the prod fork preload news fragment to two sentences Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca --- news/+prod-fork-preload.performance.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/news/+prod-fork-preload.performance.md b/news/+prod-fork-preload.performance.md index 28aa7e7d14f..f0bfcd0171f 100644 --- a/news/+prod-fork-preload.performance.md +++ b/news/+prod-fork-preload.performance.md @@ -1 +1 @@ -Production backend workers on Linux are now forked from a supervisor that has already imported the app, so framework and app code are shared copy-on-write instead of re-imported per worker. A 4-worker blank app drops from about 720 MB to about 220 MB of proportional set size and starts faster. Set `REFLEX_BACKEND_START_METHOD=spawn` for apps that are not fork-safe. +Production backend workers on Linux are now forked from a supervisor that has already imported the app, so they share its memory instead of each importing the app again, which cuts memory use and startup time. Set `REFLEX_BACKEND_START_METHOD=spawn` for apps that are not fork-safe. From 1f07f003741fe104533b1fff37415e38c13c8473 Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 13 Oct 2026 21:16:30 +0000 Subject: [PATCH 17/18] Send the run-prod telemetry event synchronously instead of via on_started Address masenf's review: the supervisor sends the run-prod event on its own thread before granian starts, so no telemetry thread exists at the fork and nothing needs to run after the workers start. This removes the on_started callback threaded through run_backend_prod, run_frontend_prod, run_uvicorn_backend_prod and run_granian_backend_prod, the NotifyingGranian subclass, telemetry._send_detached with its detached sender script, _reaps_orphans and the _paused flag. The pre-fork drain (_shutdown_executor) and the at-fork executor reset stay: the compile event is still queued on the thread pool, and a forked child must not inherit a pool whose worker thread it does not own. Claude-Session: https://claude.ai/code/session_01GbqrTagh776xkQvsopdytF --- reflex/reflex.py | 17 +- reflex/utils/exec.py | 61 ++----- reflex/utils/telemetry.py | 131 ++------------- tests/units/test_reflex.py | 25 ++- tests/units/test_telemetry.py | 285 --------------------------------- tests/units/utils/test_exec.py | 69 +------- 6 files changed, 39 insertions(+), 549 deletions(-) diff --git a/reflex/reflex.py b/reflex/reflex.py index 8793f711828..b945305df29 100644 --- a/reflex/reflex.py +++ b/reflex/reflex.py @@ -565,12 +565,9 @@ def _run_prod(running_mode: constants.RunningMode, port: int, host: str): _skip_compile() - # Post the telemetry event once the workers are running, from a separate - # process: the server may fork its supervisor again to respawn workers, so - # it must never hold a telemetry thread. - def on_started(): - """Send the run telemetry without a thread in the supervisor.""" - telemetry._send_detached("run-prod") + # Post a telemetry event. Sent synchronously: on Linux the backend forks its + # workers from this process, which must hold no telemetry thread by then. + telemetry._send("run-prod", config.telemetry_enabled) # Display custom message when there is a keyboard interrupt. atexit.register(processes.atexit_handler) @@ -582,14 +579,10 @@ def on_started(): ) if running_mode.has_backend(): exec.run_backend_prod( - host, - port, - config.loglevel.subprocess_level(), - running_mode.has_frontend(), - on_started=on_started, + host, port, config.loglevel.subprocess_level(), running_mode.has_frontend() ) else: - exec.run_frontend_prod(host, port, on_started=on_started) + exec.run_frontend_prod(host, port) def _run( diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index da6b9f2e2c8..cea18a0ff11 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -463,9 +463,7 @@ def _frontend_prod_app(): return Starlette(routes=[get_frontend_mount()]) -def run_frontend_prod( - host: str, port: int, on_started: Callable[[], Any] | None = None -): +def run_frontend_prod(host: str, port: int): """Run the frontend in production mode by serving compiled static files. Uses the same granian/uvicorn infrastructure as the backend. @@ -473,25 +471,16 @@ def run_frontend_prod( Args: host: The host to serve on. port: The port to serve on. - on_started: Called once the server's workers have been started. """ loglevel = get_config().loglevel.subprocess_level() if should_use_granian(): run_granian_backend_prod( - host, - port, - loglevel, - app_target=f"{__name__}:_frontend_prod_app", - on_started=on_started, + host, port, loglevel, app_target=f"{__name__}:_frontend_prod_app" ) else: run_uvicorn_backend_prod( - host, - port, - loglevel, - app_target=f"{__name__}:_frontend_prod_app", - on_started=on_started, + host, port, loglevel, app_target=f"{__name__}:_frontend_prod_app" ) @@ -954,7 +943,6 @@ def run_backend_prod( port: int, loglevel: constants.LogLevel = constants.LogLevel.ERROR, mount_frontend_compiled_app: bool = False, - on_started: Callable[[], Any] | None = None, ): """Run the backend. @@ -963,14 +951,13 @@ def run_backend_prod( port: The app port loglevel: The log level. mount_frontend_compiled_app: Whether to mount the compiled frontend app with the backend. - on_started: Called once the server's workers have been started. """ environment.REFLEX_MOUNT_FRONTEND_COMPILED_APP.set(mount_frontend_compiled_app) if should_use_granian(): - run_granian_backend_prod(host, port, loglevel, on_started=on_started) + run_granian_backend_prod(host, port, loglevel) else: - run_uvicorn_backend_prod(host, port, loglevel, on_started=on_started) + run_uvicorn_backend_prod(host, port, loglevel) def _get_backend_workers(): @@ -980,11 +967,7 @@ def _get_backend_workers(): def run_uvicorn_backend_prod( - host: str, - port: int, - loglevel: LogLevel, - app_target: str | None = None, - on_started: Callable[[], Any] | None = None, + host: str, port: int, loglevel: LogLevel, app_target: str | None = None ): """Run the backend in production mode using Uvicorn. @@ -993,8 +976,6 @@ def run_uvicorn_backend_prod( port: The app port loglevel: The log level. app_target: The ASGI app target to run. Defaults to the reflex app instance. - on_started: Called right before the server process is launched; it - runs in a separate interpreter, so nothing here is forked. """ import os import shlex @@ -1037,9 +1018,6 @@ def run_uvicorn_backend_prod( *("--log-level", loglevel.value), ] - if on_started is not None: - on_started() - processes.new_process( command, run=True, @@ -1088,6 +1066,8 @@ def _preload_for_fork(app_target: str | None) -> bool: prerequisites.get_app() serializers._prepare_serializers_for_fork() # Forking while the telemetry thread may hold a lock can deadlock the child. + # Nothing in the supervisor sends telemetry after this point, so no thread + # exists at any later fork either (e.g. a worker respawn). if not telemetry._shutdown_executor(): return False # Freezing keeps worker GC passes from writing to the preloaded objects' @@ -1098,11 +1078,7 @@ def _preload_for_fork(app_target: str | None) -> bool: def run_granian_backend_prod( - host: str, - port: int, - loglevel: LogLevel, - app_target: str | None = None, - on_started: Callable[[], Any] | None = None, + host: str, port: int, loglevel: LogLevel, app_target: str | None = None ): """Run the backend in production mode using Granian. @@ -1111,9 +1087,6 @@ def run_granian_backend_prod( port: The app port loglevel: The log level. app_target: The ASGI app target to run. Defaults to the reflex app instance. - on_started: Called in the supervisor once the workers have been started. - It must not start threads: granian forks the supervisor again to - respawn workers. """ import multiprocessing @@ -1131,21 +1104,7 @@ def run_granian_backend_prod( logger.debug("A telemetry send is still running; spawning workers.") multiprocessing.set_start_method("spawn", force=True) - class NotifyingGranian(Granian): # pyright: ignore[reportGeneralTypeIssues] - """Granian server that reports when its workers have been started.""" - - def startup(self, *args, **kwargs): - """Start the supervisor and its workers, then notify the caller. - - Args: - args: Positional arguments for the Granian startup. - kwargs: Keyword arguments for the Granian startup. - """ - super().startup(*args, **kwargs) - if on_started is not None: - on_started() - - granian_app = NotifyingGranian( + granian_app = Granian( target=app_target or get_app_instance_from_file(), factory=True, address=host, diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index d655fe9fec7..37212dcb0e5 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -7,7 +7,6 @@ import multiprocessing import os import platform -import subprocess import sys import threading import urllib.request @@ -23,7 +22,6 @@ from reflex_base.config import get_config from reflex_base.environment import environment from reflex_base.registry import RegistrationContext -from reflex_base.utils import log from reflex_base.utils.decorator import once, once_unless_none from reflex_base.utils.exceptions import ReflexError @@ -478,9 +476,6 @@ def _send( _executor_lock = threading.Lock() _executor: ThreadPoolExecutor | None = None -# Set by _shutdown_executor(): a process that is about to fork, and may fork -# again later, never starts another pool (and thread). Forked children reset it. -_paused = False def _get_telemetry_executor() -> ThreadPoolExecutor: @@ -495,16 +490,10 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: Returns: The shared single-worker telemetry executor. - - Raises: - RuntimeError: When the pool is stopped for a pending fork. """ global _executor if _executor is None: with _executor_lock: - if _paused: - msg = "Telemetry is paused for a pending fork." - raise RuntimeError(msg) if _executor is None: _executor = ThreadPoolExecutor( max_workers=1, thread_name_prefix="reflex-telemetry" @@ -513,13 +502,12 @@ def _get_telemetry_executor() -> ThreadPoolExecutor: def _shutdown_executor(timeout: float = 2) -> bool: - """Deliver queued telemetry and stop the worker thread for good. + """Deliver queued telemetry and stop the worker thread. - Called by a supervisor before it forks workers, so no telemetry thread is - alive at that fork or any later one (e.g. a worker respawn). From then on - in-process sends are dropped; use ``_send_detached`` instead. A send that - stalls past the timeout (e.g. on a DNS lookup) is abandoned so it cannot - hold up startup, and the caller must then not fork. + Called by a supervisor right before it forks workers, so no telemetry + thread is alive at the fork. A send that stalls past the timeout (e.g. on + a DNS lookup) is abandoned so it cannot hold up startup, and the caller + must then not fork. Args: timeout: Maximum number of seconds to wait for queued telemetry. @@ -527,11 +515,8 @@ def _shutdown_executor(timeout: float = 2) -> bool: Returns: Whether the worker thread has stopped, i.e. whether forking is safe. """ - global _executor, _paused - with _executor_lock: - _paused = True - executor = _executor - if executor is None: + global _executor + if (executor := _executor) is None: return True drained = _flush(timeout) if not drained: @@ -539,117 +524,23 @@ def _shutdown_executor(timeout: float = 2) -> bool: # Stay published while stopping: a racing send() then fails to submit (and # is suppressed) instead of starting a second pool before the fork. executor.shutdown(wait=drained, cancel_futures=True) - stopped = not any(thread.is_alive() for thread in executor._threads) with _executor_lock: if _executor is executor: _executor = None - if not stopped: - # The caller will not fork with a live thread, so sends may resume. - _paused = False - return stopped + return not any(thread.is_alive() for thread in executor._threads) def _reset_executor_after_fork() -> None: """Drop the inherited executor and lock; the child owns neither's thread.""" - global _executor, _executor_lock, _paused + global _executor, _executor_lock _executor_lock = threading.Lock() _executor = None - _paused = False if hasattr(os, "register_at_fork"): os.register_at_fork(after_in_child=_reset_executor_after_fork) -# Run by _send_detached in a fresh interpreter (argv: event, mode). In -# "detach" mode it forks once more and exits, so the supervisor reaps it at -# once and the sender is reparented to whoever reaps orphans. -_DETACH = """\ -import os -import sys - -if sys.argv[2] == "detach": - if os.fork(): - os._exit(0) - os.setsid() -""" -_SEND_EVENT = """\ -from reflex.utils import telemetry - -telemetry._process_event(sys.argv[1], True) -telemetry._flush(30) -os._exit(0) -""" -# Seconds the supervisor waits for its direct child before killing it. -_DETACHED_TIMEOUT = 10 -_PR_GET_CHILD_SUBREAPER = 37 - - -def _reaps_orphans() -> bool: - """Report whether a detached sender could be left unreaped. - - True as PID 1 (a container without an init) or as a Linux child - subreaper, which inherit the sender but never reap it as a server - supervisor, and under reflex's own ``reflex run --json`` output - supervisor running as PID 1, which inherits it but only waits for its own - child. Any other PID 1 parent is taken to be an init that reaps orphans. - - Returns: - Whether orphans of this process may be left as zombies. - """ - if os.getpid() == 1 or (os.getppid() == 1 and log.is_output_supervised()): - return True - if sys.platform != "linux": - return False - import ctypes - - flag = ctypes.c_int() - with suppress(Exception): - prctl = ctypes.CDLL(None, use_errno=True).prctl - if prctl(_PR_GET_CHILD_SUBREAPER, ctypes.byref(flag), 0, 0, 0) == 0: - return bool(flag.value) - return False - - -def _send_detached(event: str) -> None: - """Send an event without a telemetry thread in a process that forks. - - A supervisor that paused telemetry to fork workers safely hands the event - to a fresh interpreter, so it never gains a thread or waits on the - network. The direct child exits right after detaching the sender, so no - zombie is left, and ``close_fds`` keeps the server's sockets out of it. - Where orphans would come back to this process (PID 1, a subreaper), the - direct child sends itself and is waited for instead. A child that - outlives ``_DETACHED_TIMEOUT`` is killed and reaped. Any other process - sends in-process without waiting on a child. - - Args: - event: The event name. - """ - with _executor_lock: - paused = _paused - if not paused or not hasattr(os, "fork"): - # Nothing forks this process with a telemetry thread alive. - send(event) - return - with suppress(Exception): - if not get_config().telemetry_enabled: - return - mode = "attached" if _reaps_orphans() else "detach" - child = subprocess.Popen( - [sys.executable, "-c", _DETACH + _SEND_EVENT, event, mode], - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - close_fds=True, - ) - try: - child.wait(timeout=_DETACHED_TIMEOUT) - except subprocess.TimeoutExpired: - child.kill() - child.wait() - - def _current_registration_context() -> RegistrationContext | None: """Return the caller's RegistrationContext, or None if none is attached. @@ -720,12 +611,10 @@ def _submit(fn: Callable[..., Any], /, *args, **kwargs) -> None: kwargs: Keyword arguments forwarded to ``fn``. """ registration_context = _current_registration_context() - try: + with suppress(Exception): _get_telemetry_executor().submit( _run_suppressed, registration_context, fn, *args, **kwargs ) - except Exception as err: - logger.debug(f"Dropped telemetry event: {err}") def _flush(timeout: float | None = None) -> bool: diff --git a/tests/units/test_reflex.py b/tests/units/test_reflex.py index 25049959dcf..cd280ada792 100644 --- a/tests/units/test_reflex.py +++ b/tests/units/test_reflex.py @@ -180,18 +180,17 @@ def test_compile_app_worker_flushes_telemetry_on_failure(mocker): (constants.RunningMode.FRONTEND_ONLY, "run_frontend_prod"), ], ) -def test_run_prod_sends_telemetry_once_the_server_started( +def test_run_prod_sends_telemetry_synchronously( mocker, running_mode: constants.RunningMode, launcher: str ): - """The run-prod event waits for the server and leaves the supervisor. + """The run-prod event is delivered on the calling thread, before the server. - Sending it from another process keeps the supervisor, which granian may - fork again to respawn workers, free of telemetry threads. + The server forks its workers from this process, so the event must not be + queued on the telemetry thread pool. """ from reflex.utils import build, exec, processes, telemetry - send = mocker.patch.object(telemetry, "_send_detached") - mocker.patch.object(reflex, "get_config") + config = mocker.patch.object(reflex, "get_config").return_value mocker.patch.object(reflex, "_compile_app") mocker.patch.object(reflex, "_skip_compile") mocker.patch.object(build, "setup_frontend_prod") @@ -199,16 +198,16 @@ def test_run_prod_sends_telemetry_once_the_server_started( mocker.patch("atexit.register") mocker.patch.object(exec, "notify_app_running") mocker.patch.object(exec, "notify_frontend") - - def serve(*_args, on_started): - send.assert_not_called() - on_started() - - mocker.patch.object(exec, launcher, side_effect=serve) + queued = mocker.patch.object(telemetry, "send") + calls = mocker.Mock() + calls.attach_mock(mocker.patch.object(telemetry, "_send"), "send") + calls.attach_mock(mocker.patch.object(exec, launcher), "serve") reflex._run_prod(running_mode, 8000, "127.0.0.1") - send.assert_called_once_with("run-prod") + queued.assert_not_called() + calls.send.assert_called_once_with("run-prod", config.telemetry_enabled) + assert [name for name, _args, _kwargs in calls.mock_calls] == ["send", "serve"] def test_cloud_commands_registered(): diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index a6f8d1c5dc9..18b38d894fc 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -7,7 +7,6 @@ import time import uuid import warnings -from pathlib import Path from types import SimpleNamespace import pytest @@ -34,8 +33,6 @@ def _drain_telemetry_executor(): """ yield telemetry._flush() - # A test that stops the pool for a fork must not silence the next test. - telemetry._paused = False @pytest.fixture @@ -937,288 +934,6 @@ def stall(): assert queued.cancelled() -def test_send_after_pre_fork_shutdown_starts_no_thread(mocker: MockerFixture): - """Until the pending fork happens, a send cannot start a telemetry thread.""" - job = mocker.Mock() - telemetry._get_telemetry_executor() - assert telemetry._shutdown_executor() - - telemetry._submit(job) - - assert telemetry._executor is None - job.assert_not_called() - - -def test_stalled_shutdown_lifts_the_pause(): - """A caller that will not fork may send in-process again.""" - release = threading.Event() - running = threading.Event() - executor = telemetry._get_telemetry_executor() - workers = list(executor._threads) - executor.submit(lambda: running.set() or release.wait(10)) - assert running.wait(5) - try: - assert not telemetry._shutdown_executor(timeout=0.05) - finally: - release.set() - for worker in workers: - worker.join(5) - - assert not telemetry._paused - - -@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") -def test_real_fork_keeps_the_parent_paused(): - """A supervisor that may fork again never sends in-process; its child may.""" - assert telemetry._shutdown_executor() - with warnings.catch_warnings(): - # Other tests leave unrelated threads behind in this process. - warnings.simplefilter("ignore", DeprecationWarning) - pid = os.fork() - if pid == 0: - code = 1 - try: - code = int(telemetry._paused) - finally: - os._exit(code) - _, status = os.waitpid(pid, 0) - - assert os.waitstatus_to_exitcode(status) == 0 - assert telemetry._paused - with pytest.raises(RuntimeError, match="paused"): - telemetry._get_telemetry_executor() - - -@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") -def test_send_detached_hands_the_event_to_a_detached_process( - mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch -): - """A supervisor paused to fork sends through a short-lived process.""" - monkeypatch.setattr(telemetry, "_paused", True) - mocker.patch.object( - telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) - ) - popen = mocker.patch.object(telemetry.subprocess, "Popen") - mocker.patch.object(telemetry, "_reaps_orphans", return_value=False) - - telemetry._send_detached("run-prod") - - popen.assert_called_once_with( - [ - sys.executable, - "-c", - telemetry._DETACH + telemetry._SEND_EVENT, - "run-prod", - "detach", - ], - stdin=telemetry.subprocess.DEVNULL, - stdout=telemetry.subprocess.DEVNULL, - stderr=telemetry.subprocess.DEVNULL, - close_fds=True, - ) - # The direct child exits right after detaching, so it is reaped at once. - popen.return_value.wait.assert_called_once() - assert telemetry._executor is None - - -def test_send_detached_sends_in_process_when_not_paused(mocker: MockerFixture): - """A process that does not fork (uvicorn, spawned granian) does not wait - on a child process: it sends in-process, as before. - """ - send = mocker.patch.object(telemetry, "send") - popen = mocker.patch.object(telemetry.subprocess, "Popen") - - telemetry._send_detached("run-prod") - - send.assert_called_once_with("run-prod") - popen.assert_not_called() - - -def test_send_detached_sends_in_process_without_fork( - mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch -): - """Where nothing forks (Windows) the event is sent in-process.""" - monkeypatch.delattr(os, "fork", raising=False) - send = mocker.patch.object(telemetry, "send") - popen = mocker.patch.object(telemetry.subprocess, "Popen") - - telemetry._send_detached("run-prod") - - send.assert_called_once_with("run-prod") - popen.assert_not_called() - - -def test_send_detached_starts_nothing_when_disabled( - mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch -): - """With telemetry disabled no process is started.""" - monkeypatch.setattr(telemetry, "_paused", True) - mocker.patch.object( - telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=False) - ) - popen = mocker.patch.object(telemetry.subprocess, "Popen") - - telemetry._send_detached("run-prod") - - popen.assert_not_called() - - -@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") -def test_detached_send_leaves_no_child_behind( - mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch -): - """The direct child exits after detaching, so no zombie is left to reap.""" - monkeypatch.setattr(telemetry, "_paused", True) - mocker.patch.object( - telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) - ) - mocker.patch.object(telemetry, "_reaps_orphans", return_value=False) - # The real detaching prologue, but the sender exits instead of sending. - mocker.patch.object(telemetry, "_SEND_EVENT", "os._exit(0)\n") - popen = mocker.spy(telemetry.subprocess, "Popen") - - telemetry._send_detached("run-prod") - - assert popen.spy_return.returncode == 0 - - -@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") -def test_detached_send_kills_and_reaps_a_child_that_hangs( - mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch -): - """A child that does not exit in time is killed and reaped, not leaked.""" - monkeypatch.setattr(telemetry, "_paused", True) - monkeypatch.setattr(telemetry, "_DETACHED_TIMEOUT", 0.2) - mocker.patch.object( - telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) - ) - mocker.patch.object(telemetry, "_reaps_orphans", return_value=True) - mocker.patch.object(telemetry, "_SEND_EVENT", "import time\ntime.sleep(30)\n") - popen = mocker.spy(telemetry.subprocess, "Popen") - - telemetry._send_detached("run-prod") - - assert popen.spy_return.returncode == -9 - - -@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") -def test_detached_send_stays_attached_where_orphans_come_back( - mocker: MockerFixture, - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -): - """As PID 1 or a subreaper the sender is not orphaned to this process. - - Orphans would be reparented to the supervisor, which never reaps them, - so the single child sends and is waited for instead. - """ - monkeypatch.setattr(telemetry, "_paused", True) - mocker.patch.object( - telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) - ) - mocker.patch.object(telemetry, "_reaps_orphans", return_value=True) - sent = tmp_path / "sent" - mocker.patch.object( - telemetry, - "_SEND_EVENT", - f"open({str(sent)!r}, 'w').write(str(os.getppid()))\n", - ) - popen = mocker.spy(telemetry.subprocess, "Popen") - - telemetry._send_detached("run-prod") - - assert popen.spy_return.args[-1] == "attached" - assert popen.spy_return.returncode == 0 - # The sender ran in the direct child, which was reaped before returning. - assert sent.read_text() == str(os.getpid()) - - -def test_reaps_orphans_as_pid_1(mocker: MockerFixture): - """PID 1 inherits every orphan.""" - mocker.patch.object(telemetry.os, "getpid", return_value=1) - - assert telemetry._reaps_orphans() - - -def _fake_prctl(mocker: MockerFixture, subreaper: int): - """Make PR_GET_CHILD_SUBREAPER report the given flag. - - Args: - mocker: The mocker fixture. - subreaper: The flag value prctl writes back. - """ - import ctypes - - def prctl(_option, flag, *_args): - flag._obj.value = subreaper - return 0 - - mocker.patch.object(ctypes, "CDLL", return_value=SimpleNamespace(prctl=prctl)) - - -@pytest.mark.parametrize( - ("supervised", "ppid", "expected"), - [ - # `reflex run --json` as a container's PID 1: the output supervisor - # inherits the sender but only waits for its own child, the server. - (True, 1, True), - # A real init (e.g. systemd) as the parent reaps orphans. - (False, 1, False), - (True, 6, False), - (False, 6, False), - ], -) -def test_reaps_orphans_under_reflex_output_supervisor_as_pid_1( - mocker: MockerFixture, - monkeypatch: pytest.MonkeyPatch, - supervised: bool, - ppid: int, - expected: bool, -): - """Only reflex's own output supervisor as PID 1 leaves orphans unreaped.""" - if supervised: - monkeypatch.setenv("REFLEX_OUTPUT_SUPERVISED", "1") - else: - monkeypatch.delenv("REFLEX_OUTPUT_SUPERVISED", raising=False) - mocker.patch.object(telemetry.os, "getpid", return_value=7) - mocker.patch.object(telemetry.os, "getppid", return_value=ppid) - _fake_prctl(mocker, 0) - - assert telemetry._reaps_orphans() is expected - - -@pytest.mark.skipif(sys.platform != "linux", reason="Linux-only prctl") -def test_reaps_orphans_as_a_child_subreaper( - mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch -): - """A child subreaper inherits its descendants' orphans.""" - monkeypatch.delenv("REFLEX_OUTPUT_SUPERVISED", raising=False) - mocker.patch.object(telemetry.os, "getpid", return_value=7) - mocker.patch.object(telemetry.os, "getppid", return_value=6) - _fake_prctl(mocker, 1) - - assert telemetry._reaps_orphans() - - -@pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork is POSIX-only") -def test_send_detached_stays_attached_under_a_pid_1_output_supervisor( - mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch -): - """Under reflex's output supervisor as PID 1 the sender stays attached.""" - monkeypatch.setattr(telemetry, "_paused", True) - monkeypatch.setenv("REFLEX_OUTPUT_SUPERVISED", "1") - mocker.patch.object( - telemetry, "get_config", return_value=SimpleNamespace(telemetry_enabled=True) - ) - mocker.patch.object(telemetry.os, "getpid", return_value=7) - mocker.patch.object(telemetry.os, "getppid", return_value=1) - popen = mocker.patch.object(telemetry.subprocess, "Popen") - - telemetry._send_detached("run-prod") - - assert popen.call_args.args[0][-1] == "attached" - - def test_shutdown_executor_without_executor_is_a_noop(): """Nothing to drain when no telemetry was ever sent.""" assert telemetry._shutdown_executor() diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index 49f934282e9..2bc85ee2098 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -445,19 +445,13 @@ def serve(self): def test_run_granian_backend_prod_preloads_app_before_forking( mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch ): - """With fork, the app is imported, telemetry drained and the heap frozen first. - - ``on_started`` runs only once the workers have been started. - """ + """With fork, the app is imported, telemetry drained and the heap frozen first.""" monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") calls: list[str] = [] _fake_granian_prod(mocker, calls) exec_utils.run_granian_backend_prod( - host="0.0.0.0", - port=8000, - loglevel=exec_utils.LogLevel.INFO, - on_started=lambda: calls.append("started"), + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO ) assert calls == [ @@ -468,7 +462,6 @@ def test_run_granian_backend_prod_preloads_app_before_forking( "freeze", "serve", "workers", - "started", ] @@ -532,64 +525,6 @@ def test_run_granian_backend_prod_custom_target_only_freezes( assert calls == ["start:fork", "serializers", "drain", "freeze", "serve", "workers"] -@pytest.mark.parametrize("use_granian", [True, False]) -def test_run_backend_prod_forwards_on_started(mocker: MockerFixture, use_granian: bool): - """The prod backend hands ``on_started`` to whichever server it runs.""" - mocker.patch.object(exec_utils, "should_use_granian", return_value=use_granian) - launcher = mocker.patch.object( - exec_utils, - "run_granian_backend_prod" if use_granian else "run_uvicorn_backend_prod", - ) - on_started = mocker.Mock() - - exec_utils.run_backend_prod( - "0.0.0.0", 8000, exec_utils.LogLevel.INFO, on_started=on_started - ) - - launcher.assert_called_once_with( - "0.0.0.0", 8000, exec_utils.LogLevel.INFO, on_started=on_started - ) - - -@pytest.mark.parametrize("use_granian", [True, False]) -def test_run_frontend_prod_forwards_on_started( - mocker: MockerFixture, use_granian: bool -): - """The prod frontend server hands ``on_started`` to its launcher.""" - mocker.patch.object(exec_utils, "should_use_granian", return_value=use_granian) - launcher = mocker.patch.object( - exec_utils, - "run_granian_backend_prod" if use_granian else "run_uvicorn_backend_prod", - ) - on_started = mocker.Mock() - - exec_utils.run_frontend_prod("0.0.0.0", 3000, on_started=on_started) - - assert launcher.call_args.kwargs["on_started"] is on_started - - -def test_run_uvicorn_backend_prod_calls_on_started_before_launch( - mocker: MockerFixture, -): - """The server runs in a separate interpreter, so the callback runs first.""" - calls: list[str] = [] - mocker.patch.object(exec_utils, "_get_backend_workers", return_value=1) - mocker.patch( - "reflex.utils.processes.new_process", - side_effect=lambda *_args, **_kwargs: calls.append("launch"), - ) - - exec_utils.run_uvicorn_backend_prod( - "0.0.0.0", - 8000, - exec_utils.LogLevel.INFO, - app_target="app:app", - on_started=lambda: calls.append("started"), - ) - - assert calls == ["started", "launch"] - - @pytest.mark.parametrize( ("default_method", "expected"), [("fork", "fork"), ("forkserver", "fork"), ("spawn", None)], From d6cc137d7c1203fbd3825f924e9427e1535a7ff7 Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 13 Oct 2026 21:35:45 +0000 Subject: [PATCH 18/18] Keep the alias migration on the synchronous run-prod send and floor reflex-base A backend-only production run skips the compile event, so its run-prod event is the only one that can carry the one-time $create_alias migration. Send it through _process_event, and deliver the alias on the calling thread with _send so a process about to fork still starts no telemetry thread. From the worker thread this changes nothing but the order: the alias now precedes the event it rides on instead of being queued behind it. REFLEX_BACKEND_START_METHOD is new in reflex-base, so reflex's floor on it moves to 0.10.1.dev0, the next version after the reflex-base-v0.10.0 tag. Claude-Session: https://claude.ai/code/session_01GbqrTagh776xkQvsopdytF --- pyproject.toml | 2 +- reflex/reflex.py | 2 +- reflex/utils/telemetry.py | 13 ++++++++----- tests/units/test_reflex.py | 6 ++++-- tests/units/test_telemetry.py | 15 ++++++++------- 5 files changed, 22 insertions(+), 16 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e46f43cbd28..71671f49578 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "starlette >=1.3.1", "typing_extensions >=4.13.0", "wrapt >=1.17.0,<2.6", - "reflex-base >= 0.10.0a2.dev0", + "reflex-base >= 0.10.1.dev0", "reflex-components-code >= 0.10.0", "reflex-components-core >= 0.10.0", "reflex-components-dataeditor >= 0.10.0", diff --git a/reflex/reflex.py b/reflex/reflex.py index b945305df29..facbd1fa24d 100644 --- a/reflex/reflex.py +++ b/reflex/reflex.py @@ -567,7 +567,7 @@ def _run_prod(running_mode: constants.RunningMode, port: int, host: str): # Post a telemetry event. Sent synchronously: on Linux the backend forks its # workers from this process, which must hold no telemetry thread by then. - telemetry._send("run-prod", config.telemetry_enabled) + telemetry._process_event("run-prod", config.telemetry_enabled) # Display custom message when there is a keyboard interrupt. atexit.register(processes.atexit_handler) diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index 37212dcb0e5..e669935681c 100644 --- a/reflex/utils/telemetry.py +++ b/reflex/utils/telemetry.py @@ -674,8 +674,7 @@ def _maybe_alias_legacy_distinct_id(telemetry_enabled: bool | None) -> None: # Don't latch: a later enabled send in this process should retry. return - # Latch before the alias send below (which re-enters send()) so it cannot - # recurse and so the attempt happens at most once per process. + # Latch first so the attempt happens at most once per process. _legacy_alias_attempted = True # Resolve the installation id first: a brand-new install is created and @@ -687,8 +686,10 @@ def _maybe_alias_legacy_distinct_id(telemetry_enabled: bool | None) -> None: # distinct_id is the UUID form (set by get_event_defaults); send the # legacy integer as ``alias`` so PostHog coerces it to the same float as - # the historic events and merges the two persons. - send("$create_alias", telemetry_enabled, properties={"alias": installation_id}) + # the historic events and merges the two persons. Delivered on the + # calling thread: this runs from the worker, or from a caller that must + # not start one. + _send("$create_alias", telemetry_enabled, properties={"alias": installation_id}) # Record the new semantics regardless of outcome; we must not retry every # run even if the lossy legacy id failed to match. @@ -708,7 +709,9 @@ def send( pool, so neither the data gathering (blocking syscalls and subprocess calls) nor the synchronous HTTP request blocks the caller — in particular the asyncio event loop serving a running app. Delivery is best-effort: any - failure is suppressed and never surfaces to the caller. + failure is suppressed and never surfaces to the caller. A caller that must + not start a thread (a process about to fork) uses ``_process_event`` + directly instead. Args: event: The event name. diff --git a/tests/units/test_reflex.py b/tests/units/test_reflex.py index cd280ada792..b51580a7717 100644 --- a/tests/units/test_reflex.py +++ b/tests/units/test_reflex.py @@ -186,7 +186,9 @@ def test_run_prod_sends_telemetry_synchronously( """The run-prod event is delivered on the calling thread, before the server. The server forks its workers from this process, so the event must not be - queued on the telemetry thread pool. + queued on the telemetry thread pool. Going through ``_process_event`` keeps + the one-time alias migration, which a backend-only run has no other event + to ride on. """ from reflex.utils import build, exec, processes, telemetry @@ -200,7 +202,7 @@ def test_run_prod_sends_telemetry_synchronously( mocker.patch.object(exec, "notify_frontend") queued = mocker.patch.object(telemetry, "send") calls = mocker.Mock() - calls.attach_mock(mocker.patch.object(telemetry, "_send"), "send") + calls.attach_mock(mocker.patch.object(telemetry, "_process_event"), "send") calls.attach_mock(mocker.patch.object(exec, launcher), "serve") reflex._run_prod(running_mode, 8000, "127.0.0.1") diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 18b38d894fc..db2cc600e3a 100644 --- a/tests/units/test_telemetry.py +++ b/tests/units/test_telemetry.py @@ -552,13 +552,16 @@ def test_maybe_alias_sends_create_alias_for_legacy_install(mocker: MockerFixture mocker.patch.object(telemetry, "ensure_reflex_installation_id", return_value=12345) mocker.patch.object(telemetry, "has_uuid_distinct_id_semantics", return_value=False) mark = mocker.patch.object(telemetry, "mark_uuid_distinct_id_semantics") - send_mock = mocker.patch.object(telemetry, "send") + send_mock = mocker.patch.object(telemetry, "_send") + queued = mocker.patch.object(telemetry, "send") telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True) send_mock.assert_called_once_with( "$create_alias", True, properties={"alias": 12345} ) + # The alias goes out on the calling thread, never through the pool. + queued.assert_not_called() mark.assert_called_once() @@ -567,7 +570,7 @@ def test_maybe_alias_skips_for_uuid_native_install(mocker: MockerFixture): mocker.patch.object(telemetry, "ensure_reflex_installation_id", return_value=12345) mocker.patch.object(telemetry, "has_uuid_distinct_id_semantics", return_value=True) mark = mocker.patch.object(telemetry, "mark_uuid_distinct_id_semantics") - send_mock = mocker.patch.object(telemetry, "send") + send_mock = mocker.patch.object(telemetry, "_send") telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True) @@ -580,7 +583,7 @@ def test_maybe_alias_skips_without_installation_id(mocker: MockerFixture): mocker.patch.object(telemetry, "ensure_reflex_installation_id", return_value=None) has = mocker.patch.object(telemetry, "has_uuid_distinct_id_semantics") mark = mocker.patch.object(telemetry, "mark_uuid_distinct_id_semantics") - send_mock = mocker.patch.object(telemetry, "send") + send_mock = mocker.patch.object(telemetry, "_send") telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True) @@ -592,7 +595,7 @@ def test_maybe_alias_skips_without_installation_id(mocker: MockerFixture): def test_maybe_alias_skips_when_telemetry_disabled(mocker: MockerFixture): """Disabled telemetry does no work and leaves the marker unwritten.""" ensure = mocker.patch.object(telemetry, "ensure_reflex_installation_id") - send_mock = mocker.patch.object(telemetry, "send") + send_mock = mocker.patch.object(telemetry, "_send") telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=False) @@ -605,7 +608,7 @@ def test_maybe_alias_runs_at_most_once_per_process(mocker: MockerFixture): mocker.patch.object(telemetry, "ensure_reflex_installation_id", return_value=7) mocker.patch.object(telemetry, "has_uuid_distinct_id_semantics", return_value=False) mocker.patch.object(telemetry, "mark_uuid_distinct_id_semantics") - send_mock = mocker.patch.object(telemetry, "send") + send_mock = mocker.patch.object(telemetry, "_send") telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True) telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True) @@ -625,8 +628,6 @@ def test_maybe_alias_create_alias_payload( ) telemetry._maybe_alias_legacy_distinct_id(telemetry_enabled=True) - # The $create_alias is now sent on the telemetry worker thread; wait for it. - telemetry._flush() urlopen.assert_called_once() payload = posted_json(urlopen.call_args)