diff --git a/news/7142.bugfix.md b/news/7142.bugfix.md new file mode 100644 index 00000000000..735584fa951 --- /dev/null +++ b/news/7142.bugfix.md @@ -0,0 +1 @@ +Fix backend startup crashes from concurrent or truncated stateful-page marker writes. Markers are replaced atomically, remain readable by separate backend users, and are rebuilt when missing or corrupt; dry-run compilation leaves them unchanged. diff --git a/reflex/app.py b/reflex/app.py index c3d496e0cdc..cf08b167e89 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -12,7 +12,9 @@ import json import logging import operator +import os import sys +import tempfile import time import traceback import urllib.parse @@ -25,6 +27,7 @@ Sequence, ) from contextvars import Token +from pathlib import Path from types import SimpleNamespace from typing import TYPE_CHECKING, Any, overload @@ -1743,14 +1746,40 @@ def _compile( clear_hash_caches() def _write_stateful_pages_marker(self): - """Write list of routes that create dynamic states for the backend to use later.""" - if self._state is not None: - stateful_pages_marker = ( - prerequisites.get_backend_dir() / constants.Dirs.STATEFUL_PAGES - ) - stateful_pages_marker.parent.mkdir(parents=True, exist_ok=True) - with stateful_pages_marker.open("w") as f: + """Write list of routes that create dynamic states for the backend to use later. + + Multiple backend workers may write the marker at the same time, so the + content is written to a temporary file and swapped into place with + ``Path.replace`` to ensure readers only ever see a complete marker. + """ + stateful_pages_marker = ( + prerequisites.get_backend_dir() / constants.Dirs.STATEFUL_PAGES + ) + stateful_pages_marker.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_path = tempfile.mkstemp( + dir=stateful_pages_marker.parent, + prefix=f"{stateful_pages_marker.name}.", + suffix=".tmp", + ) + os.close(fd) + tmp_marker = Path(tmp_path) + try: + with tmp_marker.open("w", encoding="utf-8") as f: json.dump(list(self._stateful_pages), f) + tmp_marker.chmod(0o644) + for attempt in range(100): + try: + tmp_marker.replace(stateful_pages_marker) + break + except PermissionError: + if not constants.IS_WINDOWS or attempt == 99: + raise + # Windows readers temporarily prevent replacing their open file. + # Wait 10 milliseconds before retrying. + time.sleep(0.01) + except BaseException: + tmp_marker.unlink(missing_ok=True) + raise def add_all_routes_endpoint(self): """Add an endpoint to the app that returns all the routes.""" diff --git a/reflex/compiler/compiler.py b/reflex/compiler/compiler.py index f06f6bb9426..e39f622df7c 100644 --- a/reflex/compiler/compiler.py +++ b/reflex/compiler/compiler.py @@ -1206,6 +1206,27 @@ def _register_plugin_routes(app: App, plugins: Sequence[Plugin]) -> None: app._register_plugin_pages(plugins) +def _read_stateful_pages_marker() -> list[str] | None: + """Read the routes that create state classes from a previous compile. + + A missing marker or one truncated by an older writer requires full page + evaluation. New writers replace the marker atomically. + + Returns: + The stateful routes, or None if no valid marker has been written yet. + """ + marker = prerequisites.get_backend_dir() / constants.Dirs.STATEFUL_PAGES + try: + return json.loads(marker.read_text()) + except (FileNotFoundError, json.JSONDecodeError): + return None + except PermissionError: + if constants.IS_WINDOWS: + # A concurrent atomic replacement can temporarily block Windows readers. + return None + raise + + def compile_app( app: App, *, @@ -1231,15 +1252,14 @@ def compile_app( app._pages = {} should_compile = app._should_compile() - backend_dir = prerequisites.get_backend_dir() - if not dry_run and not should_compile and backend_dir.exists(): - stateful_pages_marker = backend_dir / constants.Dirs.STATEFUL_PAGES - if stateful_pages_marker.exists(): - with stateful_pages_marker.open("r") as file: - stateful_pages = json.load(file) - for route in stateful_pages: - logger.debug(f"BE Evaluating stateful page: {route}") - app._compile_page(route, save_page=False) + if not dry_run and not should_compile: + stateful_pages = _read_stateful_pages_marker() + else: + stateful_pages = None + if stateful_pages is not None: + for route in stateful_pages: + logger.debug(f"BE Evaluating stateful page: {route}") + app._compile_page(route, save_page=False) if app._state is not None: utils._restore_bundled_libraries() utils._compile_initial_state(app._state) @@ -1319,7 +1339,8 @@ def compile_app( app._evaluated_pages.update(compile_ctx.compiled_pages) app._stateful_pages.update(compile_ctx.stateful_routes) - app._write_stateful_pages_marker() + if not dry_run: + app._write_stateful_pages_marker() app._add_optional_endpoints() app._validate_var_dependencies() diff --git a/tests/units/compiler/test_compiler.py b/tests/units/compiler/test_compiler.py index 07307221829..e25fc85d75e 100644 --- a/tests/units/compiler/test_compiler.py +++ b/tests/units/compiler/test_compiler.py @@ -31,6 +31,28 @@ from reflex.utils import prerequisites +@pytest.mark.parametrize("content", ["", '["index",']) +def test_read_stateful_pages_marker_recovers_legacy_corruption( + tmp_path, mocker, content +): + """A marker truncated by an older writer requests full page evaluation.""" + mocker.patch("reflex.utils.prerequisites.get_backend_dir", return_value=tmp_path) + (tmp_path / constants.Dirs.STATEFUL_PAGES).write_text(content) + assert compiler._read_stateful_pages_marker() is None + + +@pytest.mark.parametrize("windows", [False, True]) +def test_read_stateful_pages_marker_sharing_violation(mocker, windows): + """An unavailable Windows marker requests evaluation without hiding POSIX errors.""" + mocker.patch.object(constants, "IS_WINDOWS", windows) + mocker.patch.object(Path, "read_text", side_effect=PermissionError) + if windows: + assert compiler._read_stateful_pages_marker() is None + else: + with pytest.raises(PermissionError): + compiler._read_stateful_pages_marker() + + @pytest.mark.parametrize( ("fields", "test_default", "test_rest"), [ @@ -922,7 +944,37 @@ def test_compile_registers_plugin_routes_on_backend_early_return( if with_stateful_marker: compile_page.assert_called_once_with("plugin-page", save_page=False) else: - compile_page.assert_not_called() + compile_page.assert_any_call("plugin-page", save_page=False) + + +@pytest.mark.usefixtures("clean_registration_context") +def test_backend_compile_evaluates_all_pages_when_marker_missing( + tmp_path: Path, mocker: MockerFixture +): + """A backend dir without a complete marker falls through to evaluating every page. + + Another worker may have created the backend dir but not yet swapped its + marker into place, so a missing marker must not be mistaken for "no + stateful pages". + """ + app = rx.App(enable_state=False) + app.add_page(lambda: rx.fragment(), route="index") + mocker.patch.object(app, "_apply_decorated_pages") + mocker.patch.object(app, "_should_compile", return_value=False) + compile_page = mocker.patch.object(app, "_compile_page") + mocker.patch.object(app, "_add_optional_endpoints") + mocker.patch.object(prerequisites, "get_backend_dir", return_value=tmp_path) + mocker.patch.object( + compiler, "get_config", return_value=rx.Config(app_name="testing", plugins=[]) + ) + + assert compiler.compile_app(app, use_rich=False) is False + + assert {call.args[0] for call in compile_page.call_args_list} == { + "index", + constants.Page404.SLUG, + } + assert json.loads((tmp_path / constants.Dirs.STATEFUL_PAGES).read_text()) == [] @pytest.mark.usefixtures("clean_registration_context") diff --git a/tests/units/test_app.py b/tests/units/test_app.py index bf41e405d56..3eadd66418b 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -9,11 +9,15 @@ import json import logging import multiprocessing +import os import pickle import re +import tempfile +import threading import unittest.mock import uuid from collections.abc import Generator +from concurrent.futures import ThreadPoolExecutor from contextlib import nullcontext as does_not_raise from importlib.util import find_spec from pathlib import Path @@ -68,6 +72,7 @@ from reflex.compiler.compiler import ( _compile_app, _memoize_stateful_app_wraps, + _read_stateful_pages_marker, _resolve_app_wrap_components, ) from reflex.compiler.plugins import default_page_plugins @@ -4998,3 +5003,172 @@ def test_compile_emits_stage_spans( parent = spans[name].parent assert parent is not None assert parent.span_id == root.get_span_context().span_id + + +def test_write_stateful_pages_marker_never_truncates_final_path( + tmp_path: Path, mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """The marker is swapped into place atomically, never opened for writing.""" + mocker.patch("reflex.utils.prerequisites.get_backend_dir", return_value=tmp_path) + marker = tmp_path / constants.Dirs.STATEFUL_PAGES + original_open = Path.open + write_opens: list[str] = [] + + def spy_open(self: Path, mode: str = "r", *args, **kwargs): + if self == marker and mode != "r": + write_opens.append(mode) + return original_open(self, mode, *args, **kwargs) + + monkeypatch.setattr(Path, "open", spy_open) + app = App(_state=rx.State) + app._stateful_pages = dict.fromkeys(["index", "about"]) + + app._write_stateful_pages_marker() + + assert write_opens == [] + assert json.loads(marker.read_text()) == ["index", "about"] + assert [p.name for p in tmp_path.iterdir()] == [constants.Dirs.STATEFUL_PAGES] + + +def test_write_stateful_pages_marker_is_always_written( + tmp_path: Path, mocker: MockerFixture +): + """Stateless apps write an empty marker so backend workers skip page evaluation.""" + mocker.patch("reflex.utils.prerequisites.get_backend_dir", return_value=tmp_path) + app = App(enable_state=False) + + app._write_stateful_pages_marker() + + assert json.loads((tmp_path / constants.Dirs.STATEFUL_PAGES).read_text()) == [] + + +@pytest.mark.skipif(os.name == "nt", reason="Unix file permissions") +def test_write_stateful_pages_marker_is_shared_readable(tmp_path, mocker): + """Backend workers running as another user can read the compiled marker.""" + mocker.patch("reflex.utils.prerequisites.get_backend_dir", return_value=tmp_path) + app = App(enable_state=False) + app._write_stateful_pages_marker() + assert (tmp_path / constants.Dirs.STATEFUL_PAGES).stat().st_mode & 0o777 == 0o644 + + +def test_write_stateful_pages_marker_closes_descriptor_on_open_failure( + tmp_path, mocker +): + """Failure to open the temporary marker must not leak its raw descriptor.""" + mocker.patch("reflex.utils.prerequisites.get_backend_dir", return_value=tmp_path) + created = mocker.spy(tempfile, "mkstemp") + mocker.patch("os.fdopen", side_effect=OSError("open failed")) + mocker.patch.object(Path, "open", side_effect=OSError("open failed")) + with pytest.raises(OSError, match="open failed"): + App(enable_state=False)._write_stateful_pages_marker() + descriptor, _ = created.spy_return + try: + with pytest.raises(OSError): + os.fstat(descriptor) + finally: + with contextlib.suppress(OSError): + os.close(descriptor) + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize(("windows", "failures"), [(True, 1), (True, 100), (False, 1)]) +def test_write_stateful_pages_marker_sharing_violation( + tmp_path, mocker, windows, failures +): + """Windows sharing violations are retried without hiding persistent failures.""" + mocker.patch("reflex.utils.prerequisites.get_backend_dir", return_value=tmp_path) + mocker.patch("reflex.app.constants.IS_WINDOWS", windows) + sleep = mocker.patch("reflex.app.time.sleep") + original_replace = Path.replace + attempts = 0 + + def replace(path, target): + """Simulate a reader holding the Windows marker open. + + Args: + path: The temporary marker. + target: The final marker. + + Returns: + The replacement path. + + Raises: + PermissionError: While the simulated reader has the marker open. + """ + nonlocal attempts + attempts += 1 + if attempts <= failures: + msg = "marker is open" + raise PermissionError(msg) + return original_replace(path, target) + + mocker.patch.object(Path, "replace", replace) + app = App(enable_state=False) + if windows and failures == 1: + app._write_stateful_pages_marker() + assert json.loads((tmp_path / constants.Dirs.STATEFUL_PAGES).read_text()) == [] + assert attempts == 2 + sleep.assert_called_once_with(0.01) + else: + with pytest.raises(PermissionError, match="marker is open"): + app._write_stateful_pages_marker() + assert attempts == (100 if windows else 1) + assert sleep.call_count == attempts - 1 + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize("existing", [False, True]) +def test_compile_dry_run_preserves_stateful_marker(compilable_app, mocker, existing): + """A dry compile neither creates nor replaces the backend route marker.""" + app, web_dir = compilable_app + mocker.patch("reflex.utils.prerequisites.get_web_dir", return_value=web_dir) + marker = web_dir / constants.Dirs.BACKEND / constants.Dirs.STATEFUL_PAGES + if existing: + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text('["previous"]') + app._compile(dry_run=True) + assert marker.exists() == existing + if existing: + assert marker.read_text() == '["previous"]' + + +def test_write_stateful_pages_marker_concurrent_readers_see_valid_json( + tmp_path: Path, mocker: MockerFixture +): + """Concurrent writers and readers of the marker never observe a partial file.""" + mocker.patch("reflex.utils.prerequisites.get_backend_dir", return_value=tmp_path) + marker = tmp_path / constants.Dirs.STATEFUL_PAGES + routes = [f"route-{i}" for i in range(4000)] + app = App(_state=rx.State) + app._stateful_pages = dict.fromkeys(routes) + app._write_stateful_pages_marker() + round_started = threading.Barrier(8, timeout=10) + + def writer(): + """Repeatedly replace the marker.""" + for _ in range(50): + round_started.wait() + app._write_stateful_pages_marker() + + def reader(): + """Check that every observed marker is complete.""" + # Backend workers read on startup; an infinite read storm can starve + # Windows replacement because its readers do not share delete access. + for _ in range(50): + round_started.wait() + content = _read_stateful_pages_marker() + if content is None: + continue + assert content == routes + + with ThreadPoolExecutor(max_workers=8) as pool: + readers = [pool.submit(reader) for _ in range(4)] + try: + writers = [pool.submit(writer) for _ in range(4)] + for future in writers: + future.result() + finally: + round_started.abort() + for future in readers: + future.result() + assert json.loads(marker.read_text()) == routes