feat: optional auth + backup/restore reliability fixes
Lint & Test / test (push) Failing after 29s

Auth is now optional: when `auth.api_keys` is empty, all endpoints are
open (no login screen, no Bearer tokens). Health endpoint reports
`auth_required` so the frontend knows which mode to use.

Backup/restore fixes:
- Auto-backup uses atomic writes (was `write_text`, risked corruption)
- Startup backup skipped if recent backup exists (<5 min cooldown),
  preventing rapid restarts from rotating out good backups
- Restore rejects all-empty backups to prevent accidental data wipes
- Store saves frozen after restore to prevent stale in-memory data
  from overwriting freshly-restored files before restart completes
- Missing stores during restore logged as warnings
- STORE_MAP completeness verified at startup against StorageConfig
This commit is contained in:
2026-03-23 14:50:25 +03:00
parent cd3137b0ec
commit 4975a74ff3
18 changed files with 189 additions and 67 deletions
@@ -3,7 +3,7 @@
import asyncio
import json
import os
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
@@ -18,6 +18,11 @@ DEFAULT_SETTINGS = {
"max_backups": 10,
}
# Skip the immediate-on-start backup if a recent backup exists within this window.
# Prevents rapid restarts from flooding the backup directory and rotating out
# good backups.
_STARTUP_BACKUP_COOLDOWN = timedelta(minutes=5)
class AutoBackupEngine:
"""Creates periodic backups of all configuration stores."""
@@ -83,11 +88,28 @@ class AutoBackupEngine:
self._task.cancel()
self._task = None
def _most_recent_backup_age(self) -> timedelta | None:
"""Return the age of the newest backup file, or None if no backups exist."""
files = list(self._backup_dir.glob("*.json"))
if not files:
return None
newest = max(files, key=lambda p: p.stat().st_mtime)
mtime = datetime.fromtimestamp(newest.stat().st_mtime, tz=timezone.utc)
return datetime.now(timezone.utc) - mtime
async def _backup_loop(self) -> None:
try:
# Perform first backup immediately on start
await self._perform_backup()
self._prune_old_backups()
# Skip immediate backup if a recent one already exists.
# Prevents rapid restarts (crashes, restores) from flooding the
# backup directory and rotating out good backups.
age = self._most_recent_backup_age()
if age is None or age > _STARTUP_BACKUP_COOLDOWN:
await self._perform_backup()
self._prune_old_backups()
else:
logger.info(
f"Skipping startup backup — most recent is only {age.total_seconds():.0f}s old"
)
interval_secs = self._settings["interval_hours"] * 3600
while True:
@@ -133,8 +155,7 @@ class AutoBackupEngine:
filename = f"ledgrab-autobackup-{timestamp}.json"
file_path = self._backup_dir / filename
content = json.dumps(backup, indent=2, ensure_ascii=False)
file_path.write_text(content, encoding="utf-8")
atomic_write_json(file_path, backup)
self._last_backup_time = now
logger.info(f"Auto-backup created: {filename}")
@@ -156,7 +177,6 @@ class AutoBackupEngine:
def get_settings(self) -> dict:
next_backup = None
if self._settings["enabled"] and self._last_backup_time:
from datetime import timedelta
next_backup = (
self._last_backup_time + timedelta(hours=self._settings["interval_hours"])
).isoformat()