fix: fall back to direct write when atomic state save hits EPERM on NFS

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

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")