diff --git a/news/+atomic-json-updates.bugfix.md b/news/+atomic-json-updates.bugfix.md new file mode 100644 index 00000000000..9e0eb6c37ec --- /dev/null +++ b/news/+atomic-json-updates.bugfix.md @@ -0,0 +1 @@ +Prevent concurrent Reflex processes from corrupting or losing updates to shared JSON metadata files. diff --git a/pyproject.toml b/pyproject.toml index 52954a08271..7ba181e89fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,6 +21,7 @@ keywords = ["web", "framework"] requires-python = ">=3.10,<4.0" dependencies = [ "click >=8.2", + "filelock >=3.32.3,<4.0", "granian[reload] >=2.7.4", "httpx >=0.26,<1.0", "packaging >=24.2,<27", diff --git a/reflex/constants/__init__.py b/reflex/constants/__init__.py index fe2ab035dbb..2e248dff038 100644 --- a/reflex/constants/__init__.py +++ b/reflex/constants/__init__.py @@ -6,6 +6,7 @@ IS_LINUX, IS_MACOS, IS_WINDOWS, + JSON_LOCKS_DIR, LOCAL_STORAGE, POLLING_MAX_HTTP_BUFFER_SIZE, PYTEST_CURRENT_TEST, @@ -68,6 +69,7 @@ "IS_LINUX", "IS_MACOS", "IS_WINDOWS", + "JSON_LOCKS_DIR", "LOCAL_STORAGE", "NOCOMPILE_FILE", "POLLING_MAX_HTTP_BUFFER_SIZE", diff --git a/reflex/constants/base.py b/reflex/constants/base.py index b500c7e5f0c..c48c3a7c1ac 100644 --- a/reflex/constants/base.py +++ b/reflex/constants/base.py @@ -2,3 +2,6 @@ """Re-export from reflex_base.""" from reflex_base.constants.base import * # pragma: no cover + +# The per-user subdirectory containing stable JSON metadata lock files. +JSON_LOCKS_DIR = "locks/json" diff --git a/reflex/utils/frontend_skeleton.py b/reflex/utils/frontend_skeleton.py index 9a5fa3d9ee3..9778aa49e2b 100644 --- a/reflex/utils/frontend_skeleton.py +++ b/reflex/utils/frontend_skeleton.py @@ -452,11 +452,18 @@ def initialize_web_directory(): """Initialize the web directory on reflex init.""" 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 - project_hash = get_project_hash() + web_dir = get_web_dir() + # Keep JSON writers out of their same-directory staging window while the + # frontend tree is removed and recreated. + with ( + path_ops._json_file_lock((web_dir / constants.Reflex.JSON).resolve()), + path_ops._json_file_lock((web_dir / constants.Dirs.ENV_JSON).resolve()), + ): + # Reuse the hash if one is already created, so we don't over-write it when running reflex init + project_hash = get_project_hash() - logger.debug(f"Copying {constants.Templates.Dirs.WEB_TEMPLATE} to {get_web_dir()}") - path_ops.copy_tree(constants.Templates.Dirs.WEB_TEMPLATE, str(get_web_dir())) + logger.debug(f"Copying {constants.Templates.Dirs.WEB_TEMPLATE} to {web_dir}") + path_ops.copy_tree(constants.Templates.Dirs.WEB_TEMPLATE, str(web_dir)) logger.debug("Restoring lockfiles.") sync_root_lockfiles_to_web() diff --git a/reflex/utils/path_ops.py b/reflex/utils/path_ops.py index 6e4da438e7d..d4ddf3dc36e 100644 --- a/reflex/utils/path_ops.py +++ b/reflex/utils/path_ops.py @@ -8,10 +8,14 @@ import shutil import stat from pathlib import Path +from typing import TYPE_CHECKING from reflex_base.config import get_config from reflex_base.environment import environment +if TYPE_CHECKING: + from filelock import BaseFileLock + # Shorthand for join. join = os.linesep.join @@ -224,36 +228,119 @@ def get_bun_path() -> Path | None: return bun_path.absolute() if bun_path else None -def update_json_file(file_path: str | Path, update_dict: dict[str, object]): +def _json_file_lock_path(file_path: Path) -> Path: + """Get the stable lock path for a JSON file. + + Args: + file_path: The normalized path of the JSON file. + + Returns: + A path in Reflex's per-user data directory keyed by the target path. + """ + import hashlib + + from reflex import constants + + normalized_path = os.path.normcase(os.fspath(file_path.resolve())) + if constants.IS_MACOS: + import unicodedata + + # posixpath.normcase is a no-op on macOS, even when the underlying APFS + # volume is case-insensitive and Unicode-normalizing. + normalized_path = unicodedata.normalize("NFC", normalized_path).casefold() + target_digest = hashlib.sha256(os.fsencode(normalized_path)).hexdigest() + lock_directory = ( + environment.REFLEX_DIR.get().expanduser().resolve() / constants.JSON_LOCKS_DIR + ) + lock_directory.mkdir(mode=0o700, parents=True, exist_ok=True) + return lock_directory / f"{target_digest}.lock" + + +def _json_file_lock(file_path: Path) -> BaseFileLock: + """Get the process-safe lock for a JSON file. + + Args: + file_path: The normalized path of the JSON file. + + Returns: + A reentrant lock shared by callers targeting the same file. + """ + # Keep this import off CLI startup paths that do not write JSON metadata. + from filelock import FileLock + + return FileLock( + _json_file_lock_path(file_path), + mode=0o600, + is_singleton=True, + fallback_to_soft=False, + preserve_lock_file=True, + ) + + +def _write_json_file(file_path: Path, value: dict[str, object]) -> None: + """Atomically replace a JSON file with a complete document. + + Args: + file_path: The destination JSON file. + value: The complete JSON object to write. + """ + import contextlib + import secrets + + open_flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0) + for _ in range(100): + temp_path = file_path.with_name(f".{file_path.name}.{secrets.token_hex(8)}.tmp") + try: + # Unlike tempfile.mkstemp's fixed 0600, mode 0666 preserves the old + # Path.touch behavior by letting the process umask set new-file mode. + temp_fd = os.open(temp_path, open_flags, 0o666) + except FileExistsError: + continue + break + else: + msg = f"Unable to allocate a temporary file for {file_path}" + raise FileExistsError(msg) + + try: + if file_path.exists(): + shutil.copymode(file_path, temp_path) + temp_file = os.fdopen(temp_fd, "w", encoding="utf-8") + temp_fd = -1 + with temp_file: + json.dump(value, temp_file, ensure_ascii=False) + temp_file.flush() + os.fsync(temp_file.fileno()) + temp_path.replace(file_path) + except BaseException: + if temp_fd != -1: + with contextlib.suppress(OSError): + os.close(temp_fd) + with contextlib.suppress(OSError): + temp_path.unlink(missing_ok=True) + raise + + +def update_json_file(file_path: str | Path, update_dict: dict[str, object]) -> None: """Update the contents of a json file. Args: file_path: the path to the JSON file. update_dict: object to update json. """ - fp = Path(file_path) + fp = Path(file_path).resolve() # Create the parent directory if it doesn't exist. fp.parent.mkdir(parents=True, exist_ok=True) - # Create the file if it doesn't exist. - fp.touch(exist_ok=True) - - # Create an empty json object if file is empty - fp.write_text("{}") if fp.stat().st_size == 0 else None - - # Read the existing json object from the file. - json_object = {} - if fp.stat().st_size: - with fp.open() as f: - json_object = json.load(f) - - # Update the json object with the new data. - json_object.update(update_dict) + with _json_file_lock(fp): + # An absent or empty file represents an empty JSON object. + json_object: dict[str, object] = {} + if fp.exists() and fp.stat().st_size: + with fp.open(encoding="utf-8") as json_file: + json_object = json.load(json_file) - # Write the updated json object to the file - with fp.open("w") as f: - json.dump(json_object, f, ensure_ascii=False) + json_object.update(update_dict) + _write_json_file(fp, json_object) def find_replace(directory: str | Path, find: str, replace: str): diff --git a/reflex/utils/prerequisites.py b/reflex/utils/prerequisites.py index 57574d7d6bf..023cd2b5ab9 100644 --- a/reflex/utils/prerequisites.py +++ b/reflex/utils/prerequisites.py @@ -122,8 +122,10 @@ def get_or_set_last_reflex_version_check_datetime(): data = json.loads(reflex_json_file.read_text()) last_version_check_datetime = data.get("last_version_check_datetime") if not last_version_check_datetime: - data.update({"last_version_check_datetime": str(datetime.now())}) - path_ops.update_json_file(reflex_json_file, data) + path_ops.update_json_file( + reflex_json_file, + {"last_version_check_datetime": str(datetime.now())}, + ) return last_version_check_datetime diff --git a/tests/units/test_prerequisites.py b/tests/units/test_prerequisites.py index 5cba1a236e3..961ab8cd035 100644 --- a/tests/units/test_prerequisites.py +++ b/tests/units/test_prerequisites.py @@ -27,6 +27,40 @@ runner = CliRunner() +def test_version_check_timestamp_update_does_not_replay_stale_json( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """Recording a version check preserves a concurrent metadata update.""" + web_dir = tmp_path / constants.Dirs.WEB + web_dir.mkdir() + reflex_json_file = web_dir / constants.Reflex.JSON + reflex_json_file.write_text( + json.dumps({"last_reflex_run_datetime": "old"}), + encoding="utf-8", + ) + monkeypatch.setattr(prerequisites, "get_web_dir", lambda: web_dir) + update_json_file = prerequisites.path_ops.update_json_file + + def update_after_concurrent_write( + file_path: Path, + update: dict[str, object], + ) -> None: + update_json_file(file_path, {"last_reflex_run_datetime": "new"}) + update_json_file(file_path, update) + + monkeypatch.setattr( + prerequisites.path_ops, + "update_json_file", + update_after_concurrent_write, + ) + + assert prerequisites.get_or_set_last_reflex_version_check_datetime() is None + data = json.loads(reflex_json_file.read_text(encoding="utf-8")) + assert data["last_reflex_run_datetime"] == "new" + assert data["last_version_check_datetime"] + + 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) diff --git a/tests/units/utils/test_path_ops.py b/tests/units/utils/test_path_ops.py new file mode 100644 index 00000000000..e06f1ab9734 --- /dev/null +++ b/tests/units/utils/test_path_ops.py @@ -0,0 +1,718 @@ +"""Tests for filesystem path operations.""" + +from __future__ import annotations + +import json +import multiprocessing +import os +import shutil +import subprocess +import sys +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest +from filelock import Timeout + +from reflex import constants +from reflex.utils import frontend_skeleton, path_ops + + +@pytest.fixture(autouse=True) +def _isolate_reflex_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Keep persistent JSON locks inside each test's temporary directory.""" + monkeypatch.setenv("REFLEX_DIR", str(tmp_path / "reflex-data")) + + +def _pause_json_update_before_write( + file_path: str, + update: dict[str, object], + inside_critical_section, + release, +) -> None: + """Pause an updater after its read while it still owns the JSON lock.""" + original_write_json_file = path_ops._write_json_file + + def pause_before_write(target: Path, value: dict[str, object]) -> None: + inside_critical_section.set() + if not release.wait(timeout=10): + msg = "timed out waiting to release JSON update" + raise TimeoutError(msg) + original_write_json_file(target, value) + + path_ops._write_json_file = pause_before_write + path_ops.update_json_file(file_path, update) + + +def _update_json_in_process( + file_path: str, + update: dict[str, object], + lock_blocked, + read_started, + completed, +) -> None: + """Apply an update and report its progress from a child process.""" + original_json_file_lock = path_ops._json_file_lock + original_json_load = path_ops.json.load + + class ProbedLock: + """Prove the public updater encounters an already-held lock.""" + + def __init__(self, lock) -> None: + self.lock = lock + + def __enter__(self): + try: + self.lock.acquire(timeout=0) + except Timeout: + lock_blocked.set() + else: + self.lock.release() + msg = "JSON updater unexpectedly acquired the held lock" + raise AssertionError(msg) + return self.lock.acquire() + + def __exit__(self, exception_type, exception, traceback): + self.lock.release() + + def probed_json_file_lock(target: Path): + return ProbedLock(original_json_file_lock(target)) + + def report_json_read(file, *args, **kwargs): + read_started.set() + return original_json_load(file, *args, **kwargs) + + path_ops._json_file_lock = probed_json_file_lock + path_ops.json.load = report_json_read + path_ops.update_json_file(file_path, update) + completed.set() + + +def _crash_during_json_update(file_path: str, partial_dump_written) -> None: + """Exit a child process while it owns the lock and staged file.""" + target = Path(file_path) + + def crashing_dump(_value, file, **_kwargs): + file.write("{") + file.flush() + os.fsync(file.fileno()) + partial_dump_written.set() + os._exit(23) + + path_ops.json.dump = crashing_dump + path_ops.update_json_file(target, {"crashed": True}) + + +def _join_process(process) -> None: + """Join a child process without leaking it after a failed assertion.""" + process.join(timeout=10) + if process.is_alive(): + process.terminate() + process.join(timeout=10) + + +def test_path_ops_keeps_filelock_import_lazy(): + """Importing path helpers alone does not pay the filelock import cost.""" + result = subprocess.run( + [ + sys.executable, + "-c", + "import sys; import reflex.utils.path_ops; print('filelock' in sys.modules)", + ], + check=True, + capture_output=True, + text=True, + ) + + assert result.stdout.strip() == "False" + + +def test_update_json_file_serializes_concurrent_updates( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """Concurrent updates merge under one critical section.""" + target = tmp_path / "reflex.json" + target.write_text(json.dumps({"base": True}), encoding="utf-8") + first_dump_started = threading.Event() + release_first_dump = threading.Event() + dump_calls = 0 + dump_calls_lock = threading.Lock() + original_dump = path_ops.json.dump + + def controlled_dump(*args, **kwargs): + nonlocal dump_calls + with dump_calls_lock: + dump_call = dump_calls + dump_calls += 1 + if dump_call == 0: + first_dump_started.set() + assert release_first_dump.wait(timeout=5) + return original_dump(*args, **kwargs) + + monkeypatch.setattr(path_ops.json, "dump", controlled_dump) + + with ThreadPoolExecutor(max_workers=2) as executor: + first_update = executor.submit( + path_ops.update_json_file, + target, + {"first": True}, + ) + assert first_dump_started.wait(timeout=5) + with ( + pytest.raises(Timeout), + path_ops._json_file_lock(target.resolve()).acquire(timeout=0), + ): + pass + second_update = executor.submit( + path_ops.update_json_file, + target, + {"second": True}, + ) + release_first_dump.set() + first_update.result(timeout=5) + second_update.result(timeout=5) + + assert json.loads(target.read_text(encoding="utf-8")) == { + "base": True, + "first": True, + "second": True, + } + + +def test_update_json_file_never_exposes_partial_json( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """Readers see the old document until the new document is complete.""" + target = tmp_path / "reflex.json" + original = {"base": True} + target.write_text(json.dumps(original), encoding="utf-8") + partial_dump_written = threading.Event() + finish_dump = threading.Event() + + def split_dump(value, file, *, ensure_ascii): + serialized = json.dumps(value, ensure_ascii=ensure_ascii) + file.write(serialized[:1]) + file.flush() + partial_dump_written.set() + assert finish_dump.wait(timeout=5) + file.write(serialized[1:]) + + monkeypatch.setattr(path_ops.json, "dump", split_dump) + + with ThreadPoolExecutor(max_workers=1) as executor: + update = executor.submit( + path_ops.update_json_file, + target, + {"payload": ["value"] * 100}, + ) + assert partial_dump_written.wait(timeout=5) + try: + observed_while_writing = json.loads(target.read_text(encoding="utf-8")) + finally: + finish_dump.set() + update.result(timeout=5) + + assert observed_while_writing == original + assert json.loads(target.read_text(encoding="utf-8")) == { + **original, + "payload": ["value"] * 100, + } + + +def test_web_directory_cleanup_waits_for_staged_json_update( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """Frontend cleanup cannot delete an active JSON staging file.""" + web_dir = tmp_path / ".web" + web_dir.mkdir() + target = web_dir / "reflex.json" + target.write_text(json.dumps({"base": True}), encoding="utf-8") + stage_written = threading.Event() + release_writer = threading.Event() + cleanup_lock_blocked = threading.Event() + cleanup_reached = threading.Event() + allow_cleanup = threading.Event() + original_dump = path_ops.json.dump + original_json_file_lock = path_ops._json_file_lock + + def pausing_dump(value, file, *, ensure_ascii): + file.write("{") + file.flush() + stage_written.set() + assert release_writer.wait(timeout=5) + file.seek(0) + file.truncate() + original_dump(value, file, ensure_ascii=ensure_ascii) + + class ProbedLock: + """Report that frontend cleanup encounters the writer's lock.""" + + def __init__(self, lock) -> None: + self.lock = lock + + def __enter__(self): + try: + self.lock.acquire(timeout=0) + except Timeout: + cleanup_lock_blocked.set() + else: + self.lock.release() + msg = "frontend cleanup unexpectedly acquired the writer lock" + raise AssertionError(msg) + return self.lock.acquire() + + def __exit__(self, exception_type, exception, traceback): + self.lock.release() + + def probed_json_file_lock(file_path: Path): + lock = original_json_file_lock(file_path) + if file_path == target.resolve(): + return ProbedLock(lock) + return lock + + def destructive_copy(_source, _destination): + cleanup_reached.set() + assert allow_cleanup.wait(timeout=5) + shutil.rmtree(web_dir) + web_dir.mkdir() + + monkeypatch.setattr(path_ops.json, "dump", pausing_dump) + monkeypatch.setattr(frontend_skeleton, "get_web_dir", lambda: web_dir) + monkeypatch.setattr(frontend_skeleton, "get_project_hash", lambda: None) + for function_name in ( + "sync_root_lockfiles_to_web", + "initialize_package_json", + "sync_web_lockfiles_to_root", + "initialize_bun_config", + "initialize_npmrc", + "update_react_router_config", + "initialize_vite_config", + "init_reflex_json", + ): + monkeypatch.setattr( + frontend_skeleton, + function_name, + lambda *args, **kwargs: None, + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + writer = executor.submit(path_ops.update_json_file, target, {"writer": True}) + assert stage_written.wait(timeout=5) + staged_files = list(web_dir.glob(".reflex.json.*.tmp")) + assert len(staged_files) == 1 + monkeypatch.setattr(path_ops, "_json_file_lock", probed_json_file_lock) + monkeypatch.setattr(path_ops, "copy_tree", destructive_copy) + cleanup = executor.submit(frontend_skeleton.initialize_web_directory) + try: + assert cleanup_lock_blocked.wait(timeout=5) + assert not cleanup_reached.is_set() + assert staged_files[0].exists() + release_writer.set() + assert cleanup_reached.wait(timeout=5) + writer.result(timeout=5) + assert json.loads(target.read_text(encoding="utf-8")) == { + "base": True, + "writer": True, + } + finally: + release_writer.set() + allow_cleanup.set() + cleanup.result(timeout=5) + + +def test_update_json_file_dump_failure_preserves_original( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """A failed serialization does not truncate the existing document.""" + target = tmp_path / "reflex.json" + original = json.dumps({"base": True}) + target.write_text(original, encoding="utf-8") + + def failing_dump(_value, file, **_kwargs): + file.write("{") + file.flush() + msg = "injected dump failure" + raise OSError(msg) + + monkeypatch.setattr(path_ops.json, "dump", failing_dump) + + with pytest.raises(OSError, match="injected dump failure"): + path_ops.update_json_file(target, {"new": True}) + + assert target.read_text(encoding="utf-8") == original + assert not list(tmp_path.glob(".reflex.json.*.tmp")) + + +def test_update_json_file_dump_failure_preserves_missing_target( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """A failed first write leaves the target absent and removes its staging file.""" + target = tmp_path / "reflex.json" + + def failing_dump(_value, file, **_kwargs): + file.write("{") + msg = "injected dump failure" + raise OSError(msg) + + monkeypatch.setattr(path_ops.json, "dump", failing_dump) + + with pytest.raises(OSError, match="injected dump failure"): + path_ops.update_json_file(target, {"new": True}) + + assert not target.exists() + assert not list(tmp_path.glob(".reflex.json.*.tmp")) + + +@pytest.mark.parametrize("replace_target_directory", [False, True]) +def test_update_json_file_processes_merge_disjoint_updates( + tmp_path: Path, + replace_target_directory: bool, +): + """The public updater excludes another process despite directory replacement.""" + target_directory = tmp_path / ".web" + target_directory.mkdir() + target = target_directory / "reflex.json" + target.write_text(json.dumps({"base": True}), encoding="utf-8") + process_context = multiprocessing.get_context("spawn") + inside_critical_section = process_context.Event() + release_update = process_context.Event() + second_lock_blocked = process_context.Event() + second_read_started = process_context.Event() + second_completed = process_context.Event() + first = process_context.Process( + target=_pause_json_update_before_write, + args=( + str(target), + {"first": True}, + inside_critical_section, + release_update, + ), + ) + second = process_context.Process( + target=_update_json_in_process, + args=( + str(target), + {"second": True}, + second_lock_blocked, + second_read_started, + second_completed, + ), + ) + + first.start() + try: + assert inside_critical_section.wait(timeout=10) + lock_path = path_ops._json_file_lock_path(target.resolve()) + assert lock_path.is_file() + if replace_target_directory: + shutil.rmtree(target_directory) + target_directory.mkdir() + assert lock_path.is_file() + second.start() + assert second_lock_blocked.wait(timeout=10) + assert not second_read_started.is_set() + assert not second_completed.is_set() + if replace_target_directory: + assert not target.exists() + else: + assert json.loads(target.read_text(encoding="utf-8")) == {"base": True} + finally: + release_update.set() + _join_process(first) + if second.pid is not None: + _join_process(second) + + assert first.exitcode == 0 + assert second.exitcode == 0 + assert second_read_started.is_set() + assert second_completed.is_set() + assert json.loads(target.read_text(encoding="utf-8")) == { + "base": True, + "first": True, + "second": True, + } + + +def test_update_json_file_crash_preserves_target_and_releases_lock(tmp_path: Path): + """A process crash leaves the target valid and releases its OS lock.""" + target = tmp_path / "reflex.json" + original = {"base": True} + target.write_text(json.dumps(original), encoding="utf-8") + process_context = multiprocessing.get_context("spawn") + partial_dump_written = process_context.Event() + process = process_context.Process( + target=_crash_during_json_update, + args=(str(target), partial_dump_written), + ) + + process.start() + assert partial_dump_written.wait(timeout=10) + _join_process(process) + + assert process.exitcode == 23 + assert json.loads(target.read_text(encoding="utf-8")) == original + # An uncatchable process exit can orphan its staging file, but that partial + # document is never installed at the public target path. + staged_files = list(tmp_path.glob(".reflex.json.*.tmp")) + assert len(staged_files) == 1 + assert staged_files[0].read_text(encoding="utf-8") == "{" + with path_ops._json_file_lock(target.resolve()).acquire(timeout=1): + pass + path_ops.update_json_file(target, {"after_crash": True}) + assert json.loads(target.read_text(encoding="utf-8")) == { + **original, + "after_crash": True, + } + + +def test_update_json_file_replace_failure_preserves_original( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """A failed atomic replacement keeps the prior document.""" + target = tmp_path / "reflex.json" + original = json.dumps({"base": True}) + target.write_text(original, encoding="utf-8") + + def failing_replace(_self, _target): + msg = "injected replace failure" + raise OSError(msg) + + monkeypatch.setattr(Path, "replace", failing_replace) + + with pytest.raises(OSError, match="injected replace failure"): + path_ops.update_json_file(target, {"new": True}) + + assert target.read_text(encoding="utf-8") == original + assert not list(tmp_path.glob(".reflex.json.*.tmp")) + + +def test_update_json_file_fsync_failure_preserves_original( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """A failed staged-file sync keeps the prior document.""" + target = tmp_path / "reflex.json" + original = json.dumps({"base": True}) + target.write_text(original, encoding="utf-8") + + def failing_fsync(_file_descriptor): + msg = "injected fsync failure" + raise OSError(msg) + + monkeypatch.setattr(path_ops.os, "fsync", failing_fsync) + + with pytest.raises(OSError, match="injected fsync failure"): + path_ops.update_json_file(target, {"new": True}) + + assert target.read_text(encoding="utf-8") == original + assert not list(tmp_path.glob(".reflex.json.*.tmp")) + + +def test_update_json_file_flush_failure_preserves_original( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """A failed staged-file flush keeps the prior document.""" + target = tmp_path / "reflex.json" + original = json.dumps({"base": True}) + target.write_text(original, encoding="utf-8") + original_fdopen = path_ops.os.fdopen + + class FlushFailingFile: + def __init__(self, file) -> None: + self.file = file + + def __enter__(self): + return self + + def __exit__(self, exception_type, exception, traceback): + self.file.close() + + def write(self, value): + return self.file.write(value) + + def flush(self): + self.file.flush() + msg = "injected flush failure" + raise OSError(msg) + + def fileno(self): + return self.file.fileno() + + def failing_fdopen(*args, **kwargs): + return FlushFailingFile(original_fdopen(*args, **kwargs)) + + monkeypatch.setattr(path_ops.os, "fdopen", failing_fdopen) + + with pytest.raises(OSError, match="injected flush failure"): + path_ops.update_json_file(target, {"new": True}) + + assert target.read_text(encoding="utf-8") == original + assert not list(tmp_path.glob(".reflex.json.*.tmp")) + + +def test_update_json_file_distinct_files_do_not_block_each_other( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + """A writer only blocks other writers for the same normalized path.""" + first_target = tmp_path / "first.json" + second_target = tmp_path / "second.json" + first_target.write_text("{}", encoding="utf-8") + second_target.write_text("{}", encoding="utf-8") + first_dump_started = threading.Event() + second_dump_started = threading.Event() + release_first_dump = threading.Event() + dump_calls = 0 + dump_calls_lock = threading.Lock() + original_dump = path_ops.json.dump + + def controlled_dump(*args, **kwargs): + nonlocal dump_calls + with dump_calls_lock: + dump_call = dump_calls + dump_calls += 1 + if dump_call == 0: + first_dump_started.set() + assert release_first_dump.wait(timeout=5) + else: + second_dump_started.set() + return original_dump(*args, **kwargs) + + monkeypatch.setattr(path_ops.json, "dump", controlled_dump) + + with ThreadPoolExecutor(max_workers=2) as executor: + first_update = executor.submit( + path_ops.update_json_file, + first_target, + {"first": True}, + ) + assert first_dump_started.wait(timeout=5) + second_update = executor.submit( + path_ops.update_json_file, + second_target, + {"second": True}, + ) + try: + assert second_dump_started.wait(timeout=5) + finally: + release_first_dump.set() + first_update.result(timeout=5) + second_update.result(timeout=5) + + +def test_json_file_lock_is_reentrant_for_normalized_path(tmp_path: Path): + """Repeated lock objects for one normalized path can nest in one thread.""" + target = (tmp_path / "nested" / ".." / "reflex.json").resolve() + first_lock = path_ops._json_file_lock(target) + second_lock = path_ops._json_file_lock(target) + + assert first_lock is second_lock + with first_lock.acquire(timeout=1), second_lock.acquire(timeout=0): + pass + + +def test_json_file_lock_sidecar_persists_with_private_permissions(tmp_path: Path): + """The stable lock sidecar remains private after releasing the lock.""" + target = tmp_path / "reflex.json" + + path_ops.update_json_file(target, {"created": True}) + + lock_path = path_ops._json_file_lock_path(target.resolve()) + assert lock_path.is_file() + assert lock_path.is_relative_to(tmp_path / "reflex-data") + if os.name != "nt": + assert lock_path.stat().st_mode & 0o077 == 0 + + +def test_update_json_file_normalizes_symlink_alias(tmp_path: Path): + """Symlink aliases update and lock the canonical target path.""" + target = tmp_path / "reflex.json" + target.write_text("{}", encoding="utf-8") + alias = tmp_path / "alias.json" + try: + alias.symlink_to(target) + except OSError as error: + pytest.skip(f"symlinks unavailable: {error}") + + path_ops.update_json_file(alias, {"updated": True}) + + assert alias.is_symlink() + assert json.loads(target.read_text(encoding="utf-8")) == {"updated": True} + assert path_ops._json_file_lock_path(target).is_file() + assert path_ops._json_file_lock_path(target) == path_ops._json_file_lock_path(alias) + + +@pytest.mark.skipif(not constants.IS_MACOS, reason="macOS path normalization") +def test_json_file_lock_normalizes_macos_case_alias(tmp_path: Path): + """Case aliases on case-insensitive macOS volumes share one lock.""" + target = tmp_path / "MixedCase.json" + target.write_text("{}", encoding="utf-8") + alias = tmp_path / "mixedcase.json" + if not alias.exists() or not alias.samefile(target): + pytest.skip("test volume is case-sensitive") + + assert path_ops._json_file_lock_path(target) == path_ops._json_file_lock_path(alias) + + +def test_update_json_file_creates_missing_parent_and_document(tmp_path: Path): + """A missing JSON file starts as an empty object.""" + target = tmp_path / "missing" / "reflex.json" + + path_ops.update_json_file(target, {"created": "yes"}) + + assert json.loads(target.read_text(encoding="utf-8")) == {"created": "yes"} + + +def test_update_json_file_treats_empty_document_as_object(tmp_path: Path): + """An empty existing file retains its historical empty-object behavior.""" + target = tmp_path / "reflex.json" + target.touch() + + path_ops.update_json_file(target, {"created": "yes"}) + + assert json.loads(target.read_text(encoding="utf-8")) == {"created": "yes"} + + +def test_update_json_file_malformed_document_is_unchanged(tmp_path: Path): + """Malformed JSON still raises without replacing its source document.""" + target = tmp_path / "reflex.json" + malformed = "{not-json" + target.write_text(malformed, encoding="utf-8") + + with pytest.raises(json.JSONDecodeError): + path_ops.update_json_file(target, {"new": True}) + + assert target.read_text(encoding="utf-8") == malformed + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_update_json_file_preserves_existing_permissions(tmp_path: Path): + """Atomic replacement retains the existing file's permission bits.""" + target = tmp_path / "reflex.json" + target.write_text("{}", encoding="utf-8") + target.chmod(0o640) + original_mode = target.stat().st_mode & 0o777 + + path_ops.update_json_file(target, {"new": True}) + + assert target.stat().st_mode & 0o777 == original_mode + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_update_json_file_new_document_honors_umask(tmp_path: Path): + """A new atomic document has the same mode as the old Path.touch flow.""" + expected_mode_file = tmp_path / "expected-mode" + expected_mode_file.touch() + target = tmp_path / "reflex.json" + + path_ops.update_json_file(target, {"new": True}) + + assert target.stat().st_mode & 0o777 == expected_mode_file.stat().st_mode & 0o777 diff --git a/uv.lock b/uv.lock index 054c6a98407..190430179bd 100644 --- a/uv.lock +++ b/uv.lock @@ -3765,6 +3765,7 @@ name = "reflex" source = { editable = "." } dependencies = [ { name = "click" }, + { name = "filelock" }, { name = "granian", extra = ["reload"] }, { name = "httpx" }, { name = "packaging" }, @@ -3861,6 +3862,7 @@ dev = [ requires-dist = [ { name = "alembic", marker = "extra == 'db'", specifier = ">=1.15.2,<2.0" }, { name = "click", specifier = ">=8.2" }, + { name = "filelock", specifier = ">=3.32.3,<4.0" }, { name = "granian", extras = ["reload"], specifier = ">=2.7.4" }, { name = "httpx", specifier = ">=0.26,<1.0" }, { name = "packaging", specifier = ">=24.2,<27" },