From c2c129db6102b88b65e9f8c296dad91e3ada0057 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Sun, 5 Jul 2026 03:42:07 -0700 Subject: [PATCH] fix: fall back to direct write when atomic state save hits EPERM on NFS --- app/state_store.py | 35 ++++++++++++++++++++++-- app/tests/test_state_store.py | 50 +++++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 2 deletions(-) diff --git a/app/state_store.py b/app/state_store.py index cab711a..2548ad6 100644 --- a/app/state_store.py +++ b/app/state_store.py @@ -62,6 +62,7 @@ class AtomicJsonStore: self.path = path self.kind = kind self.schema_version = schema_version + self._direct_write_fallback_warned = False def _ensure_parent(self) -> None: parent = os.path.dirname(self.path) @@ -96,6 +97,13 @@ class AtomicJsonStore: def save(self, data: dict[str, Any]) -> None: self._ensure_parent() payload = self._build_payload(data) + try: + self._atomic_write(payload) + except OSError as exc: + self._warn_direct_write_fallback(exc) + self._direct_write(payload) + + def _atomic_write(self, payload: dict[str, Any]) -> None: parent = os.path.dirname(self.path) or "." fd, tmp_path = tempfile.mkstemp( prefix=f".{os.path.basename(self.path)}.", @@ -105,8 +113,7 @@ class AtomicJsonStore: ) try: with os.fdopen(fd, "w", encoding="utf-8") as f: - json.dump(payload, f, ensure_ascii=False, separators=(",", ":")) - f.write("\n") + self._write_payload(payload, f) f.flush() os.fsync(f.fileno()) os.replace(tmp_path, self.path) @@ -118,6 +125,30 @@ class AtomicJsonStore: pass raise + def _direct_write(self, payload: dict[str, Any]) -> None: + with open(self.path, "w", encoding="utf-8") as f: + self._write_payload(payload, f) + f.flush() + try: + os.fsync(f.fileno()) + except OSError: + pass + + @staticmethod + def _write_payload(payload: dict[str, Any], f: Any) -> None: + json.dump(payload, f, ensure_ascii=False, separators=(",", ":")) + f.write("\n") + + def _warn_direct_write_fallback(self, exc: OSError) -> None: + if self._direct_write_fallback_warned: + return + self._direct_write_fallback_warned = True + log.warning( + "Atomic state write failed for %s (%s); falling back to direct write", + self.path, + exc, + ) + def quarantine_invalid_file(self, exc: Exception) -> None: if not os.path.exists(self.path): return diff --git a/app/tests/test_state_store.py b/app/tests/test_state_store.py index fb71b08..759d89a 100644 --- a/app/tests/test_state_store.py +++ b/app/tests/test_state_store.py @@ -4,6 +4,7 @@ import os import tempfile import unittest from datetime import datetime +from unittest.mock import patch from state_store import AtomicJsonStore, from_json_compatible, to_json_compatible @@ -21,6 +22,55 @@ class StateStoreTests(unittest.TestCase): self.assertEqual(payload["schema_version"], 2) self.assertEqual(payload["items"][0]["info"]["title"], "hello") + def test_save_falls_back_to_direct_write_when_mkstemp_fails(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "queue.json") + store = AtomicJsonStore(path, kind="persistent_queue:queue") + + with self.assertLogs("state_store", level="WARNING") as logs: + with patch( + "state_store.tempfile.mkstemp", + side_effect=PermissionError(1, "Operation not permitted"), + ): + store.save({"items": [{"key": "a"}]}) + + self.assertTrue(os.path.exists(path)) + self.assertTrue(any(path in message for message in logs.output)) + payload = store.load() + self.assertEqual(payload["items"], [{"key": "a"}]) + + def test_save_falls_back_to_direct_write_when_replace_fails(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "queue.json") + store = AtomicJsonStore(path, kind="persistent_queue:queue") + + with patch( + "state_store.os.replace", + side_effect=PermissionError(1, "Operation not permitted"), + ): + store.save({"items": [{"key": "a"}]}) + + self.assertTrue(os.path.exists(path)) + payload = store.load() + self.assertEqual(payload["items"], [{"key": "a"}]) + self.assertEqual([], [name for name in os.listdir(tmp) if name.endswith(".tmp")]) + + def test_save_reraises_when_atomic_and_direct_write_fail(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "queue.json") + store = AtomicJsonStore(path, kind="persistent_queue:queue") + + with patch( + "state_store.tempfile.mkstemp", + side_effect=PermissionError(1, "Operation not permitted"), + ): + with patch("builtins.open", side_effect=PermissionError(13, "Permission denied")): + with self.assertRaises(PermissionError) as ctx: + store.save({"items": [{"key": "a"}]}) + + self.assertEqual(ctx.exception.errno, 13) + self.assertFalse(os.path.exists(path)) + def test_invalid_file_is_quarantined(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "queue.json")