From 4748531910a3cbead75f16769c9259afadb7f3e1 Mon Sep 17 00:00:00 2001 From: Maximus Putnam Date: Wed, 28 Oct 2026 01:24:07 -0500 Subject: [PATCH 1/3] fix: serialize atomic session saves and retry Windows sharing conflicts --- .github/workflows/ci.yml | 24 ++ docs/docs/system/conversation-manager.md | 18 ++ penguin/system/atomic_storage.py | 136 +++++++++ penguin/system/session_manager.py | 89 ++++-- tests/system/test_atomic_session_storage.py | 314 ++++++++++++++++++++ 5 files changed, 556 insertions(+), 25 deletions(-) create mode 100644 penguin/system/atomic_storage.py create mode 100644 tests/system/test_atomic_session_storage.py 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..eba00a5b 100644 --- a/docs/docs/system/conversation-manager.md +++ b/docs/docs/system/conversation-manager.md @@ -141,6 +141,24 @@ 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; ordinary permission failures are returned immediately. 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..308a3974 --- /dev/null +++ b/penguin/system/atomic_storage.py @@ -0,0 +1,136 @@ +"""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_is_shared(path: Path) -> bool: + """Probe DELETE access to distinguish sharing from ACL/read-only failures. + + MoveFileEx may report access denied (5) for an open file. CreateFile reports + sharing violation (32) for a conflicting handle, while a real access denial + remains 5. The probe never changes the file and closes successful handles. + """ + 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.CloseHandle.argtypes = [wintypes.HANDLE] + kernel.CloseHandle.restype = wintypes.BOOL + handle = kernel.CreateFileW(str(path), 0x10000, 7, None, 3, 0, None) + if handle == wintypes.HANDLE(-1).value: + return ctypes.get_last_error() in (32, 33) + kernel.CloseHandle(handle) + return False + + +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): + return True + return code == 5 and ( + _windows_delete_is_shared(source) or _windows_delete_is_shared(target) + ) + + +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..b6775708 --- /dev/null +++ b/tests/system/test_atomic_session_storage.py @@ -0,0 +1,314 @@ +"""Concurrent session writes and narrowly classified Windows sharing failures.""" + +from __future__ import annotations + +import json +import os +import stat +import threading +import time +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 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)] + paths = [] + real_replace = atomic_storage.os.replace + + def observe(source, target): + paths.append(str(source)) + return real_replace(source, target) + + monkeypatch.setattr(atomic_storage.os, "replace", observe) + stop = threading.Event() + invalid = [] + + def read_continuously(): + while not stop.wait(0.001): + for path in tmp_path.glob("*.json"): + try: + json.loads(path.read_text(encoding="utf-8")) + 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(paths) == len(set(paths)) + 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,shared,attempts", + [(32, False, 2), (33, False, 2), (5, True, 2), (5, False, 1)], +) +def test_retry_requires_a_confirmed_sharing_failure( + tmp_path, monkeypatch, code, shared, 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_is_shared", lambda path: shared + ) + 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) From 3d2eb2016e47defcea4e1ddc7be89d58595844d4 Mon Sep 17 00:00:00 2001 From: Maximus Putnam Date: Wed, 28 Oct 2026 01:27:15 -0500 Subject: [PATCH 2/3] fix: handle Windows sharing conflicts that clear before probing --- docs/docs/system/conversation-manager.md | 4 +- penguin/system/atomic_storage.py | 32 +++++++++----- tests/system/test_atomic_session_storage.py | 47 ++++++++++++++++++--- 3 files changed, 65 insertions(+), 18 deletions(-) diff --git a/docs/docs/system/conversation-manager.md b/docs/docs/system/conversation-manager.md index eba00a5b..67157c79 100644 --- a/docs/docs/system/conversation-manager.md +++ b/docs/docs/system/conversation-manager.md @@ -148,7 +148,9 @@ 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; ordinary permission failures are returned immediately. There are at +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. diff --git a/penguin/system/atomic_storage.py b/penguin/system/atomic_storage.py index 308a3974..139289a4 100644 --- a/penguin/system/atomic_storage.py +++ b/penguin/system/atomic_storage.py @@ -47,12 +47,12 @@ def guarded(self: Any, *args: Any, **kwargs: Any) -> _T: return guarded -def _windows_delete_is_shared(path: Path) -> bool: +def _windows_delete_access(path: Path) -> int: """Probe DELETE access to distinguish sharing from ACL/read-only failures. - MoveFileEx may report access denied (5) for an open file. CreateFile reports - sharing violation (32) for a conflicting handle, while a real access denial - remains 5. The probe never changes the file and closes successful handles. + 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 @@ -68,13 +68,20 @@ def _windows_delete_is_shared(path: Path) -> bool: 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() in (32, 33) + return ctypes.get_last_error() kernel.CloseHandle(handle) - return False + return 0 def _sharing_violation(error: OSError, source: Path, target: Path) -> bool: @@ -82,11 +89,16 @@ def _sharing_violation(error: OSError, source: Path, target: Path) -> bool: if os.name != "nt": return False code = getattr(error, "winerror", None) - if code in (32, 33): + if code in (32, 33, 303): # sharing, lock, or delete-pending conflict return True - return code == 5 and ( - _windows_delete_is_shared(source) or _windows_delete_is_shared(target) - ) + 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: diff --git a/tests/system/test_atomic_session_storage.py b/tests/system/test_atomic_session_storage.py index b6775708..7a0189ff 100644 --- a/tests/system/test_atomic_session_storage.py +++ b/tests/system/test_atomic_session_storage.py @@ -18,6 +18,41 @@ 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 ): @@ -37,7 +72,7 @@ def read_continuously(): while not stop.wait(0.001): for path in tmp_path.glob("*.json"): try: - json.loads(path.read_text(encoding="utf-8")) + json.loads(_read_atomic_snapshot(path)) except Exception as error: invalid.append(error) @@ -191,11 +226,11 @@ def denied(source, destination): @pytest.mark.parametrize( - "code,shared,attempts", - [(32, False, 2), (33, False, 2), (5, True, 2), (5, False, 1)], + "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, shared, attempts + tmp_path, monkeypatch, code, access, attempts ): real_os = atomic_storage.os calls, sleeps = [], [] @@ -213,9 +248,7 @@ def replace(source, target): "os", SimpleNamespace(name="nt", replace=replace, fsync=real_os.fsync), ) - monkeypatch.setattr( - atomic_storage, "_windows_delete_is_shared", lambda path: shared - ) + 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") From c347531295e20a6ea84614ce07ae7a602559298a Mon Sep 17 00:00:00 2001 From: Maximus Putnam Date: Wed, 28 Oct 2026 01:32:29 -0500 Subject: [PATCH 3/3] test: distinguish temp file creation from replacement retries --- tests/system/test_atomic_session_storage.py | 29 ++++++++++++++++++--- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/tests/system/test_atomic_session_storage.py b/tests/system/test_atomic_session_storage.py index 7a0189ff..04d8f94e 100644 --- a/tests/system/test_atomic_session_storage.py +++ b/tests/system/test_atomic_session_storage.py @@ -7,6 +7,7 @@ import stat import threading import time +from collections import Counter from concurrent.futures import ThreadPoolExecutor from pathlib import Path from types import SimpleNamespace @@ -57,14 +58,32 @@ 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)] - paths = [] + 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): - paths.append(str(source)) + 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 = [] @@ -92,7 +111,11 @@ def save(index): stop.set() reader.join(2) assert not invalid - assert len(paths) == len(set(paths)) + 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