diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7f7eb473..c2442632 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -9,6 +9,30 @@ permissions: contents: read jobs: + session-storage-contract: + runs-on: ${{ matrix.os }} + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest] + python-version: ["3.11", "3.12"] + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: astral-sh/setup-uv@v6 + with: + version: "0.11.2" + - run: uv sync --locked --python '${{ matrix.python-version }}' + - name: Verify atomic saves, failure recovery, and storage contracts + run: >- + uv run --locked pytest -q + tests/system/test_atomic_session_storage.py + tests/system/test_session_save_sync.py + tests/core_runtime/test_session_goal_store.py + tests/core_runtime/test_opencode_persistence.py + locked-runtime: runs-on: ubuntu-latest timeout-minutes: 20 diff --git a/docs/docs/system/conversation-manager.md b/docs/docs/system/conversation-manager.md index 2cb5d6a4..67157c79 100644 --- a/docs/docs/system/conversation-manager.md +++ b/docs/docs/system/conversation-manager.md @@ -141,6 +141,26 @@ sequenceDiagram ## Session Management +Session, backup, and index writes use unique sibling temporary files, flush them +to disk, and atomically replace the destination. Managers for the same directory +share a reentrant lock within one process. Index updates merge changed entries +with the saved index so another manager's session entries are retained. + +On Windows, replacement retries are limited to sharing/lock violations. An +access-denied error is retried only when a delete-access probe confirms a sharing +conflict or that access has recovered; read-only and ACL denials fail immediately. +External Windows readers should allow delete sharing to avoid blocking a rename; +public manager reads use the same store lock. There are at +most six attempts with 310 ms total backoff. Failed replacements retain the previous +destination, clean up only their own temporary file, and keep cached sessions +marked dirty. Callers must check `save_session()`'s boolean result. + +The session file and index are separate atomic writes, not a cross-file +transaction. If the index write fails after the session succeeds, the save reports +failure and the existing stale-index recovery can rebuild it. Cross-process +revision merging and arbitrary concurrent mutation of a `Session` object are not +provided by the in-process lock. + ```mermaid flowchart TD UserMsg[User Message] --> CheckID{Has Conversation ID?} diff --git a/penguin/system/atomic_storage.py b/penguin/system/atomic_storage.py new file mode 100644 index 00000000..139289a4 --- /dev/null +++ b/penguin/system/atomic_storage.py @@ -0,0 +1,148 @@ +"""Atomic session-file replacement and process-local store synchronization.""" + +from __future__ import annotations + +import logging +import os +import tempfile +import threading +import time +from functools import wraps +from pathlib import Path +from typing import TYPE_CHECKING, Any, TypeVar +from weakref import WeakValueDictionary + +if TYPE_CHECKING: + from _thread import RLock + from collections.abc import Callable + +__all__ = ["atomic_write_text", "store_lock", "synchronized_store"] + +logger = logging.getLogger(__name__) +_LOCKS: WeakValueDictionary[str, RLock] = WeakValueDictionary() +_LOCKS_GUARD = threading.Lock() +_RETRY_DELAYS = (0.01, 0.02, 0.04, 0.08, 0.16) +_T = TypeVar("_T") + + +def store_lock(directory: Path) -> RLock: + """Share a reentrant lock between live managers of the same directory.""" + key = os.path.normcase(str(directory.resolve())) + with _LOCKS_GUARD: + lock = _LOCKS.get(key) + if lock is None: + lock = threading.RLock() + _LOCKS[key] = lock + return lock + + +def synchronized_store(method: Callable[..., _T]) -> Callable[..., _T]: + """Serialize manager state and related file operations in one process.""" + + @wraps(method) + def guarded(self: Any, *args: Any, **kwargs: Any) -> _T: + with self._store_lock: + return method(self, *args, **kwargs) + + return guarded + + +def _windows_delete_access(path: Path) -> int: + """Probe DELETE access to distinguish sharing from ACL/read-only failures. + + Return zero when access is now available, or the Windows error code. A + conflicting reader may close before this probe runs. Read-only attributes + and ACL denials still return 5, so they cannot be mistaken for that race. + """ + import ctypes + from ctypes import wintypes + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.CreateFileW.argtypes = [ + wintypes.LPCWSTR, + wintypes.DWORD, + wintypes.DWORD, + ctypes.c_void_p, + wintypes.DWORD, + wintypes.DWORD, + wintypes.HANDLE, + ] + kernel.CreateFileW.restype = wintypes.HANDLE + kernel.GetFileAttributesW.argtypes = [wintypes.LPCWSTR] + kernel.GetFileAttributesW.restype = wintypes.DWORD + kernel.CloseHandle.argtypes = [wintypes.HANDLE] + kernel.CloseHandle.restype = wintypes.BOOL + attributes = kernel.GetFileAttributesW(str(path)) + if attributes == 0xFFFFFFFF: + return ctypes.get_last_error() + if attributes & 1: # FILE_ATTRIBUTE_READONLY + return 5 + handle = kernel.CreateFileW(str(path), 0x10000, 7, None, 3, 0, None) + if handle == wintypes.HANDLE(-1).value: + return ctypes.get_last_error() + kernel.CloseHandle(handle) + return 0 + + +def _sharing_violation(error: OSError, source: Path, target: Path) -> bool: + """Retry only Windows sharing/lock errors, never arbitrary access denial.""" + if os.name != "nt": + return False + code = getattr(error, "winerror", None) + if code in (32, 33, 303): # sharing, lock, or delete-pending conflict + return True + if code != 5: + return False + source_access = _windows_delete_access(source) + target_access = _windows_delete_access(target) + # The helper owns a newly-created sibling source, so directory creation + # access was already verified. Require DELETE access to be available or + # explicitly sharing-blocked; never retry a remaining access denial. + return source_access in (0, 32, 33, 303) and target_access in (0, 2, 3, 32, 33, 303) + + +def _replace(source: Path, target: Path) -> None: + for attempt in range(len(_RETRY_DELAYS) + 1): + try: + os.replace(source, target) + return + except OSError as error: + if attempt == len(_RETRY_DELAYS) or not _sharing_violation( + error, source, target + ): + raise + time.sleep(_RETRY_DELAYS[attempt]) + + +def atomic_write_text(target: Path, text: str) -> None: + """Flush a unique sibling file, then replace the target or leave it intact. + + Failed attempts clean up only their own temporary file. Replacement retries + have a fixed 310ms sleep budget and are limited to confirmed Windows sharing + violations. The old target is never unlinked or made writable as a fallback. + """ + temporary = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=target.parent, + prefix=f".{target.name}.", + suffix=".tmp", + delete=False, + ) as output: + temporary = Path(output.name) + output.write(text) + output.flush() + os.fsync(output.fileno()) + _replace(temporary, target) + finally: + if temporary is not None: + try: + temporary.unlink(missing_ok=True) + except OSError: + logger.warning( + "Could not clean temporary session file %s", + temporary, + exc_info=True, + ) diff --git a/penguin/system/session_manager.py b/penguin/system/session_manager.py index 1eced003..687379c3 100644 --- a/penguin/system/session_manager.py +++ b/penguin/system/session_manager.py @@ -11,7 +11,6 @@ import json import logging import os -import shutil import threading import time from collections import OrderedDict @@ -40,6 +39,11 @@ from penguin.config import CONVERSATIONS_PATH from penguin.system.state import Message, MessageCategory, Session, create_message +from penguin.system.atomic_storage import ( + atomic_write_text, + store_lock, + synchronized_store, +) from penguin.constants import DEFAULT_MAX_MESSAGES_PER_SESSION @@ -77,6 +81,8 @@ def __init__( auto_save_interval: Seconds between auto-saves of modified sessions """ self.base_path = Path(base_path) + self._store_lock = store_lock(self.base_path) + self._saved_index: Optional[Dict[str, Dict[str, Any]]] = None self.max_messages_per_session = max_messages_per_session self.max_sessions_in_memory = max_sessions_in_memory self.format = format @@ -91,6 +97,7 @@ def __init__( # Initialize the session index self.index_path = self.base_path / "session_index.json" self.session_index = self._load_or_create_index() + self._saved_index = deepcopy(self.session_index) # Setup auto-save thread if interval > 0 self.auto_save_interval = auto_save_interval @@ -102,6 +109,7 @@ def __init__( ) self._auto_save_thread.start() + @synchronized_store def _load_or_create_index(self) -> Dict[str, Dict[str, Any]]: """Load the session index or create it if it doesn't exist.""" if not self.index_path.exists(): @@ -142,6 +150,7 @@ def _session_index_is_stale(self, index: Dict[str, Dict[str, Any]]) -> bool: except OSError: return True + @synchronized_store def _rebuild_index(self) -> Dict[str, Dict[str, Any]]: """Rebuild the session index by scanning session files.""" logger.info("Rebuilding session index from files") @@ -177,18 +186,35 @@ def _rebuild_index(self) -> Dict[str, Dict[str, Any]]: self._save_index(index) return index - def _save_index(self, index: Dict[str, Dict[str, Any]]) -> None: - """Save the session index to disk.""" + @synchronized_store + def _save_index(self, index: Dict[str, Dict[str, Any]]) -> bool: + """Atomically persist local index changes without erasing peer saves.""" try: - # Write to temp file first - fix the suffix - temp_path = Path(f"{self.index_path}.temp") # Fix: Use explicit Path constructor - with _safe_open(temp_path, 'w', encoding='utf-8') as f: - json.dump(index, f, indent=2) - - # Atomic rename - os.replace(temp_path, self.index_path) - except Exception as e: - logger.error(f"Error saving session index: {str(e)}") + merged = deepcopy(index) + if self._saved_index is not None: + try: + with _safe_open(self.index_path, "r", encoding="utf-8") as source: + on_disk = json.load(source) + except (FileNotFoundError, json.JSONDecodeError, UnicodeError): + on_disk = None + if isinstance(on_disk, dict): + merged = on_disk + for key in self._saved_index.keys() - index.keys(): + merged.pop(key, None) + for key, value in index.items(): + if ( + key not in self._saved_index + or value != self._saved_index[key] + ): + merged[key] = deepcopy(value) + atomic_write_text(self.index_path, json.dumps(merged, indent=2)) + index.clear() + index.update(merged) + self._saved_index = deepcopy(merged) + return True + except (OSError, ValueError, TypeError) as error: + logger.error("Error saving session index: %s", error) + return False def _copy_metadata_to_index_entry( self, @@ -244,6 +270,7 @@ def _auto_save_loop(self) -> None: # Sleep for the configured interval self._stop_auto_save.wait(self.auto_save_interval) + @synchronized_store def _auto_save_sessions(self) -> None: """Save all modified sessions.""" modified_sessions = [] @@ -285,6 +312,7 @@ def _auto_save_sessions(self) -> None: if saved_count > 0: logger.debug(f"Auto-saved {saved_count} session(s)") + @synchronized_store def create_session(self) -> Session: """ Create a new empty session. @@ -314,6 +342,7 @@ def create_session(self) -> Session: logger.debug(f"Created new session: {session.id}") return session + @synchronized_store def load_session(self, session_id: str) -> Optional[Session]: """ Load a session by ID with error recovery. @@ -367,7 +396,9 @@ def load_session(self, session_id: str) -> Optional[Session]: self.current_session = session # Restore from backup - shutil.copy2(backup_path, primary_path) + atomic_write_text( + primary_path, backup_path.read_text(encoding="utf-8") + ) logger.warning(f"Successfully restored session {session_id} from backup") return session except Exception as backup_error: @@ -381,6 +412,7 @@ def load_session(self, session_id: str) -> Optional[Session]: ) return self._create_recovery_session(session_id) + @synchronized_store def _add_to_session_cache(self, session_id: str, session: Session) -> None: """Add a session to the cache, managing the LRU behavior.""" # Add to sessions cache @@ -423,6 +455,7 @@ def _load_from_file(self, file_path: Path) -> Optional[Session]: return session + @synchronized_store def _create_recovery_session(self, failed_session_id: str) -> Session: """ Create a recovery session when loading fails. @@ -478,6 +511,7 @@ def _create_recovery_session(self, failed_session_id: str) -> Session: return session + @synchronized_store def save_session(self, session: Optional[Session] = None) -> bool: """ Save a session with transaction safety. @@ -498,7 +532,6 @@ def save_session(self, session: Optional[Session] = None) -> bool: return False try: - temp_path = self.base_path / f"{session.id}.{self.format}.temp" backup_path = self.base_path / f"{session.id}.{self.format}.bak" target_path = self.base_path / f"{session.id}.{self.format}" @@ -510,17 +543,12 @@ def save_session(self, session: Optional[Session] = None) -> bool: token_count = session.total_tokens session.metadata["token_count"] = token_count - # Write to temp file first - with _safe_open(temp_path, 'w', encoding='utf-8') as f: - f.write(session.to_json()) - - # Create backup of current file if it exists + # Serialize before touching disk; preserve a complete previous version. + serialized = session.to_json() if target_path.exists(): - shutil.copy2(target_path, backup_path) - - # Atomic rename of temp to target - os.replace(temp_path, target_path) - + atomic_write_text(backup_path, target_path.read_text(encoding="utf-8")) + atomic_write_text(target_path, serialized) + # Update the session index with consistent token information self.session_index[session.id] = { "created_at": session.created_at, @@ -546,7 +574,10 @@ def save_session(self, session: Optional[Session] = None) -> bool: if "continued_to" in session.metadata: self.session_index[session.id]["continued_to"] = session.metadata["continued_to"] - self._save_index(self.session_index) + if self._save_index(self.session_index) is False: + if session.id in self.sessions: + self.sessions[session.id] = (session, True) + return False # Update cache status if session.id in self.sessions: @@ -557,8 +588,11 @@ def save_session(self, session: Optional[Session] = None) -> bool: except Exception as e: logger.error(f"Error saving session {session.id}: {str(e)}") + if session.id in self.sessions: + self.sessions[session.id] = (session, True) return False + @synchronized_store def check_session_boundary(self, session: Optional[Session] = None) -> bool: """ Check if a session has reached its boundary and should transition. @@ -577,6 +611,7 @@ def check_session_boundary(self, session: Optional[Session] = None) -> bool: # Token budget concerns are handled by ContextWindowManager return len(session.messages) >= self.max_messages_per_session + @synchronized_store def create_continuation_session(self, source_session: Optional[Session] = None) -> Session: """ Create a new session that continues from an existing one. @@ -682,6 +717,7 @@ def create_continuation_session(self, source_session: Optional[Session] = None) return continuation + @synchronized_store def _get_continuation_index(self, session_id: str) -> int: """ Get the continuation index for a session using the index. @@ -700,6 +736,7 @@ def _get_continuation_index(self, session_id: str) -> int: return continuation_count + 1 + @synchronized_store def list_sessions(self, limit: int = 100, offset: int = 0) -> List[Dict]: """ List available sessions with metadata using the index. @@ -796,6 +833,7 @@ def list_sessions(self, limit: int = 100, offset: int = 0) -> List[Dict]: return result + @synchronized_store def delete_session(self, session_id: str) -> bool: """ Delete a session and its files. @@ -832,6 +870,7 @@ def delete_session(self, session_id: str) -> bool: logger.error(f"Error deleting session {session_id}: {str(e)}") return False + @synchronized_store def mark_session_modified(self, session_id: str) -> None: """Mark a session as modified in the cache. diff --git a/tests/system/test_atomic_session_storage.py b/tests/system/test_atomic_session_storage.py new file mode 100644 index 00000000..04d8f94e --- /dev/null +++ b/tests/system/test_atomic_session_storage.py @@ -0,0 +1,370 @@ +"""Concurrent session writes and narrowly classified Windows sharing failures.""" + +from __future__ import annotations + +import json +import os +import stat +import threading +import time +from collections import Counter +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from penguin.system import atomic_storage +from penguin.system.session_manager import SessionManager +from penguin.system.state import Session + + +def _read_atomic_snapshot(path): + """Keep Windows readers compatible with replacement while still reading bytes.""" + if os.name != "nt": + return path.read_text(encoding="utf-8") + import ctypes + import msvcrt + from ctypes import wintypes + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.CreateFileW.argtypes = [ + wintypes.LPCWSTR, + wintypes.DWORD, + wintypes.DWORD, + ctypes.c_void_p, + wintypes.DWORD, + wintypes.DWORD, + wintypes.HANDLE, + ] + kernel.CreateFileW.restype = wintypes.HANDLE + kernel.CloseHandle.argtypes = [wintypes.HANDLE] + kernel.CloseHandle.restype = wintypes.BOOL + # FILE_SHARE_DELETE lets the writer replace a file with an open reader. + # The separate held-reader regression deliberately withholds this right. + handle = kernel.CreateFileW(str(path), 0x80000000, 7, None, 3, 0, None) + if handle == wintypes.HANDLE(-1).value: + raise ctypes.WinError(ctypes.get_last_error()) + try: + descriptor = msvcrt.open_osfhandle(handle, os.O_RDONLY) + except OSError: + kernel.CloseHandle(handle) + raise + with os.fdopen(descriptor, "r", encoding="utf-8") as source: + return source.read() + + +def test_concurrent_same_session_writes_are_complete_and_use_unique_temps( + tmp_path, monkeypatch +): + managers = [SessionManager(str(tmp_path), auto_save_interval=0) for _ in range(4)] + created_paths, attempts = [], [] + real_temporary_file = atomic_storage.tempfile.NamedTemporaryFile + real_replace = atomic_storage.os.replace + real_sharing_violation = atomic_storage._sharing_violation + injected_error = PermissionError("injected transient sharing conflict") + + def create_temporary(*args, **kwargs): + temporary = real_temporary_file(*args, **kwargs) + created_paths.append(str(temporary.name)) + return temporary + + def observe(source, target): + attempts.append(str(source)) + if len(attempts) == 1: + raise injected_error + return real_replace(source, target) + + monkeypatch.setattr(atomic_storage.tempfile, "NamedTemporaryFile", create_temporary) + monkeypatch.setattr(atomic_storage.os, "replace", observe) + monkeypatch.setattr( + atomic_storage, + "_sharing_violation", + lambda error, source, target: ( + error is injected_error or real_sharing_violation(error, source, target) + ), + ) + stop = threading.Event() + invalid = [] + + def read_continuously(): + while not stop.wait(0.001): + for path in tmp_path.glob("*.json"): + try: + json.loads(_read_atomic_snapshot(path)) + except Exception as error: + invalid.append(error) + + reader = threading.Thread(target=read_continuously) + reader.start() + + def save(index): + session = Session( + id="shared", metadata={"revision": index, "payload": str(index) * 2000} + ) + return managers[index % 4].save_session(session) + + try: + with ThreadPoolExecutor(max_workers=8) as pool: + assert all(pool.map(save, range(32))) + finally: + stop.set() + reader.join(2) + assert not invalid + assert len(created_paths) == len(set(created_paths)) + assert set(attempts) == set(created_paths) + attempts_per_write = Counter(attempts) + assert attempts_per_write[created_paths[0]] >= 2 + assert all(1 <= count <= 6 for count in attempts_per_write.values()) + for name in ("shared.json", "shared.json.bak"): + data = json.loads((tmp_path / name).read_text(encoding="utf-8")) + assert data["metadata"]["payload"] == str(data["metadata"]["revision"]) * 2000 + index = json.loads((tmp_path / "session_index.json").read_text()) + current = json.loads((tmp_path / "shared.json").read_text()) + assert index["shared"]["last_active"] == current["last_active"] + assert not list(tmp_path.glob("*.tmp")) + + +def test_separate_managers_keep_peer_index_entries_and_metadata(tmp_path): + first = SessionManager(str(tmp_path), auto_save_interval=0) + second = SessionManager(str(tmp_path), auto_save_interval=0) + one, two = Session(id="one"), Session(id="two") + assert first.save_session(one) + assert second.save_session(two) + one.metadata["title"] = "first changed" + assert first.save_session(one) + two.metadata["title"] = "second changed" + assert second.save_session(two) + index = json.loads(first.index_path.read_text(encoding="utf-8")) + assert set(index) == {"one", "two"} + assert index["one"]["title"] == "first changed" + assert index["two"]["title"] == "second changed" + + +def test_concurrent_distinct_sessions_keep_all_index_entries(tmp_path): + managers = [SessionManager(str(tmp_path), auto_save_interval=0) for _ in range(4)] + + def save(index): + return managers[index % 4].save_session(Session(id=f"session-{index}")) + + with ThreadPoolExecutor(max_workers=8) as pool: + assert all(pool.map(save, range(24))) + index = json.loads((tmp_path / "session_index.json").read_text()) + assert set(index) == {f"session-{i}" for i in range(24)} + + +def test_serialization_failure_never_changes_existing_files(tmp_path, monkeypatch): + manager = SessionManager(str(tmp_path), auto_save_interval=0) + session = manager.create_session() + assert manager.save_session(session) + original = {path: path.read_bytes() for path in tmp_path.iterdir()} + + def broken_json(): + raise ValueError("cannot serialize") + + monkeypatch.setattr(session, "to_json", broken_json) + assert not manager.save_session(session) + assert {path: path.read_bytes() for path in tmp_path.iterdir()} == original + + +def test_flush_failure_preserves_target_and_removes_own_temp(tmp_path, monkeypatch): + target = tmp_path / "target.json" + target.write_text("old") + + def failed_flush(_fd): + raise OSError("disk flush failed") + + monkeypatch.setattr(atomic_storage.os, "fsync", failed_flush) + with pytest.raises(OSError, match="flush"): + atomic_storage.atomic_write_text(target, "new") + assert target.read_text() == "old" + assert not list(tmp_path.glob("*.tmp")) + + +def test_backup_recovery_restores_complete_session_atomically(tmp_path): + manager = SessionManager(str(tmp_path), auto_save_interval=0) + session = manager.create_session() + session.metadata["revision"] = "old" + assert manager.save_session(session) + session.metadata["revision"] = "new" + assert manager.save_session(session) + target = tmp_path / f"{session.id}.json" + backup = target.with_suffix(".json.bak") + expected = backup.read_bytes() + target.write_text("{incomplete", encoding="utf-8") + reloaded = SessionManager(str(tmp_path), auto_save_interval=0) + recovered = reloaded.load_session(session.id) + assert recovered.metadata["revision"] == "old" + assert target.read_bytes() == expected + assert backup.read_bytes() == expected + assert not list(tmp_path.glob("*.tmp")) + + +def test_failed_session_replace_preserves_previous_data_and_marks_dirty( + tmp_path, monkeypatch +): + manager = SessionManager(str(tmp_path), auto_save_interval=0) + session = manager.create_session() + assert manager.save_session(session) + target = tmp_path / f"{session.id}.json" + original = target.read_bytes() + session.metadata["title"] = "unsaved" + real_replace = atomic_storage.os.replace + + def denied(source, destination): + if Path(destination) == target: + raise PermissionError("real permission denial") + return real_replace(source, destination) + + monkeypatch.setattr(atomic_storage.os, "replace", denied) + assert not manager.save_session(session) + assert target.read_bytes() == original + assert target.with_suffix(".json.bak").read_bytes() == original + assert manager.sessions[session.id][1] is True + assert not list(tmp_path.glob("*.tmp")) + + +def test_failed_index_save_is_not_reported_as_complete(tmp_path, monkeypatch): + manager = SessionManager(str(tmp_path), auto_save_interval=0) + session = manager.create_session() + assert manager.save_session(session) + old_index = manager.index_path.read_bytes() + session.metadata["title"] = "new" + real_replace = atomic_storage.os.replace + + def denied(source, destination): + if Path(destination) == manager.index_path: + raise PermissionError("index denied") + return real_replace(source, destination) + + monkeypatch.setattr(atomic_storage.os, "replace", denied) + assert not manager.save_session(session) + assert manager.index_path.read_bytes() == old_index + assert manager.sessions[session.id][1] is True + assert ( + json.loads((tmp_path / f"{session.id}.json").read_text())["metadata"]["title"] + == "new" + ) + assert not list(tmp_path.glob("*.tmp")) + + +@pytest.mark.parametrize( + "code,access,attempts", + [(32, 0, 2), (33, 0, 2), (303, 0, 2), (5, 32, 2), (5, 0, 2), (5, 5, 1)], +) +def test_retry_requires_a_confirmed_sharing_failure( + tmp_path, monkeypatch, code, access, attempts +): + real_os = atomic_storage.os + calls, sleeps = [], [] + error = PermissionError("injected Windows error") + error.winerror = code + + def replace(source, target): + calls.append((source, target)) + if len(calls) == 1: + raise error + return real_os.replace(source, target) + + monkeypatch.setattr( + atomic_storage, + "os", + SimpleNamespace(name="nt", replace=replace, fsync=real_os.fsync), + ) + monkeypatch.setattr(atomic_storage, "_windows_delete_access", lambda path: access) + monkeypatch.setattr(atomic_storage.time, "sleep", sleeps.append) + target = tmp_path / "data.json" + target.write_text("old") + if attempts == 1: + with pytest.raises(PermissionError): + atomic_storage.atomic_write_text(target, "new") + assert target.read_text() == "old" + else: + atomic_storage.atomic_write_text(target, "new") + assert target.read_text() == "new" + assert len(calls) == attempts + assert len(sleeps) == attempts - 1 + assert not list(tmp_path.glob("*.tmp")) + + +def test_sharing_retry_exhaustion_is_bounded_and_preserves_target( + tmp_path, monkeypatch +): + error = PermissionError("persistent sharing violation") + error.winerror = 32 + calls, sleeps = [], [] + + def blocked(source, target): + calls.append(source) + raise error + + monkeypatch.setattr( + atomic_storage, + "os", + SimpleNamespace(name="nt", replace=blocked, fsync=os.fsync), + ) + monkeypatch.setattr(atomic_storage.time, "sleep", sleeps.append) + target = tmp_path / "data.json" + target.write_text("old") + with pytest.raises(PermissionError): + atomic_storage.atomic_write_text(target, "new") + assert len(calls) == 6 + assert sum(sleeps) == pytest.approx(0.31) + assert target.read_text() == "old" + assert not list(tmp_path.glob("*.tmp")) + + +@pytest.mark.skipif(os.name != "nt", reason="Windows sharing semantics") +def test_windows_open_reader_releases_during_replace_retry(tmp_path, monkeypatch): + target = tmp_path / "held.json" + target.write_text("old") + held = target.open("r") + attempted = threading.Event() + real_replace = atomic_storage.os.replace + errors = [] + + def replace(source, destination): + try: + return real_replace(source, destination) + except OSError as error: + errors.append(error.winerror) + attempted.set() + raise + + def release(): + assert attempted.wait(2) + time.sleep(0.05) + held.close() + + monkeypatch.setattr(atomic_storage.os, "replace", replace) + thread = threading.Thread(target=release) + thread.start() + try: + atomic_storage.atomic_write_text(target, "new") + finally: + held.close() + thread.join(2) + assert errors and errors[0] == 5 + assert target.read_text() == "new" + + +@pytest.mark.skipif(os.name != "nt", reason="Windows read-only attribute") +def test_windows_read_only_target_is_not_retried_or_changed(tmp_path, monkeypatch): + target = tmp_path / "readonly.json" + target.write_text("old") + target.chmod(stat.S_IREAD) + calls = [] + real_replace = atomic_storage.os.replace + + def replace(source, destination): + calls.append(source) + return real_replace(source, destination) + + monkeypatch.setattr(atomic_storage.os, "replace", replace) + try: + with pytest.raises(PermissionError): + atomic_storage.atomic_write_text(target, "new") + assert len(calls) == 1 + assert target.read_text() == "old" + finally: + target.chmod(stat.S_IWRITE)