fix: atomic config writes to prevent zero-byte truncation on disk full

open(path, "w") truncates immediately; if the write then fails (no space),
the config is lost. Switch all config save paths to write a temp file in
the same directory, then os.replace() atomically, so the original is only
replaced after the new content is fully flushed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

frigate/api/app.py

@@ -54,6 +54,7 @@ from frigate.types import JobStatusTypesEnum
 from frigate.util.builtin import (
     clean_camera_user_pass,
     deep_merge,
+    atomic_write_config,
     flatten_config_data,
     load_labels,
     process_config_query_string,
@@ -453,10 +454,7 @@ def config_save(save_option: str, body: Any = Body(media_type="text/plain")):
     # Save the config to file
     try:
         config_file = find_config_file()
-
-        with open(config_file, "w") as f:
-            f.write(new_config)
-            f.close()
+        atomic_write_config(config_file, new_config)
     except Exception:
         return JSONResponse(
             content=(
@@ -678,9 +676,7 @@ def config_set(request: Request, body: AppConfigSetBody):
                 try:
                     config = FrigateConfig.parse(new_raw_config)
                 except ValidationError as e:
-                    with open(config_file, "w") as f:
-                        f.write(old_raw_config)
-                        f.close()
+                    atomic_write_config(config_file, old_raw_config)
                     logger.error(
                         f"Config Validation Error:\n\n{str(traceback.format_exc())}"
                     )
@@ -706,9 +702,7 @@ def config_set(request: Request, body: AppConfigSetBody):
                         status_code=400,
                     )
                 except Exception:
-                    with open(config_file, "w") as f:
-                        f.write(old_raw_config)
-                        f.close()
+                    atomic_write_config(config_file, old_raw_config)
                     logger.error(f"\nConfig Error:\n\n{str(traceback.format_exc())}")
                     return JSONResponse(
                         content=(

frigate/test/test_config.py

@@ -1,5 +1,6 @@
 import json
 import os
+import tempfile
 import unittest
 from unittest.mock import patch
 
@@ -10,7 +11,7 @@ from ruamel.yaml.constructor import DuplicateKeyError
 from frigate.config import BirdseyeModeEnum, FrigateConfig
 from frigate.const import MODEL_CACHE_DIR
 from frigate.detectors import DetectorTypeEnum
-from frigate.util.builtin import deep_merge, load_labels
+from frigate.util.builtin import atomic_write_config, deep_merge, load_labels
 
 
 class TestConfig(unittest.TestCase):
@@ -1677,5 +1678,51 @@ class TestConfig(unittest.TestCase):
         self.assertRaises(ValueError, lambda: FrigateConfig(**config))
 
 
+class TestAtomicWriteConfig(unittest.TestCase):
+    def test_writes_content(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "config.yml")
+            atomic_write_config(path, "foo: bar\n")
+            with open(path) as f:
+                self.assertEqual(f.read(), "foo: bar\n")
+
+    def test_preserves_original_when_replace_fails(self):
+        """os.replace failure (e.g. cross-device) must leave the original intact."""
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "config.yml")
+            original = "original: content\n"
+            with open(path, "w") as f:
+                f.write(original)
+
+            with patch("frigate.util.builtin.os.replace", side_effect=OSError("fail")):
+                with self.assertRaises(OSError):
+                    atomic_write_config(path, "new: content\n")
+
+            with open(path) as f:
+                self.assertEqual(f.read(), original)
+
+    def test_preserves_original_when_write_fails(self):
+        """Disk-full during temp-file write must leave the original intact."""
+        from unittest.mock import MagicMock
+
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "config.yml")
+            original = "original: content\n"
+            with open(path, "w") as f:
+                f.write(original)
+
+            mock_fh = MagicMock()
+            mock_fh.__enter__ = lambda s: mock_fh
+            mock_fh.__exit__ = MagicMock(return_value=False)
+            mock_fh.write.side_effect = OSError("No space left on device")
+
+            with patch("frigate.util.builtin.os.fdopen", return_value=mock_fh):
+                with self.assertRaises(OSError):
+                    atomic_write_config(path, "new: content\n")
+
+            with open(path) as f:
+                self.assertEqual(f.read(), original)
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)

frigate/util/builtin.py

@@ -6,11 +6,14 @@ import datetime
 import logging
 import math
 import multiprocessing.queues
+import os
 import queue
 import re
 import shlex
 import struct
+import tempfile
 import urllib.parse
+from io import StringIO
 from collections.abc import Mapping
 from multiprocessing.managers import ValueProxy
 from pathlib import Path
@@ -242,6 +245,26 @@ def split_config_key_path(key_path_str: str) -> list[str]:
     return parts
 
 
+def atomic_write_config(file_path: str, content: str) -> None:
+    """Write content to file_path atomically via a temp file + os.replace.
+
+    Prevents truncation to zero bytes on out-of-disk-space errors: the
+    original file is only replaced after the new content is fully written.
+    """
+    dir_path = os.path.dirname(os.path.abspath(file_path))
+    fd, tmp_path = tempfile.mkstemp(dir=dir_path)
+    try:
+        with os.fdopen(fd, "w") as f:
+            f.write(content)
+        os.replace(tmp_path, file_path)
+    except Exception:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
+
+
 def update_yaml_file_bulk(file_path: str, updates: Dict[str, Any]):
     yaml = YAML()
     yaml.indent(mapping=2, sequence=4, offset=2)
@@ -268,8 +291,9 @@ def update_yaml_file_bulk(file_path: str, updates: Dict[str, Any]):
         data = update_yaml(data, key_path, new_value)
 
     try:
-        with open(file_path, "w") as f:
-            yaml.dump(data, f)
+        buf = StringIO()
+        yaml.dump(data, buf)
+        atomic_write_config(file_path, buf.getvalue())
     except Exception as e:
         logger.error(f"Unable to write to Frigate config file {file_path}: {e}")
 

frigate/util/config.py

@@ -4,12 +4,13 @@ import asyncio
 import logging
 import os
 import shutil
+from io import StringIO
 from typing import Any, Optional, Union
 
 from ruamel.yaml import YAML
 
 from frigate.const import CONFIG_DIR, EXPORT_DIR
-from frigate.util.builtin import deep_merge
+from frigate.util.builtin import atomic_write_config, deep_merge
 from frigate.util.services import get_video_properties
 
 logger = logging.getLogger(__name__)
@@ -53,11 +54,14 @@ def migrate_frigate_config(config_file: str):
     logger.info("copying config as backup...")
     shutil.copy(config_file, os.path.join(CONFIG_DIR, "backup_config.yaml"))
 
+    def _dump_and_write(cfg):
+        buf = StringIO()
+        yaml.dump(cfg, buf)
+        atomic_write_config(config_file, buf.getvalue())
+
     if previous_version < "0.14":
         logger.info(f"Migrating frigate config from {previous_version} to 0.14...")
-        new_config = migrate_014(config)
-        with open(config_file, "w") as f:
-            yaml.dump(new_config, f)
+        _dump_and_write(migrate_014(config))
         previous_version = "0.14"
 
         logger.info("Migrating export file names...")
@@ -73,37 +77,27 @@ def migrate_frigate_config(config_file: str):
 
     if previous_version < "0.15-0":
         logger.info(f"Migrating frigate config from {previous_version} to 0.15-0...")
-        new_config = migrate_015_0(config)
-        with open(config_file, "w") as f:
-            yaml.dump(new_config, f)
+        _dump_and_write(migrate_015_0(config))
         previous_version = "0.15-0"
 
     if previous_version < "0.15-1":
         logger.info(f"Migrating frigate config from {previous_version} to 0.15-1...")
-        new_config = migrate_015_1(config)
-        with open(config_file, "w") as f:
-            yaml.dump(new_config, f)
+        _dump_and_write(migrate_015_1(config))
         previous_version = "0.15-1"
 
     if previous_version < "0.16-0":
         logger.info(f"Migrating frigate config from {previous_version} to 0.16-0...")
-        new_config = migrate_016_0(config)
-        with open(config_file, "w") as f:
-            yaml.dump(new_config, f)
+        _dump_and_write(migrate_016_0(config))
         previous_version = "0.16-0"
 
     if previous_version < "0.17-0":
         logger.info(f"Migrating frigate config from {previous_version} to 0.17-0...")
-        new_config = migrate_017_0(config)
-        with open(config_file, "w") as f:
-            yaml.dump(new_config, f)
+        _dump_and_write(migrate_017_0(config))
         previous_version = "0.17-0"
 
     if previous_version < "0.18-0":
         logger.info(f"Migrating frigate config from {previous_version} to 0.18-0...")
-        new_config = migrate_018_0(config)
-        with open(config_file, "w") as f:
-            yaml.dump(new_config, f)
+        _dump_and_write(migrate_018_0(config))
         previous_version = "0.18-0"
 
     logger.info("Finished frigate config migration...")