Serialize concurrent frontend dependency installs

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.

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.

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.

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

packages/reflex-base/src/reflex_base/utils/decorator.py

@@ -77,18 +77,53 @@ 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 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
 
     return None, None
 

reflex/utils/frontend_lock.py

@@ -0,0 +1,181 @@
+"""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 reflex_base import constants
+
+_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_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_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.
+    # 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):
+            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(
+        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_fd: int) -> None:
+    """Acquire an advisory lock on an open project lock file descriptor.
+
+    Args:
+        lock_fd: Raw descriptor kept open for the lifetime of the lock.
+    """
+    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:
+                os.lseek(lock_fd, 0, os.SEEK_SET)
+                msvcrt.locking(  # pyright: ignore[reportAttributeAccessIssue]
+                    lock_fd,
+                    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_fd, fcntl.LOCK_EX)
+
+
+def _release_project_file_lock(lock_fd: int) -> None:
+    """Release an advisory lock on an open project lock file descriptor.
+
+    Args:
+        lock_fd: Raw descriptor previously passed to the acquire helper.
+    """
+    os.lseek(lock_fd, 0, os.SEEK_SET)
+    if constants.IS_WINDOWS:
+        import msvcrt
+
+        msvcrt.locking(  # pyright: ignore[reportAttributeAccessIssue]
+            lock_fd,
+            msvcrt.LK_UNLCK,  # pyright: ignore[reportAttributeAccessIssue]
+            1,
+        )
+        return
+
+    import fcntl
+
+    fcntl.flock(lock_fd, 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()
+
+    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()
+        if lock_path in held_paths:
+            yield
+            return
+
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        with _project_lock_fds_guard:
+            lock_fd = os.open(
+                lock_path,
+                os.O_RDWR | os.O_CREAT | getattr(os, "O_BINARY", 0),
+                0o666,
+            )
+            try:
+                _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:
+                    try:
+                        if path_held:
+                            held_paths.remove(lock_path)
+                    finally:
+                        if lock_acquired:
+                            _release_project_file_lock(lock_fd)
+                finally:
+                    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.
+        if os.getpid() == owner_pid:
+            thread_lock.release()

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