fix: configio thread safety, tmp cleanup, backup collision, dns empty handling

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

hbd/server/configio.py

@@ -8,8 +8,12 @@ from datetime import datetime
 from ruamel.yaml import YAML
 
 _write_lock = threading.Lock()
-_yaml = YAML()
+
-_yaml.preserve_quotes = True
+
+def _make_yaml() -> YAML:
+    y = YAML()
+    y.preserve_quotes = True
+    return y
 
 # Top-level keys managed by the 'server' logical section
 _SERVER_KEYS = [
@@ -26,7 +30,7 @@ _DNS_KEYS = ["nsupdate_bin", "dyndomains", "dyndnshosts", "drophosts"]
 def read_roundtrip(path: str):
     """Load .hb.yaml with ruamel.yaml, preserving comments and ordering."""
     with open(path, "r", encoding="utf-8") as f:
-        return _yaml.load(f)
+        return _make_yaml().load(f)
 
 
 def write_config(path: str, data) -> None:
@@ -40,6 +44,10 @@ def write_config(path: str, data) -> None:
     with _write_lock:
         ts = datetime.now().strftime("%Y%m%d-%H%M%S")
         backup_path = f"{path}.bak.{ts}"
+        n = 0
+        while os.path.exists(backup_path):
+            n += 1
+            backup_path = f"{path}.bak.{ts}-{n}"
         if os.path.exists(path):
             with open(path, "rb") as src, open(backup_path, "wb") as dst:
                 dst.write(src.read())
@@ -47,9 +55,16 @@ def write_config(path: str, data) -> None:
         for old in backups[10:]:
             os.unlink(old)
         tmp = f"{path}.tmp"
+        try:
             with open(tmp, "w", encoding="utf-8") as f:
-            _yaml.dump(data, f)
+                _make_yaml().dump(data, f)
             os.replace(tmp, path)
+        except Exception:
+            try:
+                os.unlink(tmp)
+            except OSError:
+                pass
+            raise
 
 
 def list_backups(path: str) -> list:
@@ -75,7 +90,7 @@ def apply_structured_section(data, section: str, values: dict) -> None:
 
 def apply_yaml_section(data, section: str, yaml_text: str) -> None:
     """Replace the named logical section by parsing yaml_text."""
-    parsed = _yaml.load(yaml_text)
+    parsed = _make_yaml().load(yaml_text)
     if section == "notification_channels":
         data["notification_channels"] = parsed
     elif section == "thresholds":
@@ -87,5 +102,8 @@ def apply_yaml_section(data, section: str, yaml_text: str) -> None:
             for key in _DNS_KEYS:
                 if key in parsed:
                     data[key] = parsed[key]
+        else:
+            for key in _DNS_KEYS:
+                data.pop(key, None)
     else:
         raise ValueError(f"Unknown YAML section: {section!r}")

tests/test_configio.py

@@ -155,3 +155,8 @@ def test_apply_structured_section_unknown_raises(tmp_path):
     data = configio.read_roundtrip(str(f))
     with pytest.raises(ValueError, match="Unknown structured section"):
         configio.apply_structured_section(data, "nope", {"x": 1})
+
+
+def test_read_roundtrip_missing_file_raises(tmp_path):
+    with pytest.raises(FileNotFoundError):
+        configio.read_roundtrip(str(tmp_path / "nonexistent.yaml"))