diff --git a/news/7142.bugfix.md b/news/7142.bugfix.md new file mode 100644 index 00000000000..32e5d78563e --- /dev/null +++ b/news/7142.bugfix.md @@ -0,0 +1 @@ +Fix a startup race in backend-only mode with multiple workers where a worker could read a truncated `.web/backend/stateful_pages.json` and crash with `JSONDecodeError`. The marker is now written atomically, and a worker that finds no marker evaluates all pages instead of assuming there are none. diff --git a/reflex/app.py b/reflex/app.py index f378a74ab9f..363eba39915 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 @@ -1714,14 +1717,29 @@ 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", + ) + tmp_marker = Path(tmp_path) + try: + with os.fdopen(fd, "w") as f: json.dump(list(self._stateful_pages), f) + tmp_marker.replace(stateful_pages_marker) + 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 487b3845963..8d7e8fdc7e3 100644 --- a/reflex/compiler/compiler.py +++ b/reflex/compiler/compiler.py @@ -1206,6 +1206,23 @@ 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. + + The marker is swapped into place atomically, so it is either complete or + absent. It may be absent because no compile has happened yet or because a + concurrently starting worker has not finished writing it. + + Returns: + The stateful routes, or None if no marker has been written yet. + """ + marker = prerequisites.get_backend_dir() / constants.Dirs.STATEFUL_PAGES + try: + return json.loads(marker.read_text()) + except FileNotFoundError: + return None + + def compile_app( app: App, *, @@ -1231,15 +1248,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) diff --git a/tests/units/compiler/test_compiler.py b/tests/units/compiler/test_compiler.py index 5b02be0ad80..6d5e29af9cc 100644 --- a/tests/units/compiler/test_compiler.py +++ b/tests/units/compiler/test_compiler.py @@ -888,7 +888,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 e5ead218a92..68bcff0227e 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -11,6 +11,7 @@ import multiprocessing import pickle import re +import threading import unittest.mock import uuid from collections.abc import Generator @@ -4799,3 +4800,82 @@ 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()) == [] + + +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) + stop = threading.Event() + errors: list[BaseException] = [] + + def writer(): + for _ in range(50): + app._write_stateful_pages_marker() + + def reader(): + while not stop.is_set(): + try: + content = marker.read_text() + except FileNotFoundError: + continue + try: + assert json.loads(content) == routes + except (AssertionError, json.JSONDecodeError) as exc: + errors.append(exc) + return + + writers = [threading.Thread(target=writer) for _ in range(4)] + readers = [threading.Thread(target=reader) for _ in range(4)] + for thread in readers + writers: + thread.start() + for thread in writers: + thread.join() + stop.set() + for thread in readers: + thread.join() + + assert errors == [] + assert json.loads(marker.read_text()) == routes