Fix race condition in stateful pages marker with atomic writes

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.

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."""

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()
 

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")

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