diff --git a/app/state_store.py b/app/state_store.py index cab711a..940a882 100644 --- a/app/state_store.py +++ b/app/state_store.py @@ -2,6 +2,7 @@ from __future__ import annotations import base64 import collections.abc +import errno import json import logging import os @@ -17,6 +18,25 @@ STATE_SCHEMA_VERSION = 2 _BYTES_MARKER = "__metube_bytes__" _DATETIME_MARKER = "__metube_datetime__" +# Errnos that signal the filesystem cannot support the temp-file + rename +# atomic-write strategy (for example an NFS-backed state dir returning EPERM on +# mkstemp). These are safe to fall back on because they mean the atomic +# mechanism is unavailable, not that the data write itself failed. Errors like +# ENOSPC/EIO are deliberately excluded so a genuine storage failure surfaces +# instead of silently truncating an existing good state file. +_ATOMIC_UNSUPPORTED_ERRNOS = frozenset( + e + for e in ( + errno.EPERM, + errno.EACCES, + errno.ENOSYS, + errno.EINVAL, + getattr(errno, "EOPNOTSUPP", None), + getattr(errno, "ENOTSUP", None), + ) + if e is not None +) + def to_json_compatible(value: Any) -> Any: if value is None or isinstance(value, (bool, int, float, str)): @@ -62,6 +82,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 +117,16 @@ 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: + if exc.errno not in _ATOMIC_UNSUPPORTED_ERRNOS: + raise + self._warn_direct_write_fallback(exc) + self._direct_write(payload) + + def _atomic_write(self, payload: dict[str, Any]) -> None: + text = self._serialize(payload) parent = os.path.dirname(self.path) or "." fd, tmp_path = tempfile.mkstemp( prefix=f".{os.path.basename(self.path)}.", @@ -105,10 +136,9 @@ class AtomicJsonStore: ) try: with os.fdopen(fd, "w", encoding="utf-8") as f: - json.dump(payload, f, ensure_ascii=False, separators=(",", ":")) - f.write("\n") + f.write(text) f.flush() - os.fsync(f.fileno()) + self._best_effort_fsync(f.fileno()) os.replace(tmp_path, self.path) self._fsync_directory(parent) except Exception: @@ -118,6 +148,57 @@ class AtomicJsonStore: pass raise + def _direct_write(self, payload: dict[str, Any]) -> None: + # Serialize before truncating so a serialization failure never destroys + # the existing state file (the atomic path gets this for free via its + # temp file). + text = self._serialize(payload) + # Create with 0o600 so the fallback keeps the owner-only permissions the + # atomic path gets from mkstemp; state files can contain URLs and + # per-download option overrides that must not leak on shared mounts. + fd = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as f: + # The 0o600 mode above only applies when the file is created; force + # it on rewrites too so an existing, broadly-permissioned state file + # is tightened to match the atomic path. Best-effort because some + # network filesystems reject chmod, and that must not re-crash save. + try: + os.fchmod(f.fileno(), 0o600) + except OSError: + pass + f.write(text) + f.flush() + self._best_effort_fsync(f.fileno()) + # Make the new directory entry durable too, matching the atomic path. + self._fsync_directory(os.path.dirname(self.path) or ".") + + @staticmethod + def _best_effort_fsync(fileno: int) -> None: + # Tolerate fsync being unsupported on the underlying filesystem (for + # example a network mount that returns EINVAL/ENOSYS), but let genuine + # storage failures such as ENOSPC/EIO surface so a non-durable write is + # never reported as success. An unsupported fsync must not by itself + # abandon the atomic rename path. + try: + os.fsync(fileno) + except OSError as exc: + if exc.errno not in _ATOMIC_UNSUPPORTED_ERRNOS: + raise + + @staticmethod + def _serialize(payload: dict[str, Any]) -> str: + return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\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..15e0880 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,135 @@ 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)) + # Fallback keeps owner-only permissions, matching the atomic path. + self.assertEqual(os.stat(path).st_mode & 0o777, 0o600) + payload = store.load() + self.assertEqual(payload["items"], [{"key": "a"}]) + + def test_fallback_tightens_permissions_on_existing_file(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "queue.json") + with open(path, "w", encoding="utf-8") as f: + f.write("{}") + os.chmod(path, 0o644) + + store = AtomicJsonStore(path, kind="persistent_queue:queue") + with patch( + "state_store.tempfile.mkstemp", + side_effect=PermissionError(1, "Operation not permitted"), + ): + store.save({"items": [{"key": "a"}]}) + + self.assertEqual(os.stat(path).st_mode & 0o777, 0o600) + self.assertEqual(store.load()["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( + "state_store.os.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_unsupported_fsync_keeps_atomic_path(self): + # fsync being unsupported (EINVAL/ENOSYS) must not by itself trigger the + # direct-write fallback; the atomic temp-file + rename path still runs. + import errno as _errno + + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "queue.json") + store = AtomicJsonStore(path, kind="persistent_queue:queue") + + with patch( + "state_store.os.fsync", + side_effect=OSError(_errno.EINVAL, "Invalid argument"), + ): + with self.assertNoLogs("state_store", level="WARNING"): + store.save({"items": [{"key": "a"}]}) + + self.assertEqual(store.load()["items"], [{"key": "a"}]) + self.assertEqual([], [name for name in os.listdir(tmp) if name.endswith(".tmp")]) + + def test_save_reraises_and_preserves_state_on_non_atomic_errno(self): + # A storage failure such as ENOSPC is not an "atomic unavailable" + # signal, so it must surface instead of falling back to a direct write + # that would truncate the existing good state file. + import errno as _errno + + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "queue.json") + store = AtomicJsonStore(path, kind="persistent_queue:queue") + store.save({"items": [{"key": "good"}]}) + + with patch( + "state_store.tempfile.mkstemp", + side_effect=OSError(_errno.ENOSPC, "No space left on device"), + ): + with self.assertRaises(OSError) as ctx: + store.save({"items": [{"key": "new"}]}) + + self.assertEqual(ctx.exception.errno, _errno.ENOSPC) + # Existing state is untouched. + self.assertEqual(store.load()["items"], [{"key": "good"}]) + + def test_serialization_failure_preserves_existing_state(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "queue.json") + store = AtomicJsonStore(path, kind="persistent_queue:queue") + store.save({"items": [{"key": "good"}]}) + + # Even on the fallback path, a non-serializable payload must raise + # before the existing good state file is touched. + with patch( + "state_store.tempfile.mkstemp", + side_effect=PermissionError(1, "Operation not permitted"), + ): + with self.assertRaises(TypeError): + store.save({"items": object()}) + + self.assertEqual(store.load()["items"], [{"key": "good"}]) + def test_invalid_file_is_quarantined(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "queue.json")