From 7020987e3a343b0b1bb13c709fe8fb4a756ad9b5 Mon Sep 17 00:00:00 2001 From: Alek Date: Sun, 6 Sep 2026 00:58:44 -0700 Subject: [PATCH 1/3] Serialize concurrent frontend dependency installs --- news/+serialize-frontend-installs.bugfix.md | 1 + .../news/+atomic-procedure-cache.bugfix.md | 1 + .../src/reflex_base/constants/base.py | 2 + .../src/reflex_base/constants/config.py | 1 + .../src/reflex_base/utils/decorator.py | 31 +- reflex/utils/frontend_lock.py | 113 +++ reflex/utils/frontend_skeleton.py | 59 +- reflex/utils/js_runtimes.py | 15 +- tests/units/test_prerequisites.py | 665 +++++++++++++++++- 9 files changed, 867 insertions(+), 21 deletions(-) create mode 100644 news/+serialize-frontend-installs.bugfix.md create mode 100644 packages/reflex-base/news/+atomic-procedure-cache.bugfix.md create mode 100644 reflex/utils/frontend_lock.py diff --git a/news/+serialize-frontend-installs.bugfix.md b/news/+serialize-frontend-installs.bugfix.md new file mode 100644 index 00000000000..5ab58462229 --- /dev/null +++ b/news/+serialize-frontend-installs.bugfix.md @@ -0,0 +1 @@ +Serialize frontend dependency installation per app and atomically persist package-manager files so concurrent Reflex commands cannot corrupt or duplicate an install. diff --git a/packages/reflex-base/news/+atomic-procedure-cache.bugfix.md b/packages/reflex-base/news/+atomic-procedure-cache.bugfix.md new file mode 100644 index 00000000000..2adc97b6a80 --- /dev/null +++ b/packages/reflex-base/news/+atomic-procedure-cache.bugfix.md @@ -0,0 +1 @@ +Write disk-backed procedure caches atomically and recover automatically from truncated cache data. diff --git a/packages/reflex-base/src/reflex_base/constants/base.py b/packages/reflex-base/src/reflex_base/constants/base.py index b5c9079517e..221238e3201 100644 --- a/packages/reflex-base/src/reflex_base/constants/base.py +++ b/packages/reflex-base/src/reflex_base/constants/base.py @@ -21,6 +21,8 @@ class Dirs(SimpleNamespace): """Various directories/paths used by Reflex.""" # The frontend directories in a project. + # Stable project-root lock used while replacing WEB and installing packages. + FRONTEND_INSTALL_LOCK = ".reflex.frontend.lock" # The web folder where the frontend app is compiled to. WEB = ".web" # The directory where uploaded files are stored. diff --git a/packages/reflex-base/src/reflex_base/constants/config.py b/packages/reflex-base/src/reflex_base/constants/config.py index 1ff594f12b1..886f28e5bab 100644 --- a/packages/reflex-base/src/reflex_base/constants/config.py +++ b/packages/reflex-base/src/reflex_base/constants/config.py @@ -40,6 +40,7 @@ class GitIgnore(SimpleNamespace): FILE = Path(".gitignore") # Files to gitignore. DEFAULTS = { + Dirs.FRONTEND_INSTALL_LOCK, Dirs.WEB, Dirs.STATES, "*.db", diff --git a/packages/reflex-base/src/reflex_base/utils/decorator.py b/packages/reflex-base/src/reflex_base/utils/decorator.py index e2b27ee1f01..f33ce80e47a 100644 --- a/packages/reflex-base/src/reflex_base/utils/decorator.py +++ b/packages/reflex-base/src/reflex_base/utils/decorator.py @@ -77,18 +77,43 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: def _write_cached_procedure_file(payload: str, cache_file: Path, value: object): + import contextlib import pickle + import uuid + if cache_file.is_symlink(): + cache_file = cache_file.resolve() cache_file.parent.mkdir(parents=True, exist_ok=True) - cache_file.write_bytes(pickle.dumps((payload, value))) + mode = cache_file.stat().st_mode if cache_file.exists() else None + temporary_path = cache_file.with_name(f".{cache_file.name}.{uuid.uuid4().hex}.tmp") + created = False + try: + with temporary_path.open("xb") as temporary_file: + created = True + temporary_file.write(pickle.dumps((payload, value))) + if mode is not None: + temporary_path.chmod(mode & 0o7777) + temporary_path.replace(cache_file) + except BaseException: + if created: + with contextlib.suppress(OSError): + temporary_path.unlink(missing_ok=True) + raise def _read_cached_procedure_file(cache_file: Path) -> tuple[str | None, object]: import pickle if cache_file.exists(): - with cache_file.open("rb") as f: - return pickle.loads(f.read()) + try: + with cache_file.open("rb") as f: + payload, value = pickle.loads(f.read()) + if not isinstance(payload, str): + return None, None + except (pickle.UnpicklingError, EOFError, TypeError, ValueError) as err: + logger.debug(f"Ignoring invalid procedure cache {cache_file}: {err}") + else: + return payload, value return None, None diff --git a/reflex/utils/frontend_lock.py b/reflex/utils/frontend_lock.py new file mode 100644 index 00000000000..265158e5d8e --- /dev/null +++ b/reflex/utils/frontend_lock.py @@ -0,0 +1,113 @@ +"""Cross-process locking for frontend project mutations.""" + +from __future__ import annotations + +import contextlib +import errno +import os +import threading +import time +from collections.abc import Iterator +from pathlib import Path +from typing import BinaryIO + +from reflex_base import constants + +_project_locks_guard = threading.Lock() +_project_locks: dict[Path, threading.RLock] = {} +_project_locks_held = threading.local() +_WINDOWS_LOCK_RETRY_DELAY = 0.05 + + +def _acquire_project_file_lock(lock_file: BinaryIO) -> None: + """Acquire an advisory lock on an open project lock file. + + Args: + lock_file: Binary file kept open for the lifetime of the lock. + """ + lock_file.seek(0, os.SEEK_END) + if lock_file.tell() == 0: + lock_file.write(b"\0") + lock_file.flush() + lock_file.seek(0) + + if constants.IS_WINDOWS: + import msvcrt + + while True: + try: + lock_file.seek(0) + msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] + lock_file.fileno(), + msvcrt.LK_NBLCK, # pyright: ignore[reportAttributeAccessIssue] + 1, + ) + except OSError as err: # noqa: PERF203 # contention requires retrying + if err.errno not in {errno.EACCES, errno.EAGAIN, errno.EDEADLK}: + raise + time.sleep(_WINDOWS_LOCK_RETRY_DELAY) + else: + return + + import fcntl + + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + + +def _release_project_file_lock(lock_file: BinaryIO) -> None: + """Release an advisory lock on an open project lock file. + + Args: + lock_file: Binary file previously passed to the acquire helper. + """ + lock_file.seek(0) + if constants.IS_WINDOWS: + import msvcrt + + msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] + lock_file.fileno(), + msvcrt.LK_UNLCK, # pyright: ignore[reportAttributeAccessIssue] + 1, + ) + return + + import fcntl + + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +@contextlib.contextmanager +def frontend_project_lock() -> Iterator[None]: + """Serialize frontend directory mutations for the current app. + + The stable project-root lock file is outside ``.web`` and ``reflex.lock`` + because both directories may be replaced during recovery. Advisory OS locks + are released automatically if a process exits, while the in-process + reentrant lock makes nested use from the same thread safe. + + Yields: + Once this process exclusively owns the current app's frontend lock. + """ + lock_path = (Path.cwd() / constants.Dirs.FRONTEND_INSTALL_LOCK).resolve() + with _project_locks_guard: + thread_lock = _project_locks.get(lock_path) + if thread_lock is None: + thread_lock = _project_locks[lock_path] = threading.RLock() + + with thread_lock: + held_paths = getattr(_project_locks_held, "paths", None) + if held_paths is None: + held_paths = _project_locks_held.paths = set() + if lock_path in held_paths: + yield + return + + lock_path.parent.mkdir(parents=True, exist_ok=True) + with lock_path.open("a+b") as lock_file: + _acquire_project_file_lock(lock_file) + held_paths.add(lock_path) + try: + yield + finally: + held_paths.remove(lock_path) + _release_project_file_lock(lock_file) diff --git a/reflex/utils/frontend_skeleton.py b/reflex/utils/frontend_skeleton.py index 9a5fa3d9ee3..189b598d544 100644 --- a/reflex/utils/frontend_skeleton.py +++ b/reflex/utils/frontend_skeleton.py @@ -320,16 +320,50 @@ def _copy_if_exists(src: Path, dest: Path, prune: bool = True) -> bool: return True return False - if dest.exists() and dest.read_bytes() == src.read_bytes(): + contents = src.read_bytes() + if dest.exists() and dest.read_bytes() == contents: return False changed = dest.exists() - path_ops.mkdir(dest.parent) + mode = dest.stat().st_mode if changed else src.stat().st_mode logger.debug(f"Copying {src} to {dest}") - path_ops.cp(src, dest) + _write_bytes_atomic(dest, contents, mode=mode) return changed +def _write_bytes_atomic(path: Path, contents: bytes, mode: int | None = None) -> None: + """Atomically replace a file with complete new contents. + + Args: + path: Destination path to replace. Existing symlinks are followed so + their link objects remain intact. + contents: Bytes to write before committing the replacement. + mode: File mode to apply. Existing target permissions are preserved by + default; a new generated file uses the process umask. + """ + import contextlib + + if path.is_symlink(): + path = path.resolve() + path.parent.mkdir(parents=True, exist_ok=True) + if mode is None and path.exists(): + mode = path.stat().st_mode + temporary_path = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + created = False + try: + with temporary_path.open("xb") as temporary_file: + created = True + temporary_file.write(contents) + if mode is not None: + temporary_path.chmod(mode & 0o7777) + temporary_path.replace(path) + except BaseException: + if created: + with contextlib.suppress(OSError): + temporary_path.unlink(missing_ok=True) + raise + + def sync_root_lockfile_to_web(filename: str, prune: bool = True) -> bool: """Mirror a single persisted lockfile into ``.web``. @@ -372,7 +406,7 @@ def sync_root_package_json_to_web() -> bool: changed = output_path.exists() path_ops.mkdir(output_path.parent) - output_path.write_text(rendered) + _write_bytes_atomic(output_path, rendered.encode()) return changed @@ -398,10 +432,7 @@ def sync_web_lockfile_to_root(filename: str): web = get_web_lockfile_path(filename) if not web.exists(): return - root = get_root_lockfile_path(filename) - path_ops.mkdir(root.parent) - logger.debug(f"Copying {web} to {root}") - path_ops.cp(web, root) + _copy_if_exists(web, get_root_lockfile_path(filename), prune=False) def sync_web_lockfiles_to_root(): @@ -450,6 +481,14 @@ def _read_persisted_package_json() -> dict: def initialize_web_directory(): """Initialize the web directory on reflex init.""" + from reflex.utils.frontend_lock import frontend_project_lock + + with frontend_project_lock(): + _initialize_web_directory() + + +def _initialize_web_directory(): + """Initialize the web directory while its project lock is held.""" logger.info("Initializing the web directory.") # Reuse the hash if one is already created, so we don't over-write it when running reflex init @@ -598,14 +637,14 @@ def update_package_json_overrides() -> bool: package_json["overrides"] = {**overrides, **constants.PackageJson.OVERRIDES} logger.debug(f"Applying framework overrides to {package_json_path}") - package_json_path.write_text(json.dumps(package_json)) + _write_bytes_atomic(package_json_path, json.dumps(package_json).encode()) return True def initialize_package_json(): """Render and write in .web the package.json file.""" output_path = get_web_dir() / constants.PackageJson.PATH - output_path.write_text(_compile_package_json()) + _write_bytes_atomic(output_path, _compile_package_json().encode()) def _compile_vite_config(config: Config): diff --git a/reflex/utils/js_runtimes.py b/reflex/utils/js_runtimes.py index 2f5b9a5188d..1983b3bd07e 100644 --- a/reflex/utils/js_runtimes.py +++ b/reflex/utils/js_runtimes.py @@ -783,9 +783,12 @@ def _install_frontend_packages( def install_frontend_packages(packages: set[str], config: Config): """Install frontend packages while respecting the canonical root bun.lock.""" - install_package_managers = tuple( - get_nodejs_compatible_package_managers(raise_on_none=True) - ) - _sync_root_lockfiles_for_frontend_install() - _install_frontend_packages(set(packages), config, install_package_managers) - frontend_skeleton.sync_web_lockfiles_to_root() + from reflex.utils.frontend_lock import frontend_project_lock + + with frontend_project_lock(): + install_package_managers = tuple( + get_nodejs_compatible_package_managers(raise_on_none=True) + ) + _sync_root_lockfiles_for_frontend_install() + _install_frontend_packages(set(packages), config, install_package_managers) + frontend_skeleton.sync_web_lockfiles_to_root() diff --git a/tests/units/test_prerequisites.py b/tests/units/test_prerequisites.py index 5cba1a236e3..759210eadf8 100644 --- a/tests/units/test_prerequisites.py +++ b/tests/units/test_prerequisites.py @@ -1,11 +1,16 @@ import json +import multiprocessing +import os +import pickle import shutil +import sys import tempfile import uuid from collections.abc import Callable, Generator +from contextlib import ExitStack, contextmanager from dataclasses import dataclass from pathlib import Path -from typing import Protocol +from typing import Any, Protocol import pytest from click.testing import CliRunner @@ -16,7 +21,7 @@ from reflex.reflex import cli from reflex.testing import chdir -from reflex.utils import frontend_skeleton, js_runtimes, prerequisites +from reflex.utils import frontend_lock, frontend_skeleton, js_runtimes, prerequisites from reflex.utils.frontend_skeleton import ( _compile_vite_config, _update_react_router_config, @@ -27,6 +32,141 @@ runner = CliRunner() +def _hold_frontend_project_lock( + project_dir: str, + acquired: Any, + release: Any, + *, + exit_without_cleanup: bool = False, + nested: bool = False, +) -> None: + """Hold a frontend project lock in a spawned process. + + Args: + project_dir: App root whose lock should be acquired. + acquired: Multiprocessing event set after acquisition. + release: Multiprocessing event that releases a normal holder. + exit_without_cleanup: Exit the process without running ``finally`` blocks. + nested: Acquire the same project lock twice before signalling. + """ + os.chdir(project_dir) + with ExitStack() as stack: + stack.enter_context(frontend_lock.frontend_project_lock()) + if nested: + stack.enter_context(frontend_lock.frontend_project_lock()) + acquired.set() + if exit_without_cleanup: + os._exit(0) + if not release.wait(20): + msg = "Timed out waiting to release frontend project lock" + raise TimeoutError(msg) + + +def _frontend_project_file_lock_is_contended(project_dir: Path) -> bool: + """Probe the OS-level project lock without blocking. + + Args: + project_dir: App root whose lock file should be probed. + + Returns: + Whether another process currently owns the project lock. + """ + lock_path = project_dir / constants.Dirs.FRONTEND_INSTALL_LOCK + with lock_path.open("r+b") as lock_file: + lock_file.seek(0) + if constants.IS_WINDOWS: + import msvcrt + + try: + msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] + lock_file.fileno(), + msvcrt.LK_NBLCK, # pyright: ignore[reportAttributeAccessIssue] + 1, + ) + except OSError as err: + if err.errno in { + frontend_lock.errno.EACCES, + frontend_lock.errno.EAGAIN, + frontend_lock.errno.EDEADLK, + }: + return True + raise + lock_file.seek(0) + msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] + lock_file.fileno(), + msvcrt.LK_UNLCK, # pyright: ignore[reportAttributeAccessIssue] + 1, + ) + return False + + import fcntl + + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as err: + if err.errno in { + frontend_lock.errno.EACCES, + frontend_lock.errno.EAGAIN, + }: + return True + raise + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + return False + + +def _install_frontend_packages_in_process( + project_dir: str, + attempting: Any, + package_manager_entered: Any, + release_package_manager: Any, + active_installs: Any, + max_active_installs: Any, +) -> None: + """Run a deterministic fake frontend install in a spawned process. + + Args: + project_dir: App root containing the shared ``.web`` directory. + attempting: Event set immediately before the public install call. + package_manager_entered: Event set if this process invokes the package manager. + release_package_manager: Event allowing the fake package manager to finish. + active_installs: Shared count of active package-manager calls. + max_active_installs: Shared maximum active package-manager calls. + """ + os.chdir(project_dir) + constants.PackageJson.DEPENDENCIES = {} # pyright: ignore[reportAttributeAccessIssue] + constants.PackageJson.DEV_DEPENDENCIES = {} # pyright: ignore[reportAttributeAccessIssue] + constants.PackageJson.OVERRIDES = {} # pyright: ignore[reportAttributeAccessIssue] + js_runtimes.get_nodejs_compatible_package_managers = lambda raise_on_none=True: ( + "bun", + ) + + def run_package_manager(args, **kwargs) -> None: + """Block a fake package-manager add so a competing process can contend.""" + if "add" not in args: + return + with active_installs.get_lock(): + active_installs.value += 1 + max_active_installs.value = max( + max_active_installs.value, active_installs.value + ) + package_manager_entered.set() + try: + if not release_package_manager.wait(20): + msg = "Timed out waiting to finish fake frontend install" + raise TimeoutError(msg) + package_json_path = Path(constants.Dirs.WEB) / constants.PackageJson.PATH + package_json = json.loads(package_json_path.read_text()) + package_json.setdefault("dependencies", {})["race-pkg"] = "1.0.0" + package_json_path.write_text(json.dumps(package_json)) + finally: + with active_installs.get_lock(): + active_installs.value -= 1 + + js_runtimes.processes.run_process_with_fallbacks = run_package_manager + attempting.set() + js_runtimes.install_frontend_packages({"race-pkg@1.0.0"}, Config(app_name="test")) + + def _patch_web_dir(monkeypatch: pytest.MonkeyPatch, web_dir: Path): monkeypatch.setattr(frontend_skeleton, "get_web_dir", lambda: web_dir) monkeypatch.setattr(js_runtimes, "get_web_dir", lambda: web_dir) @@ -134,6 +274,344 @@ def install(packages: set[str] | None = None) -> None: yield env +def test_install_frontend_packages_locks_the_complete_transaction( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The project lock covers manager selection, cache work, and persistence.""" + calls: list[str] = [] + lock_held = False + + @contextmanager + def project_lock(): + nonlocal lock_held + calls.append("lock-enter") + lock_held = True + try: + yield + finally: + lock_held = False + calls.append("lock-exit") + + def record(name: str, result=None): + """Return a stub that records a call made while the lock is held.""" + + def stub(*args, **kwargs): + assert lock_held + calls.append(name) + return result + + return stub + + monkeypatch.setattr(frontend_lock, "frontend_project_lock", project_lock) + monkeypatch.setattr( + js_runtimes, + "get_nodejs_compatible_package_managers", + record("select-manager", ("bun",)), + ) + monkeypatch.setattr( + js_runtimes, + "_sync_root_lockfiles_for_frontend_install", + record("sync-root"), + ) + monkeypatch.setattr( + js_runtimes, + "_install_frontend_packages", + record("install"), + ) + monkeypatch.setattr( + frontend_skeleton, + "sync_web_lockfiles_to_root", + record("sync-web"), + ) + + with chdir(tmp_path): + js_runtimes.install_frontend_packages(set(), Config(app_name="test")) + + assert calls == [ + "lock-enter", + "select-manager", + "sync-root", + "install", + "sync-web", + "lock-exit", + ] + + +def test_frontend_project_lock_is_reentrant_and_ignored(tmp_path: Path) -> None: + """Nested use in one thread succeeds and its stable file is gitignored.""" + ctx = multiprocessing.get_context("spawn") + acquired = ctx.Event() + release = ctx.Event() + process = ctx.Process( + target=_hold_frontend_project_lock, + args=(str(tmp_path), acquired, release), + kwargs={"nested": True}, + ) + + nested_lock_acquired = False + started = False + try: + process.start() + started = True + nested_lock_acquired = acquired.wait(20) + finally: + release.set() + if started: + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert nested_lock_acquired + assert process.exitcode == 0 + assert (tmp_path / constants.Dirs.FRONTEND_INSTALL_LOCK).exists() + assert constants.Dirs.FRONTEND_INSTALL_LOCK in constants.GitIgnore.DEFAULTS + + +def test_frontend_project_lock_releases_after_exception(tmp_path: Path) -> None: + """An exception in a transaction does not strand its project lock.""" + error = RuntimeError("failed install") + with chdir(tmp_path): + with ( + pytest.raises(RuntimeError, match="failed install"), + frontend_lock.frontend_project_lock(), + ): + raise error + + with frontend_lock.frontend_project_lock(): + assert Path(constants.Dirs.FRONTEND_INSTALL_LOCK).exists() + + +def test_frontend_project_lock_blocks_same_project(tmp_path: Path) -> None: + """A held project lock is positively contended by another process.""" + ctx = multiprocessing.get_context("spawn") + acquired = ctx.Event() + release = ctx.Event() + process = ctx.Process( + target=_hold_frontend_project_lock, + args=(str(tmp_path), acquired, release), + ) + + lock_was_acquired = False + lock_was_contended = False + try: + process.start() + lock_was_acquired = acquired.wait(20) + if lock_was_acquired: + lock_was_contended = _frontend_project_file_lock_is_contended(tmp_path) + finally: + release.set() + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert lock_was_acquired + assert lock_was_contended + assert process.exitcode == 0 + + +def test_frontend_project_file_lock_retries_windows_contention( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mocker, +) -> None: + """The Windows path waits on contention and unlocks the same byte range.""" + fake_msvcrt = mocker.Mock() + fake_msvcrt.LK_NBLCK = 1 + fake_msvcrt.LK_UNLCK = 2 + contention = OSError(frontend_lock.errno.EACCES, "lock is held") + fake_msvcrt.locking.side_effect = [contention, contention, None, None] + sleep = mocker.patch.object(frontend_lock.time, "sleep") + monkeypatch.setattr(constants, "IS_WINDOWS", True) + monkeypatch.setitem(sys.modules, "msvcrt", fake_msvcrt) + lock_path = tmp_path / "lock" + + with lock_path.open("w+b") as lock_file: + frontend_lock._acquire_project_file_lock(lock_file) + frontend_lock._release_project_file_lock(lock_file) + + assert [call.args[1:] for call in fake_msvcrt.locking.call_args_list] == [ + (fake_msvcrt.LK_NBLCK, 1), + (fake_msvcrt.LK_NBLCK, 1), + (fake_msvcrt.LK_NBLCK, 1), + (fake_msvcrt.LK_UNLCK, 1), + ] + assert sleep.call_args_list == [ + mocker.call(frontend_lock._WINDOWS_LOCK_RETRY_DELAY), + mocker.call(frontend_lock._WINDOWS_LOCK_RETRY_DELAY), + ] + + +def test_frontend_project_locks_are_scoped_per_project(tmp_path: Path) -> None: + """Two separate app roots can hold their frontend locks concurrently.""" + ctx = multiprocessing.get_context("spawn") + project_a = tmp_path / "project-a" + project_b = tmp_path / "project-b" + project_a.mkdir() + project_b.mkdir() + acquired_a = ctx.Event() + acquired_b = ctx.Event() + release = ctx.Event() + processes = [ + ctx.Process( + target=_hold_frontend_project_lock, + args=(str(project_a), acquired_a, release), + ), + ctx.Process( + target=_hold_frontend_project_lock, + args=(str(project_b), acquired_b, release), + ), + ] + + for process in processes: + process.start() + try: + assert acquired_a.wait(20) + assert acquired_b.wait(20) + finally: + release.set() + for process in processes: + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert [process.exitcode for process in processes] == [0, 0] + + +def test_frontend_project_lock_recovers_after_crashed_process(tmp_path: Path) -> None: + """The OS releases the project lock when its owning process exits abruptly.""" + ctx = multiprocessing.get_context("spawn") + crashed_acquired = ctx.Event() + unused_release = ctx.Event() + crashed = ctx.Process( + target=_hold_frontend_project_lock, + args=(str(tmp_path), crashed_acquired, unused_release), + kwargs={"exit_without_cleanup": True}, + ) + crashed_lock_acquired = False + crashed_started = False + try: + crashed.start() + crashed_started = True + crashed_lock_acquired = crashed_acquired.wait(20) + crashed.join(20) + finally: + if crashed_started and crashed.is_alive(): + crashed.terminate() + crashed.join(5) + + assert crashed_lock_acquired + assert crashed.exitcode == 0 + + recovered_acquired = ctx.Event() + release = ctx.Event() + recovered = ctx.Process( + target=_hold_frontend_project_lock, + args=(str(tmp_path), recovered_acquired, release), + ) + recovered.start() + try: + assert recovered_acquired.wait(20) + finally: + release.set() + recovered.join(20) + if recovered.is_alive(): + recovered.terminate() + recovered.join(5) + + assert recovered.exitcode == 0 + + +def test_concurrent_frontend_installs_share_completed_cache( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A second process waits, then observes the first process's install cache.""" + web_dir = tmp_path / constants.Dirs.WEB + web_dir.mkdir() + _stub_framework_packages(monkeypatch) + with chdir(tmp_path): + frontend_skeleton.initialize_package_json() + + ctx = multiprocessing.get_context("spawn") + first_attempting = ctx.Event() + second_attempting = ctx.Event() + first_entered = ctx.Event() + second_entered = ctx.Event() + release_package_manager = ctx.Event() + active_installs = ctx.Value("i", 0) + max_active_installs = ctx.Value("i", 0) + first = ctx.Process( + target=_install_frontend_packages_in_process, + args=( + str(tmp_path), + first_attempting, + first_entered, + release_package_manager, + active_installs, + max_active_installs, + ), + ) + second = ctx.Process( + target=_install_frontend_packages_in_process, + args=( + str(tmp_path), + second_attempting, + second_entered, + release_package_manager, + active_installs, + max_active_installs, + ), + ) + + started_processes = [] + first_attempted_install = False + first_called_package_manager = False + second_attempted_install = False + second_called_package_manager_while_first_active = False + try: + first.start() + started_processes.append(first) + first_attempted_install = first_attempting.wait(20) + if first_attempted_install: + first_called_package_manager = first_entered.wait(20) + if first_called_package_manager: + second.start() + started_processes.append(second) + second_attempted_install = second_attempting.wait(20) + if second_attempted_install: + second_called_package_manager_while_first_active = second_entered.wait(1) + finally: + release_package_manager.set() + for process in started_processes: + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert first_attempted_install + assert first_called_package_manager + assert second_attempted_install + assert not second_called_package_manager_while_first_active + assert not second_entered.is_set() + assert first.exitcode == 0 + assert second.exitcode == 0 + assert max_active_installs.value == 1 + assert active_installs.value == 0 + + cache_file = web_dir / "reflex.install_frontend_packages.cached" + cache_payload, cache_value = pickle.loads(cache_file.read_bytes()) + assert isinstance(cache_payload, str) + assert cache_value is None + root_package_json = ( + tmp_path / constants.Bun.ROOT_LOCKFILE_DIR / constants.PackageJson.PATH + ) + assert json.loads(root_package_json.read_text()) == json.loads( + (web_dir / constants.PackageJson.PATH).read_text() + ) + + _SKELETON_INITIALIZERS = ( "initialize_package_json", "initialize_bun_config", @@ -167,6 +645,31 @@ def _stub_skeleton_initializers(monkeypatch): _stub_skeleton_initializers_except(monkeypatch) +def test_initialize_web_directory_holds_frontend_project_lock( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reinitialization cannot replace ``.web`` during an install transaction.""" + lock_held = False + + @contextmanager + def project_lock(): + nonlocal lock_held + lock_held = True + try: + yield + finally: + lock_held = False + + def initialize() -> None: + assert lock_held + + monkeypatch.setattr(frontend_lock, "frontend_project_lock", project_lock) + monkeypatch.setattr(frontend_skeleton, "_initialize_web_directory", initialize) + + frontend_skeleton.initialize_web_directory() + assert not lock_held + + @pytest.mark.parametrize( ("config", "export", "expected_output"), [ @@ -355,6 +858,66 @@ def test_sync_root_lockfiles_to_web_processes_package_json(tmp_path, monkeypatch assert web_pkg["scripts"]["export"] == constants.PackageJson.Commands.EXPORT +def test_sync_web_lockfile_to_root_preserves_old_file_if_replace_fails( + install_packages_env: InstallPackagesEnv, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failed atomic commit leaves the previous persisted lockfile valid.""" + env = install_packages_env + env.web_lock.write_text("new-lock") + env.root_lock.write_text("old-lock") + original_replace = Path.replace + error = OSError("simulated replace failure") + + def fail_root_replace(path: Path, target: Path) -> Path: + if target == env.root_lock: + raise error + return original_replace(path, target) + + monkeypatch.setattr(Path, "replace", fail_root_replace) + + with pytest.raises(OSError, match="simulated replace failure"): + frontend_skeleton.sync_web_lockfile_to_root(constants.Bun.LOCKFILE_PATH) + + assert env.root_lock.read_text() == "old-lock" + assert list(env.root_lock.parent.glob(f".{env.root_lock.name}.*.tmp")) == [] + + +@pytest.mark.skipif(constants.IS_WINDOWS, reason="Windows exposes limited chmod modes") +def test_sync_web_lockfile_to_root_preserves_existing_mode( + install_packages_env: InstallPackagesEnv, +) -> None: + """Atomic replacement does not make a shared persisted lockfile private.""" + env = install_packages_env + env.web_lock.write_text("new-lock") + env.root_lock.write_text("old-lock") + env.root_lock.chmod(0o644) + + frontend_skeleton.sync_web_lockfile_to_root(constants.Bun.LOCKFILE_PATH) + + assert env.root_lock.read_text() == "new-lock" + assert env.root_lock.stat().st_mode & 0o777 == 0o644 + + +def test_sync_web_lockfile_to_root_preserves_existing_symlink( + install_packages_env: InstallPackagesEnv, +) -> None: + """Atomic replacement follows an existing persisted lockfile symlink.""" + env = install_packages_env + env.web_lock.write_text("new-lock") + shared_lock = env.tmp_path / "shared-bun.lock" + shared_lock.write_text("old-lock") + try: + env.root_lock.symlink_to(shared_lock) + except OSError as err: + pytest.skip(f"Cannot create symlink on this platform: {err}") + + frontend_skeleton.sync_web_lockfile_to_root(constants.Bun.LOCKFILE_PATH) + + assert env.root_lock.is_symlink() + assert shared_lock.read_text() == "new-lock" + + def test_install_frontend_packages_syncs_root_bun_lock( install_packages_env: InstallPackagesEnv, ): @@ -1661,6 +2224,104 @@ def _function_with_no_args_fn(): assert call_count == 2 +def test_cached_procedure_treats_corrupt_file_as_miss(tmp_path: Path) -> None: + """A cache truncated by a killed process is recomputed and repaired.""" + cache_file = tmp_path / "procedure.cached" + cache_file.write_bytes(b"not a pickle") + call_count = 0 + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "payload") + def procedure() -> str: + nonlocal call_count + call_count += 1 + return "recomputed" + + assert procedure() == "recomputed" + assert procedure() == "recomputed" + assert call_count == 1 + assert pickle.loads(cache_file.read_bytes()) == ("payload", "recomputed") + + +def test_cached_procedure_propagates_cache_io_errors( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An inaccessible cache is not mistaken for an ordinary cache miss.""" + cache_file = tmp_path / "procedure.cached" + cache_file.write_bytes(pickle.dumps(("payload", "cached"))) + call_count = 0 + original_open = Path.open + error = PermissionError("cache access denied") + + def deny_cache_read(path: Path, *args, **kwargs): + if path == cache_file: + raise error + return original_open(path, *args, **kwargs) + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "payload") + def procedure() -> str: + nonlocal call_count + call_count += 1 + return "recomputed" + + monkeypatch.setattr(Path, "open", deny_cache_read) + + with pytest.raises(PermissionError, match="cache access denied"): + procedure() + + assert call_count == 0 + + +def test_cached_procedure_preserves_old_file_if_replace_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A failed cache commit leaves the previous complete pickle readable.""" + cache_file = tmp_path / "procedure.cached" + payload = "old" + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: payload) + def procedure() -> str: + return payload + + assert procedure() == "old" + old_cache = cache_file.read_bytes() + payload = "new" + original_replace = Path.replace + error = OSError("simulated replace failure") + + def fail_cache_replace(path: Path, target: Path) -> Path: + if target == cache_file: + raise error + return original_replace(path, target) + + monkeypatch.setattr(Path, "replace", fail_cache_replace) + + with pytest.raises(OSError, match="simulated replace failure"): + procedure() + + assert cache_file.read_bytes() == old_cache + assert pickle.loads(cache_file.read_bytes()) == ("old", "old") + assert list(tmp_path.glob(f".{cache_file.name}.*.tmp")) == [] + + +def test_cached_procedure_preserves_existing_symlink(tmp_path: Path) -> None: + """Atomic cache updates follow an existing cache symlink.""" + cache_target = tmp_path / "shared-procedure.cached" + cache_target.write_bytes(pickle.dumps(("old", "old"))) + cache_file = tmp_path / "procedure.cached" + try: + cache_file.symlink_to(cache_target.name) + except OSError as err: + pytest.skip(f"Cannot create symlink on this platform: {err}") + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "new") + def procedure() -> str: + return "new" + + assert procedure() == "new" + assert cache_file.is_symlink() + assert pickle.loads(cache_target.read_bytes()) == ("new", "new") + + def test_get_cpu_info(): cpu_info = get_cpu_info() assert cpu_info is not None From a1c9efdda5dbbc306dde02ef8fb4131e66cbce96 Mon Sep 17 00:00:00 2001 From: Alek Date: Sun, 6 Sep 2026 01:26:38 -0700 Subject: [PATCH 2/3] Harden frontend install locking and cache keys --- .../src/reflex_base/utils/decorator.py | 12 +- reflex/utils/frontend_lock.py | 49 +- reflex/utils/js_runtimes.py | 40 +- .../units/reflex_base/utils/test_decorator.py | 203 +++++++ tests/units/test_prerequisites.py | 502 +++--------------- tests/units/utils/test_frontend_lock.py | 350 ++++++++++++ 6 files changed, 709 insertions(+), 447 deletions(-) create mode 100644 tests/units/reflex_base/utils/test_decorator.py create mode 100644 tests/units/utils/test_frontend_lock.py diff --git a/packages/reflex-base/src/reflex_base/utils/decorator.py b/packages/reflex-base/src/reflex_base/utils/decorator.py index f33ce80e47a..565707a71b0 100644 --- a/packages/reflex-base/src/reflex_base/utils/decorator.py +++ b/packages/reflex-base/src/reflex_base/utils/decorator.py @@ -110,7 +110,17 @@ def _read_cached_procedure_file(cache_file: Path) -> tuple[str | None, object]: payload, value = pickle.loads(f.read()) if not isinstance(payload, str): return None, None - except (pickle.UnpicklingError, EOFError, TypeError, ValueError) as err: + except OSError: + # Permission and filesystem failures are operational errors, not + # evidence that the cache payload itself is corrupt. + raise + except MemoryError: + # Do not turn process resource exhaustion into an expensive retry. + raise + except Exception as err: + # Besides UnpicklingError, pickle may raise AttributeError, + # EOFError, ImportError, IndexError, or validation errors for + # malformed data. All mean the cache must be recomputed. logger.debug(f"Ignoring invalid procedure cache {cache_file}: {err}") else: return payload, value diff --git a/reflex/utils/frontend_lock.py b/reflex/utils/frontend_lock.py index 265158e5d8e..6db0196ebc7 100644 --- a/reflex/utils/frontend_lock.py +++ b/reflex/utils/frontend_lock.py @@ -16,7 +16,29 @@ _project_locks_guard = threading.Lock() _project_locks: dict[Path, threading.RLock] = {} _project_locks_held = threading.local() +# 50 milliseconds between contended Windows lock attempts. _WINDOWS_LOCK_RETRY_DELAY = 0.05 +_open_project_lock_files: set[BinaryIO] = set() + + +def _reset_project_locks_after_fork() -> None: + """Discard inherited lock ownership in a forked child process.""" + global _project_locks_guard, _project_locks_held + + # ``flock`` ownership follows the inherited open file description. Close + # the child's duplicate so a fresh open below blocks on the parent rather + # than inheriting or deadlocking against its own copy of the lock. + for lock_file in _open_project_lock_files: + with contextlib.suppress(OSError): + lock_file.close() + _open_project_lock_files.clear() + _project_locks.clear() + _project_locks_guard = threading.Lock() + _project_locks_held = threading.local() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_reset_project_locks_after_fork) def _acquire_project_file_lock(lock_file: BinaryIO) -> None: @@ -94,7 +116,9 @@ def frontend_project_lock() -> Iterator[None]: if thread_lock is None: thread_lock = _project_locks[lock_path] = threading.RLock() - with thread_lock: + owner_pid = os.getpid() + thread_lock.acquire() + try: held_paths = getattr(_project_locks_held, "paths", None) if held_paths is None: held_paths = _project_locks_held.paths = set() @@ -104,10 +128,23 @@ def frontend_project_lock() -> Iterator[None]: lock_path.parent.mkdir(parents=True, exist_ok=True) with lock_path.open("a+b") as lock_file: - _acquire_project_file_lock(lock_file) - held_paths.add(lock_path) + with _project_locks_guard: + _open_project_lock_files.add(lock_file) try: - yield + _acquire_project_file_lock(lock_file) + held_paths.add(lock_path) + try: + yield + finally: + if os.getpid() == owner_pid: + held_paths.remove(lock_path) + _release_project_file_lock(lock_file) finally: - held_paths.remove(lock_path) - _release_project_file_lock(lock_file) + if os.getpid() == owner_pid: + with _project_locks_guard: + _open_project_lock_files.discard(lock_file) + finally: + # A child forked inside the yielded transaction has fresh process-local + # state and must not release the inherited parent-side RLock. + if os.getpid() == owner_pid: + thread_lock.release() diff --git a/reflex/utils/js_runtimes.py b/reflex/utils/js_runtimes.py index 1983b3bd07e..9ef7447ca48 100644 --- a/reflex/utils/js_runtimes.py +++ b/reflex/utils/js_runtimes.py @@ -585,21 +585,24 @@ def _pinned_args_from_constants(deps: dict[str, str]) -> set[str]: def _frontend_packages_cache_payload( packages: set[str], - config: Config, + development_deps: set[str], + frozen_lockfile: bool, install_package_managers: Sequence[str], ) -> str: """Cache fingerprint for frontend package installs. Args: packages: Custom packages requested by the caller. - config: The active Reflex config. + development_deps: Development packages contributed by plugins. + frozen_lockfile: Whether installs must enforce the existing lockfile. install_package_managers: The package manager paths in priority order. Returns: Stable fingerprint string for the cached procedure. """ return ( - f"{sorted(packages)!r},{config.json()},{list(install_package_managers)!r}," + f"{sorted(packages)!r},{sorted(development_deps)!r},{frozen_lockfile!r}," + f"{list(install_package_managers)!r}," f"{sorted(constants.PackageJson.DEPENDENCIES.items())!r}," f"{sorted(constants.PackageJson.DEV_DEPENDENCIES.items())!r}," f"{sorted(constants.PackageJson.OVERRIDES.items())!r}" @@ -612,7 +615,8 @@ def _frontend_packages_cache_payload( ) def _install_frontend_packages( packages: set[str], - config: Config, + development_deps: set[str], + frozen_lockfile: bool, install_package_managers: Sequence[str], ): """Installs the base and custom frontend packages. @@ -637,12 +641,15 @@ def _install_frontend_packages( Args: packages: Custom packages requested by the caller (from ``Config.frontend_packages`` and inferred component imports). - config: The active Reflex config. + development_deps: Development packages contributed by plugins. + frozen_lockfile: Whether installs must enforce the existing lockfile. install_package_managers: The package manager paths in priority order (primary plus fallbacks). Example: - >>> install_frontend_packages({"react", "react-dom"}, get_config()) + >>> _install_frontend_packages( + ... {"react", "react-dom"}, set(), True, ("bun",) + ... ) """ env = ( { @@ -667,13 +674,6 @@ def _install_frontend_packages( env=env, ) - # Resolve plugin-contributed deps up front so we know the full needed - # set before deciding which entries in package.json are stale. - development_deps: set[str] = set() - for plugin in config.plugins: - development_deps.update(plugin.get_frontend_development_dependencies()) - packages.update(plugin.get_frontend_dependencies()) - wanted_dep_names = set(constants.PackageJson.DEPENDENCIES.keys()) | { _extract_package_name(p) for p in packages } @@ -717,7 +717,7 @@ def _install_frontend_packages( frontend_skeleton.get_web_lockfile_path(name).exists() for name in frontend_skeleton.LOCKFILE_NAMES ): - _run_initial_install(primary_package_manager, env, config.frozen_lockfile) + _run_initial_install(primary_package_manager, env, frozen_lockfile) # Framework overrides are withheld while the persisted package.json is # restored so the frozen install above sees exactly the file that produced @@ -790,5 +790,15 @@ def install_frontend_packages(packages: set[str], config: Config): get_nodejs_compatible_package_managers(raise_on_none=True) ) _sync_root_lockfiles_for_frontend_install() - _install_frontend_packages(set(packages), config, install_package_managers) + resolved_packages = set(packages) + development_deps: set[str] = set() + for plugin in config.plugins: + development_deps.update(plugin.get_frontend_development_dependencies()) + resolved_packages.update(plugin.get_frontend_dependencies()) + _install_frontend_packages( + resolved_packages, + development_deps, + config.frozen_lockfile, + install_package_managers, + ) frontend_skeleton.sync_web_lockfiles_to_root() diff --git a/tests/units/reflex_base/utils/test_decorator.py b/tests/units/reflex_base/utils/test_decorator.py new file mode 100644 index 00000000000..31859e7f22f --- /dev/null +++ b/tests/units/reflex_base/utils/test_decorator.py @@ -0,0 +1,203 @@ +import pickle +import tempfile +from pathlib import Path + +import pytest +from reflex_base.utils.decorator import cached_procedure + + +def test_cached_procedure(): + call_count = 0 + + temp_file = tempfile.mktemp() + + @cached_procedure( + cache_file_path=lambda: Path(temp_file), payload_fn=lambda: "constant" + ) + def _function_with_no_args(): + nonlocal call_count + call_count += 1 + + _function_with_no_args() + assert call_count == 1 + _function_with_no_args() + assert call_count == 1 + + call_count = 0 + + another_temp_file = tempfile.mktemp() + + @cached_procedure( + cache_file_path=lambda: Path(another_temp_file), + payload_fn=lambda *args, **kwargs: f"{repr(args), repr(kwargs)}", + ) + def _function_with_some_args(*args, **kwargs): + nonlocal call_count + call_count += 1 + + _function_with_some_args(1, y=2) + assert call_count == 1 + _function_with_some_args(1, y=2) + assert call_count == 1 + _function_with_some_args(100, y=300) + assert call_count == 2 + _function_with_some_args(100, y=300) + assert call_count == 2 + + call_count = 0 + + @cached_procedure( + cache_file_path=lambda: Path(tempfile.mktemp()), payload_fn=lambda: "constant" + ) + def _function_with_no_args_fn(): + nonlocal call_count + call_count += 1 + + _function_with_no_args_fn() + assert call_count == 1 + _function_with_no_args_fn() + assert call_count == 2 + + +def test_cached_procedure_treats_corrupt_file_as_miss(tmp_path: Path) -> None: + """A cache truncated by a killed process is recomputed and repaired.""" + cache_file = tmp_path / "procedure.cached" + cache_file.write_bytes(b"not a pickle") + call_count = 0 + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "payload") + def procedure() -> str: + nonlocal call_count + call_count += 1 + return "recomputed" + + assert procedure() == "recomputed" + assert procedure() == "recomputed" + assert call_count == 1 + assert pickle.loads(cache_file.read_bytes()) == ("payload", "recomputed") + + +def test_cached_procedure_treats_missing_pickle_global_as_miss( + tmp_path: Path, +) -> None: + """A pickle referring to an unavailable class is recomputed and repaired.""" + cache_file = tmp_path / "procedure.cached" + cache_file.write_bytes(b"cno_such_module\nthing\n.") + call_count = 0 + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "payload") + def procedure() -> str: + nonlocal call_count + call_count += 1 + return "recomputed" + + assert procedure() == "recomputed" + assert procedure() == "recomputed" + assert call_count == 1 + assert pickle.loads(cache_file.read_bytes()) == ("payload", "recomputed") + + +def test_cached_procedure_propagates_cache_io_errors( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An inaccessible cache is not mistaken for an ordinary cache miss.""" + cache_file = tmp_path / "procedure.cached" + cache_file.write_bytes(pickle.dumps(("payload", "cached"))) + call_count = 0 + original_open = Path.open + error = PermissionError("cache access denied") + + def deny_cache_read(path: Path, *args, **kwargs): + if path == cache_file: + raise error + return original_open(path, *args, **kwargs) + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "payload") + def procedure() -> str: + nonlocal call_count + call_count += 1 + return "recomputed" + + monkeypatch.setattr(Path, "open", deny_cache_read) + + with pytest.raises(PermissionError, match="cache access denied"): + procedure() + + assert call_count == 0 + + +def test_cached_procedure_propagates_memory_errors( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Resource exhaustion is not mistaken for malformed cache data.""" + cache_file = tmp_path / "procedure.cached" + cache_file.write_bytes(pickle.dumps(("payload", "cached"))) + call_count = 0 + error = MemoryError("out of memory") + + def fail_to_unpickle(_contents: bytes): + raise error + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "payload") + def procedure() -> str: + nonlocal call_count + call_count += 1 + return "recomputed" + + monkeypatch.setattr(pickle, "loads", fail_to_unpickle) + + with pytest.raises(MemoryError, match="out of memory"): + procedure() + + assert call_count == 0 + + +def test_cached_procedure_preserves_old_file_if_replace_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A failed cache commit leaves the previous complete pickle readable.""" + cache_file = tmp_path / "procedure.cached" + payload = "old" + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: payload) + def procedure() -> str: + return payload + + assert procedure() == "old" + old_cache = cache_file.read_bytes() + payload = "new" + original_replace = Path.replace + error = OSError("simulated replace failure") + + def fail_cache_replace(path: Path, target: Path) -> Path: + if target == cache_file: + raise error + return original_replace(path, target) + + monkeypatch.setattr(Path, "replace", fail_cache_replace) + + with pytest.raises(OSError, match="simulated replace failure"): + procedure() + + assert cache_file.read_bytes() == old_cache + assert pickle.loads(cache_file.read_bytes()) == ("old", "old") + assert list(tmp_path.glob(f".{cache_file.name}.*.tmp")) == [] + + +def test_cached_procedure_preserves_existing_symlink(tmp_path: Path) -> None: + """Atomic cache updates follow an existing cache symlink.""" + cache_target = tmp_path / "shared-procedure.cached" + cache_target.write_bytes(pickle.dumps(("old", "old"))) + cache_file = tmp_path / "procedure.cached" + try: + cache_file.symlink_to(cache_target.name) + except OSError as err: + pytest.skip(f"Cannot create symlink on this platform: {err}") + + @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "new") + def procedure() -> str: + return "new" + + assert procedure() == "new" + assert cache_file.is_symlink() + assert pickle.loads(cache_target.read_bytes()) == ("new", "new") diff --git a/tests/units/test_prerequisites.py b/tests/units/test_prerequisites.py index 759210eadf8..1d8c0928784 100644 --- a/tests/units/test_prerequisites.py +++ b/tests/units/test_prerequisites.py @@ -3,11 +3,10 @@ import os import pickle import shutil -import sys import tempfile import uuid from collections.abc import Callable, Generator -from contextlib import ExitStack, contextmanager +from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Protocol @@ -17,7 +16,6 @@ from reflex_base import constants from reflex_base.config import Config from reflex_base.utils import log -from reflex_base.utils.decorator import cached_procedure from reflex.reflex import cli from reflex.testing import chdir @@ -32,88 +30,6 @@ runner = CliRunner() -def _hold_frontend_project_lock( - project_dir: str, - acquired: Any, - release: Any, - *, - exit_without_cleanup: bool = False, - nested: bool = False, -) -> None: - """Hold a frontend project lock in a spawned process. - - Args: - project_dir: App root whose lock should be acquired. - acquired: Multiprocessing event set after acquisition. - release: Multiprocessing event that releases a normal holder. - exit_without_cleanup: Exit the process without running ``finally`` blocks. - nested: Acquire the same project lock twice before signalling. - """ - os.chdir(project_dir) - with ExitStack() as stack: - stack.enter_context(frontend_lock.frontend_project_lock()) - if nested: - stack.enter_context(frontend_lock.frontend_project_lock()) - acquired.set() - if exit_without_cleanup: - os._exit(0) - if not release.wait(20): - msg = "Timed out waiting to release frontend project lock" - raise TimeoutError(msg) - - -def _frontend_project_file_lock_is_contended(project_dir: Path) -> bool: - """Probe the OS-level project lock without blocking. - - Args: - project_dir: App root whose lock file should be probed. - - Returns: - Whether another process currently owns the project lock. - """ - lock_path = project_dir / constants.Dirs.FRONTEND_INSTALL_LOCK - with lock_path.open("r+b") as lock_file: - lock_file.seek(0) - if constants.IS_WINDOWS: - import msvcrt - - try: - msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] - lock_file.fileno(), - msvcrt.LK_NBLCK, # pyright: ignore[reportAttributeAccessIssue] - 1, - ) - except OSError as err: - if err.errno in { - frontend_lock.errno.EACCES, - frontend_lock.errno.EAGAIN, - frontend_lock.errno.EDEADLK, - }: - return True - raise - lock_file.seek(0) - msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] - lock_file.fileno(), - msvcrt.LK_UNLCK, # pyright: ignore[reportAttributeAccessIssue] - 1, - ) - return False - - import fcntl - - try: - fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - except OSError as err: - if err.errno in { - frontend_lock.errno.EACCES, - frontend_lock.errno.EAGAIN, - }: - return True - raise - fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) - return False - - def _install_frontend_packages_in_process( project_dir: str, attempting: Any, @@ -121,6 +37,7 @@ def _install_frontend_packages_in_process( release_package_manager: Any, active_installs: Any, max_active_installs: Any, + config_attribute_order: tuple[str, ...], ) -> None: """Run a deterministic fake frontend install in a spawned process. @@ -131,6 +48,7 @@ def _install_frontend_packages_in_process( release_package_manager: Event allowing the fake package manager to finish. active_installs: Shared count of active package-manager calls. max_active_installs: Shared maximum active package-manager calls. + config_attribute_order: Forced iteration order for internal Config metadata. """ os.chdir(project_dir) constants.PackageJson.DEPENDENCIES = {} # pyright: ignore[reportAttributeAccessIssue] @@ -140,6 +58,15 @@ def _install_frontend_packages_in_process( "bun", ) + class OrderedAttributeSet(set[str]): + """Set with a forced iteration order to model distinct process hashes.""" + + def __iter__(self): + return iter(config_attribute_order) + + config = Config(app_name="test") + config._non_default_attributes = OrderedAttributeSet(config_attribute_order) + def run_package_manager(args, **kwargs) -> None: """Block a fake package-manager add so a competing process can contend.""" if "add" not in args: @@ -164,7 +91,7 @@ def run_package_manager(args, **kwargs) -> None: js_runtimes.processes.run_process_with_fallbacks = run_package_manager attempting.set() - js_runtimes.install_frontend_packages({"race-pkg@1.0.0"}, Config(app_name="test")) + js_runtimes.install_frontend_packages({"race-pkg@1.0.0"}, config) def _patch_web_dir(monkeypatch: pytest.MonkeyPatch, web_dir: Path): @@ -323,207 +250,36 @@ def stub(*args, **kwargs): "sync_web_lockfiles_to_root", record("sync-web"), ) + config = Config(app_name="test") + + class FakePlugin: + def get_frontend_development_dependencies(self): + assert lock_held + calls.append("resolve-dev-dependencies") + return set() + + def get_frontend_dependencies(self): + assert lock_held + calls.append("resolve-dependencies") + return set() + + monkeypatch.setattr(config, "plugins", [FakePlugin()]) with chdir(tmp_path): - js_runtimes.install_frontend_packages(set(), Config(app_name="test")) + js_runtimes.install_frontend_packages(set(), config) assert calls == [ "lock-enter", "select-manager", "sync-root", + "resolve-dev-dependencies", + "resolve-dependencies", "install", "sync-web", "lock-exit", ] -def test_frontend_project_lock_is_reentrant_and_ignored(tmp_path: Path) -> None: - """Nested use in one thread succeeds and its stable file is gitignored.""" - ctx = multiprocessing.get_context("spawn") - acquired = ctx.Event() - release = ctx.Event() - process = ctx.Process( - target=_hold_frontend_project_lock, - args=(str(tmp_path), acquired, release), - kwargs={"nested": True}, - ) - - nested_lock_acquired = False - started = False - try: - process.start() - started = True - nested_lock_acquired = acquired.wait(20) - finally: - release.set() - if started: - process.join(20) - if process.is_alive(): - process.terminate() - process.join(5) - - assert nested_lock_acquired - assert process.exitcode == 0 - assert (tmp_path / constants.Dirs.FRONTEND_INSTALL_LOCK).exists() - assert constants.Dirs.FRONTEND_INSTALL_LOCK in constants.GitIgnore.DEFAULTS - - -def test_frontend_project_lock_releases_after_exception(tmp_path: Path) -> None: - """An exception in a transaction does not strand its project lock.""" - error = RuntimeError("failed install") - with chdir(tmp_path): - with ( - pytest.raises(RuntimeError, match="failed install"), - frontend_lock.frontend_project_lock(), - ): - raise error - - with frontend_lock.frontend_project_lock(): - assert Path(constants.Dirs.FRONTEND_INSTALL_LOCK).exists() - - -def test_frontend_project_lock_blocks_same_project(tmp_path: Path) -> None: - """A held project lock is positively contended by another process.""" - ctx = multiprocessing.get_context("spawn") - acquired = ctx.Event() - release = ctx.Event() - process = ctx.Process( - target=_hold_frontend_project_lock, - args=(str(tmp_path), acquired, release), - ) - - lock_was_acquired = False - lock_was_contended = False - try: - process.start() - lock_was_acquired = acquired.wait(20) - if lock_was_acquired: - lock_was_contended = _frontend_project_file_lock_is_contended(tmp_path) - finally: - release.set() - process.join(20) - if process.is_alive(): - process.terminate() - process.join(5) - - assert lock_was_acquired - assert lock_was_contended - assert process.exitcode == 0 - - -def test_frontend_project_file_lock_retries_windows_contention( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - mocker, -) -> None: - """The Windows path waits on contention and unlocks the same byte range.""" - fake_msvcrt = mocker.Mock() - fake_msvcrt.LK_NBLCK = 1 - fake_msvcrt.LK_UNLCK = 2 - contention = OSError(frontend_lock.errno.EACCES, "lock is held") - fake_msvcrt.locking.side_effect = [contention, contention, None, None] - sleep = mocker.patch.object(frontend_lock.time, "sleep") - monkeypatch.setattr(constants, "IS_WINDOWS", True) - monkeypatch.setitem(sys.modules, "msvcrt", fake_msvcrt) - lock_path = tmp_path / "lock" - - with lock_path.open("w+b") as lock_file: - frontend_lock._acquire_project_file_lock(lock_file) - frontend_lock._release_project_file_lock(lock_file) - - assert [call.args[1:] for call in fake_msvcrt.locking.call_args_list] == [ - (fake_msvcrt.LK_NBLCK, 1), - (fake_msvcrt.LK_NBLCK, 1), - (fake_msvcrt.LK_NBLCK, 1), - (fake_msvcrt.LK_UNLCK, 1), - ] - assert sleep.call_args_list == [ - mocker.call(frontend_lock._WINDOWS_LOCK_RETRY_DELAY), - mocker.call(frontend_lock._WINDOWS_LOCK_RETRY_DELAY), - ] - - -def test_frontend_project_locks_are_scoped_per_project(tmp_path: Path) -> None: - """Two separate app roots can hold their frontend locks concurrently.""" - ctx = multiprocessing.get_context("spawn") - project_a = tmp_path / "project-a" - project_b = tmp_path / "project-b" - project_a.mkdir() - project_b.mkdir() - acquired_a = ctx.Event() - acquired_b = ctx.Event() - release = ctx.Event() - processes = [ - ctx.Process( - target=_hold_frontend_project_lock, - args=(str(project_a), acquired_a, release), - ), - ctx.Process( - target=_hold_frontend_project_lock, - args=(str(project_b), acquired_b, release), - ), - ] - - for process in processes: - process.start() - try: - assert acquired_a.wait(20) - assert acquired_b.wait(20) - finally: - release.set() - for process in processes: - process.join(20) - if process.is_alive(): - process.terminate() - process.join(5) - - assert [process.exitcode for process in processes] == [0, 0] - - -def test_frontend_project_lock_recovers_after_crashed_process(tmp_path: Path) -> None: - """The OS releases the project lock when its owning process exits abruptly.""" - ctx = multiprocessing.get_context("spawn") - crashed_acquired = ctx.Event() - unused_release = ctx.Event() - crashed = ctx.Process( - target=_hold_frontend_project_lock, - args=(str(tmp_path), crashed_acquired, unused_release), - kwargs={"exit_without_cleanup": True}, - ) - crashed_lock_acquired = False - crashed_started = False - try: - crashed.start() - crashed_started = True - crashed_lock_acquired = crashed_acquired.wait(20) - crashed.join(20) - finally: - if crashed_started and crashed.is_alive(): - crashed.terminate() - crashed.join(5) - - assert crashed_lock_acquired - assert crashed.exitcode == 0 - - recovered_acquired = ctx.Event() - release = ctx.Event() - recovered = ctx.Process( - target=_hold_frontend_project_lock, - args=(str(tmp_path), recovered_acquired, release), - ) - recovered.start() - try: - assert recovered_acquired.wait(20) - finally: - release.set() - recovered.join(20) - if recovered.is_alive(): - recovered.terminate() - recovered.join(5) - - assert recovered.exitcode == 0 - - def test_concurrent_frontend_installs_share_completed_cache( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -551,6 +307,7 @@ def test_concurrent_frontend_installs_share_completed_cache( release_package_manager, active_installs, max_active_installs, + ("loglevel", "app_name"), ), ) second = ctx.Process( @@ -562,6 +319,7 @@ def test_concurrent_frontend_installs_share_completed_cache( release_package_manager, active_installs, max_active_installs, + ("app_name", "loglevel"), ), ) @@ -1067,6 +825,51 @@ def run_package_manager(args, **kwargs): return calls +def test_frontend_package_cache_uses_resolved_plugin_dependencies( + install_packages_env: InstallPackagesEnv, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Set iteration order in plugin config does not perturb the cache key.""" + env = install_packages_env + calls = _record_calls(env) + + class OrderedStringSet(set[str]): + def __init__(self, values: tuple[str, ...]): + super().__init__(values) + self.values = values + + def __iter__(self): + return iter(self.values) + + @dataclass + class FakePlugin: + dependencies: OrderedStringSet + + def get_frontend_dependencies(self): + return self.dependencies + + def get_frontend_development_dependencies(self): + return set() + + monkeypatch.setattr( + env.config, + "plugins", + [FakePlugin(OrderedStringSet(("plugin-b", "plugin-a")))], + ) + first_config_json = env.config.json() + env.install() + monkeypatch.setattr( + env.config, + "plugins", + [FakePlugin(OrderedStringSet(("plugin-a", "plugin-b")))], + ) + second_config_json = env.config.json() + env.install() + + assert first_config_json != second_config_json + assert len([call for call in calls if "add" in call]) == 1 + + def test_install_frontend_packages_pinned_packages_single_call( install_packages_env: InstallPackagesEnv, ): @@ -2171,157 +1974,6 @@ def test_extract_package_name(): assert js_runtimes._extract_package_name("@scope/pkg@1.2.3") == "@scope/pkg" -def test_cached_procedure(): - call_count = 0 - - temp_file = tempfile.mktemp() - - @cached_procedure( - cache_file_path=lambda: Path(temp_file), payload_fn=lambda: "constant" - ) - def _function_with_no_args(): - nonlocal call_count - call_count += 1 - - _function_with_no_args() - assert call_count == 1 - _function_with_no_args() - assert call_count == 1 - - call_count = 0 - - another_temp_file = tempfile.mktemp() - - @cached_procedure( - cache_file_path=lambda: Path(another_temp_file), - payload_fn=lambda *args, **kwargs: f"{repr(args), repr(kwargs)}", - ) - def _function_with_some_args(*args, **kwargs): - nonlocal call_count - call_count += 1 - - _function_with_some_args(1, y=2) - assert call_count == 1 - _function_with_some_args(1, y=2) - assert call_count == 1 - _function_with_some_args(100, y=300) - assert call_count == 2 - _function_with_some_args(100, y=300) - assert call_count == 2 - - call_count = 0 - - @cached_procedure( - cache_file_path=lambda: Path(tempfile.mktemp()), payload_fn=lambda: "constant" - ) - def _function_with_no_args_fn(): - nonlocal call_count - call_count += 1 - - _function_with_no_args_fn() - assert call_count == 1 - _function_with_no_args_fn() - assert call_count == 2 - - -def test_cached_procedure_treats_corrupt_file_as_miss(tmp_path: Path) -> None: - """A cache truncated by a killed process is recomputed and repaired.""" - cache_file = tmp_path / "procedure.cached" - cache_file.write_bytes(b"not a pickle") - call_count = 0 - - @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "payload") - def procedure() -> str: - nonlocal call_count - call_count += 1 - return "recomputed" - - assert procedure() == "recomputed" - assert procedure() == "recomputed" - assert call_count == 1 - assert pickle.loads(cache_file.read_bytes()) == ("payload", "recomputed") - - -def test_cached_procedure_propagates_cache_io_errors( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - """An inaccessible cache is not mistaken for an ordinary cache miss.""" - cache_file = tmp_path / "procedure.cached" - cache_file.write_bytes(pickle.dumps(("payload", "cached"))) - call_count = 0 - original_open = Path.open - error = PermissionError("cache access denied") - - def deny_cache_read(path: Path, *args, **kwargs): - if path == cache_file: - raise error - return original_open(path, *args, **kwargs) - - @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "payload") - def procedure() -> str: - nonlocal call_count - call_count += 1 - return "recomputed" - - monkeypatch.setattr(Path, "open", deny_cache_read) - - with pytest.raises(PermissionError, match="cache access denied"): - procedure() - - assert call_count == 0 - - -def test_cached_procedure_preserves_old_file_if_replace_fails( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - """A failed cache commit leaves the previous complete pickle readable.""" - cache_file = tmp_path / "procedure.cached" - payload = "old" - - @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: payload) - def procedure() -> str: - return payload - - assert procedure() == "old" - old_cache = cache_file.read_bytes() - payload = "new" - original_replace = Path.replace - error = OSError("simulated replace failure") - - def fail_cache_replace(path: Path, target: Path) -> Path: - if target == cache_file: - raise error - return original_replace(path, target) - - monkeypatch.setattr(Path, "replace", fail_cache_replace) - - with pytest.raises(OSError, match="simulated replace failure"): - procedure() - - assert cache_file.read_bytes() == old_cache - assert pickle.loads(cache_file.read_bytes()) == ("old", "old") - assert list(tmp_path.glob(f".{cache_file.name}.*.tmp")) == [] - - -def test_cached_procedure_preserves_existing_symlink(tmp_path: Path) -> None: - """Atomic cache updates follow an existing cache symlink.""" - cache_target = tmp_path / "shared-procedure.cached" - cache_target.write_bytes(pickle.dumps(("old", "old"))) - cache_file = tmp_path / "procedure.cached" - try: - cache_file.symlink_to(cache_target.name) - except OSError as err: - pytest.skip(f"Cannot create symlink on this platform: {err}") - - @cached_procedure(cache_file_path=lambda: cache_file, payload_fn=lambda: "new") - def procedure() -> str: - return "new" - - assert procedure() == "new" - assert cache_file.is_symlink() - assert pickle.loads(cache_target.read_bytes()) == ("new", "new") - - def test_get_cpu_info(): cpu_info = get_cpu_info() assert cpu_info is not None diff --git a/tests/units/utils/test_frontend_lock.py b/tests/units/utils/test_frontend_lock.py new file mode 100644 index 00000000000..4fe99cd9fd4 --- /dev/null +++ b/tests/units/utils/test_frontend_lock.py @@ -0,0 +1,350 @@ +import multiprocessing +import os +import sys +from contextlib import ExitStack +from pathlib import Path +from typing import Any + +import pytest +from reflex_base import constants + +from reflex.testing import chdir +from reflex.utils import frontend_lock + + +def _hold_frontend_project_lock( + project_dir: str, + acquired: Any, + release: Any, + *, + exit_without_cleanup: bool = False, + nested: bool = False, +) -> None: + """Hold a frontend project lock in a spawned process. + + Args: + project_dir: App root whose lock should be acquired. + acquired: Multiprocessing event set after acquisition. + release: Multiprocessing event that releases a normal holder. + exit_without_cleanup: Exit the process without running ``finally`` blocks. + nested: Acquire the same project lock twice before signalling. + """ + os.chdir(project_dir) + with ExitStack() as stack: + stack.enter_context(frontend_lock.frontend_project_lock()) + if nested: + stack.enter_context(frontend_lock.frontend_project_lock()) + acquired.set() + if exit_without_cleanup: + os._exit(0) + if not release.wait(20): + msg = "Timed out waiting to release frontend project lock" + raise TimeoutError(msg) + + +def _fork_while_holding_frontend_project_lock( + project_dir: str, + child_attempting: Any, + child_acquired: Any, + release_parent: Any, +) -> None: + """Fork while locked and have the child attempt the same project lock. + + Args: + project_dir: App root whose lock should be acquired. + child_attempting: Event set before the child tries to acquire the lock. + child_acquired: Event set after the child acquires the lock. + release_parent: Event allowing the parent to release its lock. + """ + os.chdir(project_dir) + child_pid = None + with frontend_lock.frontend_project_lock(): + child_pid = os.fork() + if child_pid == 0: + try: + child_attempting.set() + with frontend_lock.frontend_project_lock(): + child_acquired.set() + finally: + os._exit(0) + + if not release_parent.wait(20): + msg = "Timed out waiting to release parent frontend project lock" + raise TimeoutError(msg) + + if child_pid is not None: + _, status = os.waitpid(child_pid, 0) + if status != 0: + msg = f"Forked lock contender exited with status {status}" + raise RuntimeError(msg) + + +def _frontend_project_file_lock_is_contended(project_dir: Path) -> bool: + """Probe the OS-level project lock without blocking. + + Args: + project_dir: App root whose lock file should be probed. + + Returns: + Whether another process currently owns the project lock. + """ + lock_path = project_dir / constants.Dirs.FRONTEND_INSTALL_LOCK + with lock_path.open("r+b") as lock_file: + lock_file.seek(0) + if constants.IS_WINDOWS: + import msvcrt + + try: + msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] + lock_file.fileno(), + msvcrt.LK_NBLCK, # pyright: ignore[reportAttributeAccessIssue] + 1, + ) + except OSError as err: + if err.errno in { + frontend_lock.errno.EACCES, + frontend_lock.errno.EAGAIN, + frontend_lock.errno.EDEADLK, + }: + return True + raise + lock_file.seek(0) + msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] + lock_file.fileno(), + msvcrt.LK_UNLCK, # pyright: ignore[reportAttributeAccessIssue] + 1, + ) + return False + + import fcntl + + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as err: + if err.errno in { + frontend_lock.errno.EACCES, + frontend_lock.errno.EAGAIN, + }: + return True + raise + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + return False + + +def test_frontend_project_lock_is_reentrant_and_ignored(tmp_path: Path) -> None: + """Nested use in one thread succeeds and its stable file is gitignored.""" + ctx = multiprocessing.get_context("spawn") + acquired = ctx.Event() + release = ctx.Event() + process = ctx.Process( + target=_hold_frontend_project_lock, + args=(str(tmp_path), acquired, release), + kwargs={"nested": True}, + ) + + nested_lock_acquired = False + started = False + try: + process.start() + started = True + nested_lock_acquired = acquired.wait(20) + finally: + release.set() + if started: + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert nested_lock_acquired + assert process.exitcode == 0 + assert (tmp_path / constants.Dirs.FRONTEND_INSTALL_LOCK).exists() + assert constants.Dirs.FRONTEND_INSTALL_LOCK in constants.GitIgnore.DEFAULTS + + +def test_frontend_project_lock_releases_after_exception(tmp_path: Path) -> None: + """An exception in a transaction does not strand its project lock.""" + error = RuntimeError("failed install") + with chdir(tmp_path): + with ( + pytest.raises(RuntimeError, match="failed install"), + frontend_lock.frontend_project_lock(), + ): + raise error + + with frontend_lock.frontend_project_lock(): + assert Path(constants.Dirs.FRONTEND_INSTALL_LOCK).exists() + + +def test_frontend_project_lock_blocks_same_project(tmp_path: Path) -> None: + """A held project lock is positively contended by another process.""" + ctx = multiprocessing.get_context("spawn") + acquired = ctx.Event() + release = ctx.Event() + process = ctx.Process( + target=_hold_frontend_project_lock, + args=(str(tmp_path), acquired, release), + ) + + lock_was_acquired = False + lock_was_contended = False + try: + process.start() + lock_was_acquired = acquired.wait(20) + if lock_was_acquired: + lock_was_contended = _frontend_project_file_lock_is_contended(tmp_path) + finally: + release.set() + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert lock_was_acquired + assert lock_was_contended + assert process.exitcode == 0 + + +@pytest.mark.skipif(constants.IS_WINDOWS, reason="os.fork is unavailable on Windows") +def test_forked_child_reacquires_frontend_project_lock(tmp_path: Path) -> None: + """A forked child cannot inherit reentrant ownership from its parent.""" + ctx = multiprocessing.get_context("spawn") + child_attempting = ctx.Event() + child_acquired = ctx.Event() + release_parent = ctx.Event() + process = ctx.Process( + target=_fork_while_holding_frontend_project_lock, + args=(str(tmp_path), child_attempting, child_acquired, release_parent), + ) + + child_started = False + child_acquired_while_parent_held = False + try: + process.start() + child_started = child_attempting.wait(20) + if child_started: + child_acquired_while_parent_held = child_acquired.wait(1) + finally: + release_parent.set() + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert child_started + assert not child_acquired_while_parent_held + assert child_acquired.is_set() + assert process.exitcode == 0 + + +def test_frontend_project_file_lock_retries_windows_contention( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mocker, +) -> None: + """The Windows path waits on contention and unlocks the same byte range.""" + fake_msvcrt = mocker.Mock() + fake_msvcrt.LK_NBLCK = 1 + fake_msvcrt.LK_UNLCK = 2 + contention = OSError(frontend_lock.errno.EACCES, "lock is held") + fake_msvcrt.locking.side_effect = [contention, contention, None, None] + sleep = mocker.patch.object(frontend_lock.time, "sleep") + monkeypatch.setattr(constants, "IS_WINDOWS", True) + monkeypatch.setitem(sys.modules, "msvcrt", fake_msvcrt) + lock_path = tmp_path / "lock" + + with lock_path.open("w+b") as lock_file: + frontend_lock._acquire_project_file_lock(lock_file) + frontend_lock._release_project_file_lock(lock_file) + + assert [call.args[1:] for call in fake_msvcrt.locking.call_args_list] == [ + (fake_msvcrt.LK_NBLCK, 1), + (fake_msvcrt.LK_NBLCK, 1), + (fake_msvcrt.LK_NBLCK, 1), + (fake_msvcrt.LK_UNLCK, 1), + ] + assert sleep.call_args_list == [ + mocker.call(frontend_lock._WINDOWS_LOCK_RETRY_DELAY), + mocker.call(frontend_lock._WINDOWS_LOCK_RETRY_DELAY), + ] + + +def test_frontend_project_locks_are_scoped_per_project(tmp_path: Path) -> None: + """Two separate app roots can hold their frontend locks concurrently.""" + ctx = multiprocessing.get_context("spawn") + project_a = tmp_path / "project-a" + project_b = tmp_path / "project-b" + project_a.mkdir() + project_b.mkdir() + acquired_a = ctx.Event() + acquired_b = ctx.Event() + release = ctx.Event() + processes = [ + ctx.Process( + target=_hold_frontend_project_lock, + args=(str(project_a), acquired_a, release), + ), + ctx.Process( + target=_hold_frontend_project_lock, + args=(str(project_b), acquired_b, release), + ), + ] + + for process in processes: + process.start() + try: + assert acquired_a.wait(20) + assert acquired_b.wait(20) + finally: + release.set() + for process in processes: + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert [process.exitcode for process in processes] == [0, 0] + + +def test_frontend_project_lock_recovers_after_crashed_process(tmp_path: Path) -> None: + """The OS releases the project lock when its owning process exits abruptly.""" + ctx = multiprocessing.get_context("spawn") + crashed_acquired = ctx.Event() + unused_release = ctx.Event() + crashed = ctx.Process( + target=_hold_frontend_project_lock, + args=(str(tmp_path), crashed_acquired, unused_release), + kwargs={"exit_without_cleanup": True}, + ) + crashed_lock_acquired = False + crashed_started = False + try: + crashed.start() + crashed_started = True + crashed_lock_acquired = crashed_acquired.wait(20) + crashed.join(20) + finally: + if crashed_started and crashed.is_alive(): + crashed.terminate() + crashed.join(5) + + assert crashed_lock_acquired + assert crashed.exitcode == 0 + + recovered_acquired = ctx.Event() + release = ctx.Event() + recovered = ctx.Process( + target=_hold_frontend_project_lock, + args=(str(tmp_path), recovered_acquired, release), + ) + recovered.start() + try: + assert recovered_acquired.wait(20) + finally: + release.set() + recovered.join(20) + if recovered.is_alive(): + recovered.terminate() + recovered.join(5) + + assert recovered.exitcode == 0 From 7c6ffe56e97016fb0c918dd95f6b73297e93edd2 Mon Sep 17 00:00:00 2001 From: Alek Date: Sun, 6 Sep 2026 01:48:16 -0700 Subject: [PATCH 3/3] Make frontend lock cleanup fork-safe --- reflex/utils/frontend_lock.py | 105 +++++++++----- tests/units/utils/test_frontend_lock.py | 178 +++++++++++++++++++++++- 2 files changed, 239 insertions(+), 44 deletions(-) diff --git a/reflex/utils/frontend_lock.py b/reflex/utils/frontend_lock.py index 6db0196ebc7..730cf6177fb 100644 --- a/reflex/utils/frontend_lock.py +++ b/reflex/utils/frontend_lock.py @@ -9,7 +9,6 @@ import time from collections.abc import Iterator from pathlib import Path -from typing import BinaryIO from reflex_base import constants @@ -18,49 +17,65 @@ _project_locks_held = threading.local() # 50 milliseconds between contended Windows lock attempts. _WINDOWS_LOCK_RETRY_DELAY = 0.05 -_open_project_lock_files: set[BinaryIO] = set() +_open_project_lock_fds: set[int] = set() +_project_lock_fds_guard = threading.RLock() + + +def _prepare_project_locks_for_fork() -> None: + """Prevent a fork from splitting an open/close FD transition.""" + _project_lock_fds_guard.acquire() + + +def _resume_project_locks_after_fork() -> None: + """Release the parent's fork transition guard.""" + _project_lock_fds_guard.release() def _reset_project_locks_after_fork() -> None: """Discard inherited lock ownership in a forked child process.""" - global _project_locks_guard, _project_locks_held + global _project_lock_fds_guard, _project_locks_guard, _project_locks_held # ``flock`` ownership follows the inherited open file description. Close # the child's duplicate so a fresh open below blocks on the parent rather # than inheriting or deadlocking against its own copy of the lock. - for lock_file in _open_project_lock_files: + # Only raw descriptor operations are safe here: buffered file objects may + # retain a lock owned by a different thread that vanished during the fork. + for lock_fd in _open_project_lock_fds: with contextlib.suppress(OSError): - lock_file.close() - _open_project_lock_files.clear() + os.close(lock_fd) + _open_project_lock_fds.clear() _project_locks.clear() + _project_lock_fds_guard = threading.RLock() _project_locks_guard = threading.Lock() _project_locks_held = threading.local() if hasattr(os, "register_at_fork"): - os.register_at_fork(after_in_child=_reset_project_locks_after_fork) + os.register_at_fork( + before=_prepare_project_locks_for_fork, + after_in_parent=_resume_project_locks_after_fork, + after_in_child=_reset_project_locks_after_fork, + ) -def _acquire_project_file_lock(lock_file: BinaryIO) -> None: - """Acquire an advisory lock on an open project lock file. +def _acquire_project_file_lock(lock_fd: int) -> None: + """Acquire an advisory lock on an open project lock file descriptor. Args: - lock_file: Binary file kept open for the lifetime of the lock. + lock_fd: Raw descriptor kept open for the lifetime of the lock. """ - lock_file.seek(0, os.SEEK_END) - if lock_file.tell() == 0: - lock_file.write(b"\0") - lock_file.flush() - lock_file.seek(0) + if os.lseek(lock_fd, 0, os.SEEK_END) == 0: + os.write(lock_fd, b"\0") + os.lseek(lock_fd, 0, os.SEEK_SET) if constants.IS_WINDOWS: import msvcrt while True: try: - lock_file.seek(0) + os.lseek(lock_fd, 0, os.SEEK_SET) msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] - lock_file.fileno(), + lock_fd, msvcrt.LK_NBLCK, # pyright: ignore[reportAttributeAccessIssue] 1, ) @@ -73,21 +88,21 @@ def _acquire_project_file_lock(lock_file: BinaryIO) -> None: import fcntl - fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + fcntl.flock(lock_fd, fcntl.LOCK_EX) -def _release_project_file_lock(lock_file: BinaryIO) -> None: - """Release an advisory lock on an open project lock file. +def _release_project_file_lock(lock_fd: int) -> None: + """Release an advisory lock on an open project lock file descriptor. Args: - lock_file: Binary file previously passed to the acquire helper. + lock_fd: Raw descriptor previously passed to the acquire helper. """ - lock_file.seek(0) + os.lseek(lock_fd, 0, os.SEEK_SET) if constants.IS_WINDOWS: import msvcrt msvcrt.locking( # pyright: ignore[reportAttributeAccessIssue] - lock_file.fileno(), + lock_fd, msvcrt.LK_UNLCK, # pyright: ignore[reportAttributeAccessIssue] 1, ) @@ -95,7 +110,7 @@ def _release_project_file_lock(lock_file: BinaryIO) -> None: import fcntl - fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + fcntl.flock(lock_fd, fcntl.LOCK_UN) @contextlib.contextmanager @@ -127,22 +142,38 @@ def frontend_project_lock() -> Iterator[None]: return lock_path.parent.mkdir(parents=True, exist_ok=True) - with lock_path.open("a+b") as lock_file: - with _project_locks_guard: - _open_project_lock_files.add(lock_file) + with _project_lock_fds_guard: + lock_fd = os.open( + lock_path, + os.O_RDWR | os.O_CREAT | getattr(os, "O_BINARY", 0), + 0o666, + ) try: - _acquire_project_file_lock(lock_file) - held_paths.add(lock_path) + _open_project_lock_fds.add(lock_fd) + except BaseException: + os.close(lock_fd) + raise + lock_acquired = False + path_held = False + try: + _acquire_project_file_lock(lock_fd) + lock_acquired = True + held_paths.add(lock_path) + path_held = True + yield + finally: + if os.getpid() == owner_pid: try: - yield + try: + if path_held: + held_paths.remove(lock_path) + finally: + if lock_acquired: + _release_project_file_lock(lock_fd) finally: - if os.getpid() == owner_pid: - held_paths.remove(lock_path) - _release_project_file_lock(lock_file) - finally: - if os.getpid() == owner_pid: - with _project_locks_guard: - _open_project_lock_files.discard(lock_file) + with _project_lock_fds_guard: + _open_project_lock_fds.discard(lock_fd) + os.close(lock_fd) finally: # A child forked inside the yielded transaction has fresh process-local # state and must not release the inherited parent-side RLock. diff --git a/tests/units/utils/test_frontend_lock.py b/tests/units/utils/test_frontend_lock.py index 4fe99cd9fd4..2857a42aabd 100644 --- a/tests/units/utils/test_frontend_lock.py +++ b/tests/units/utils/test_frontend_lock.py @@ -1,7 +1,10 @@ import multiprocessing import os +import signal import sys -from contextlib import ExitStack +import threading +import time +from contextlib import ExitStack, suppress from pathlib import Path from typing import Any @@ -12,6 +15,35 @@ from reflex.utils import frontend_lock +def _wait_for_forked_child(child_pid: int) -> int: + """Wait a bounded time for a raw-fork child, killing and reaping on timeout. + + Args: + child_pid: PID returned by ``os.fork`` in the parent. + + Returns: + The child status returned by ``os.waitpid``. + + Raises: + TimeoutError: If the child does not exit within ten seconds. + """ + # Ten seconds gives normal child lock handoff ample time while bounding the + # deadlock regression that these tests are specifically designed to catch. + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + waited_pid, status = os.waitpid(child_pid, os.WNOHANG) + if waited_pid == child_pid: + return status + # Poll every 50 milliseconds without blocking the test process. + time.sleep(0.05) + + with suppress(ProcessLookupError): + os.kill(child_pid, signal.SIGKILL) + os.waitpid(child_pid, 0) + msg = f"Forked lock contender {child_pid} did not exit" + raise TimeoutError(msg) + + def _hold_frontend_project_lock( project_dir: str, acquired: Any, @@ -61,22 +93,95 @@ def _fork_while_holding_frontend_project_lock( with frontend_lock.frontend_project_lock(): child_pid = os.fork() if child_pid == 0: + exit_code = 0 try: child_attempting.set() with frontend_lock.frontend_project_lock(): child_acquired.set() - finally: - os._exit(0) + except BaseException: + exit_code = 1 + os._exit(exit_code) if not release_parent.wait(20): msg = "Timed out waiting to release parent frontend project lock" raise TimeoutError(msg) if child_pid is not None: - _, status = os.waitpid(child_pid, 0) + status = _wait_for_forked_child(child_pid) + if status != 0: + msg = f"Forked lock contender exited with status {status}" + raise RuntimeError(msg) + + +def _fork_while_another_thread_holds_frontend_project_lock( + project_dir: str, + child_attempting: Any, + child_acquired: Any, + release_parent: Any, +) -> None: + """Fork while a background thread owns the project lock. + + Args: + project_dir: App root whose lock should be acquired. + child_attempting: Event set before the child tries to acquire the lock. + child_acquired: Event set after the child acquires the lock. + release_parent: Event allowing the parent thread to release its lock. + """ + os.chdir(project_dir) + holder_acquired = threading.Event() + release_holder = threading.Event() + holder_errors: list[BaseException] = [] + + def hold_lock() -> None: + try: + with frontend_lock.frontend_project_lock(): + holder_acquired.set() + if not release_holder.wait(20): + msg = "Timed out waiting to release background lock holder" + holder_errors.append(TimeoutError(msg)) + except BaseException as err: + holder_errors.append(err) + + holder = threading.Thread(target=hold_lock) + holder.start() + child_pid = None + worker_error: BaseException | None = None + try: + if not holder_acquired.wait(20): + msg = "Timed out waiting for background thread to acquire project lock" + worker_error = TimeoutError(msg) + else: + child_pid = os.fork() + if child_pid == 0: + exit_code = 0 + try: + child_attempting.set() + with frontend_lock.frontend_project_lock(): + child_acquired.set() + except BaseException: + exit_code = 1 + os._exit(exit_code) + + if not release_parent.wait(20): + msg = "Timed out waiting to release background lock holder" + worker_error = TimeoutError(msg) + finally: + release_holder.set() + holder.join(20) + + if holder.is_alive(): + msg = "Background lock holder did not exit" + raise TimeoutError(msg) + if holder_errors: + msg = "Background lock holder failed" + raise RuntimeError(msg) from holder_errors[0] + if child_pid is not None: + status = _wait_for_forked_child(child_pid) if status != 0: msg = f"Forked lock contender exited with status {status}" raise RuntimeError(msg) + if worker_error is not None: + raise worker_error def _frontend_project_file_lock_is_contended(project_dir: Path) -> bool: @@ -176,6 +281,24 @@ def test_frontend_project_lock_releases_after_exception(tmp_path: Path) -> None: assert Path(constants.Dirs.FRONTEND_INSTALL_LOCK).exists() +def test_frontend_project_lock_uses_raw_file_descriptor( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Public locking never creates a buffered file object for child cleanup.""" + error = AssertionError("Path.open must not be used for the project lock") + + def fail_buffered_open(*args, **kwargs): + raise error + + monkeypatch.setattr(Path, "open", fail_buffered_open) + + with chdir(tmp_path), frontend_lock.frontend_project_lock(): + assert frontend_lock._open_project_lock_fds + assert all( + isinstance(lock_fd, int) for lock_fd in frontend_lock._open_project_lock_fds + ) + + def test_frontend_project_lock_blocks_same_project(tmp_path: Path) -> None: """A held project lock is positively contended by another process.""" ctx = multiprocessing.get_context("spawn") @@ -237,6 +360,40 @@ def test_forked_child_reacquires_frontend_project_lock(tmp_path: Path) -> None: assert process.exitcode == 0 +@pytest.mark.skipif(constants.IS_WINDOWS, reason="os.fork is unavailable on Windows") +def test_forked_child_reacquires_lock_owned_by_another_thread( + tmp_path: Path, +) -> None: + """A child resets lock state inherited from a vanished background thread.""" + ctx = multiprocessing.get_context("spawn") + child_attempting = ctx.Event() + child_acquired = ctx.Event() + release_parent = ctx.Event() + process = ctx.Process( + target=_fork_while_another_thread_holds_frontend_project_lock, + args=(str(tmp_path), child_attempting, child_acquired, release_parent), + ) + + child_started = False + child_acquired_while_parent_held = False + try: + process.start() + child_started = child_attempting.wait(20) + if child_started: + child_acquired_while_parent_held = child_acquired.wait(1) + finally: + release_parent.set() + process.join(20) + if process.is_alive(): + process.terminate() + process.join(5) + + assert child_started + assert not child_acquired_while_parent_held + assert child_acquired.is_set() + assert process.exitcode == 0 + + def test_frontend_project_file_lock_retries_windows_contention( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -253,9 +410,16 @@ def test_frontend_project_file_lock_retries_windows_contention( monkeypatch.setitem(sys.modules, "msvcrt", fake_msvcrt) lock_path = tmp_path / "lock" - with lock_path.open("w+b") as lock_file: - frontend_lock._acquire_project_file_lock(lock_file) - frontend_lock._release_project_file_lock(lock_file) + lock_fd = os.open( + lock_path, + os.O_RDWR | os.O_CREAT | getattr(os, "O_BINARY", 0), + 0o666, + ) + try: + frontend_lock._acquire_project_file_lock(lock_fd) + frontend_lock._release_project_file_lock(lock_fd) + finally: + os.close(lock_fd) assert [call.args[1:] for call in fake_msvcrt.locking.call_args_list] == [ (fake_msvcrt.LK_NBLCK, 1),