diff --git a/tests/hermes_cli/test_atomic_json_write.py b/tests/hermes_cli/test_atomic_json_write.py index 3068cceea11f..1849efdbf6b8 100644 --- a/tests/hermes_cli/test_atomic_json_write.py +++ b/tests/hermes_cli/test_atomic_json_write.py @@ -7,7 +7,7 @@ import pytest -from utils import atomic_json_write +from utils import atomic_json_write, atomic_replace class TestAtomicJsonWrite: @@ -165,6 +165,28 @@ def test_mode_applied_when_supported(self, tmp_path): actual = stat_mod.S_IMODE(target.stat().st_mode) assert actual == 0o600 + def test_atomic_replace_retries_transient_permission_error(self, tmp_path): + import utils + + target = tmp_path / "data.json" + tmp_file = tmp_path / "data.tmp" + tmp_file.write_text('{"ok": true}', encoding="utf-8") + calls = 0 + real_replace = os.replace + + def flaky_replace(src, dst): + nonlocal calls + calls += 1 + if calls < 3: + raise PermissionError("temporarily locked") + real_replace(src, dst) + + with patch.object(utils.os, "replace", side_effect=flaky_replace): + assert atomic_replace(tmp_file, target) == str(target) + + assert calls == 3 + assert json.loads(target.read_text(encoding="utf-8")) == {"ok": True} + def test_concurrent_writes_dont_corrupt(self, tmp_path): """Multiple rapid writes should each produce valid JSON.""" import threading diff --git a/utils.py b/utils.py index b2c9f1e83ff2..f15e5b1d1888 100644 --- a/utils.py +++ b/utils.py @@ -5,6 +5,7 @@ import os import stat import tempfile +import time from pathlib import Path from typing import Any, Union from urllib.parse import urlparse @@ -15,6 +16,8 @@ TRUTHY_STRINGS = frozenset({"1", "true", "yes", "on"}) +_ATOMIC_REPLACE_MAX_ATTEMPTS = 8 +_ATOMIC_REPLACE_RETRY_DELAY = 0.01 def is_truthy_value(value: Any, default: bool = False) -> bool: @@ -78,7 +81,14 @@ def atomic_replace(tmp_path: Union[str, Path], target: Union[str, Path]) -> str: """ target_str = str(target) real_path = os.path.realpath(target_str) if os.path.islink(target_str) else target_str - os.replace(str(tmp_path), real_path) + for attempt in range(_ATOMIC_REPLACE_MAX_ATTEMPTS): + try: + os.replace(str(tmp_path), real_path) + break + except PermissionError: + if attempt == _ATOMIC_REPLACE_MAX_ATTEMPTS - 1: + raise + time.sleep(_ATOMIC_REPLACE_RETRY_DELAY * (attempt + 1)) return real_path