From 5ce3e06e3476d0955b58fce423a0dd1bc073afba Mon Sep 17 00:00:00 2001 From: Hao Wang Date: Wed, 22 Jul 2026 21:26:51 +0800 Subject: [PATCH] fix(backup): serialize and atomically publish snapshots --- hermes_cli/backup.py | 232 +++++++++++++++++++--- tests/hermes_cli/test_backup_stability.py | 105 ++++++++++ 2 files changed, 310 insertions(+), 27 deletions(-) create mode 100644 tests/hermes_cli/test_backup_stability.py diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index 3a5b7cd4d3cf..43748fd3e207 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -15,8 +15,10 @@ import sqlite3 import sys import tempfile +import threading import time import zipfile +from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional @@ -84,6 +86,7 @@ # File names to skip (runtime state that's meaningless on another machine) _EXCLUDED_NAMES = { + ".backup.lock", "gateway.pid", "cron.pid", } @@ -132,6 +135,85 @@ _EXTERNAL_PREFIX = "_external/" +class BackupInProgressError(RuntimeError): + """Raised when another process already owns the Hermes backup slot.""" + + +class _SQLiteSnapshotError(RuntimeError): + pass + + +@contextmanager +def _backup_operation_lock(hermes_home: Path, timeout_seconds: float = 0.25): + """Acquire one cross-process backup slot for full and quick snapshots.""" + lock_path = hermes_home / ".backup.lock" + lock_path.parent.mkdir(parents=True, exist_ok=True) + handle = lock_path.open("a+b") + acquired = False + deadline = time.monotonic() + max(0.0, timeout_seconds) + try: + if os.name == "nt": + import msvcrt + + if lock_path.stat().st_size == 0: + handle.write(b" ") + handle.flush() + while True: + try: + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + acquired = True + break + except (OSError, PermissionError): + if time.monotonic() >= deadline: + raise BackupInProgressError("another Hermes backup is already running") + time.sleep(0.05) + else: + import fcntl + + while True: + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + acquired = True + break + except (BlockingIOError, OSError): + if time.monotonic() >= deadline: + raise BackupInProgressError("another Hermes backup is already running") + time.sleep(0.05) + + yield + finally: + if acquired: + try: + if os.name == "nt": + import msvcrt + + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + except (OSError, PermissionError): + pass + handle.close() + + +@contextmanager +def _atomic_output_path(final_path: Path): + """Yield a hidden sibling path and publish it only after a clean close.""" + partial_path = final_path.with_name( + f".{final_path.name}.{os.getpid()}-{threading.get_ident()}.partial" + ) + partial_path.unlink(missing_ok=True) + try: + yield partial_path + os.replace(partial_path, final_path) + except BaseException: + partial_path.unlink(missing_ok=True) + raise + + def _collect_memory_provider_external_paths() -> List[Path]: """Return existing absolute paths the active memory provider stores outside HERMES_HOME, resolved from config only (no network, no init). @@ -509,6 +591,17 @@ def run_backup(args) -> None: print(f"Error: Hermes home directory not found at {hermes_root}") sys.exit(1) + try: + with _backup_operation_lock(hermes_root): + _run_backup_locked(args, hermes_root) + except BackupInProgressError as exc: + print(f"Error: {exc}") + raise SystemExit(2) from exc + + +def _run_backup_locked(args, hermes_root: Path) -> None: + """Write a full backup while the cross-process backup slot is held.""" + # Determine output path if args.output: out_path = Path(args.output).expanduser().resolve() @@ -528,6 +621,8 @@ def run_backup(args) -> None: out_path.parent.mkdir(parents=True, exist_ok=True) # Collect files + scan_started = time.monotonic() + logger.info("backup phase=scan status=started") print(f"Scanning {display_hermes_home()} ...") files_to_add: list[tuple[Path, Path]] = [] # (absolute, relative) skipped_dirs = set() @@ -582,18 +677,30 @@ def run_backup(args) -> None: external_to_add.append((fpath, arcname)) if not files_to_add and not external_to_add: + logger.info( + "backup phase=scan status=empty duration_ms=%.1f", + (time.monotonic() - scan_started) * 1000, + ) print("No files to back up.") return # Create the zip file_count = len(files_to_add) + len(external_to_add) + logger.info( + "backup phase=scan status=complete duration_ms=%.1f files=%d", + (time.monotonic() - scan_started) * 1000, + file_count, + ) + logger.info("backup phase=archive status=started files=%d", file_count) print(f"Backing up {file_count} files ...") total_bytes = 0 errors = [] t0 = time.monotonic() - with zipfile.ZipFile(out_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as zf: + with _atomic_output_path(out_path) as archive_path, zipfile.ZipFile( + archive_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6 + ) as zf: for i, (abs_path, rel_path) in enumerate(files_to_add, 1): try: # Safe copy for SQLite databases (handles WAL mode) @@ -624,6 +731,11 @@ def run_backup(args) -> None: # Progress every 500 files if i % 500 == 0: print(f" {i}/{file_count} files ...") + logger.info( + "backup phase=archive status=progress completed=%d total=%d", + i, + file_count, + ) # External memory-provider state, stored under the ``_external/`` arc # prefix. These never include ``.db`` files in practice (config/env @@ -638,6 +750,13 @@ def run_backup(args) -> None: elapsed = time.monotonic() - t0 zip_size = out_path.stat().st_size + logger.info( + "backup phase=archive status=complete duration_ms=%.1f files=%d errors=%d bytes=%d", + elapsed * 1000, + file_count, + len(errors), + zip_size, + ) # Summary print() @@ -1013,6 +1132,23 @@ def create_quick_snapshot( hermes_home: Optional[Path] = None, keep: Optional[int] = None, max_file_size: Optional[int] = None, +) -> Optional[str]: + """Create one atomic quick snapshot while holding the shared backup slot.""" + home = hermes_home or get_hermes_home() + with _backup_operation_lock(home): + return _create_quick_snapshot_locked( + label=label, + hermes_home=home, + keep=keep, + max_file_size=max_file_size, + ) + + +def _create_quick_snapshot_locked( + label: Optional[str] = None, + hermes_home: Optional[Path] = None, + keep: Optional[int] = None, + max_file_size: Optional[int] = None, ) -> Optional[str]: """Create a quick state snapshot of critical files. @@ -1058,9 +1194,17 @@ def _too_large(path: Path, rel_name: str) -> bool: return True ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") - snap_id = f"{ts}-{label}" if label else ts + base_snap_id = f"{ts}-{label}" if label else ts + snap_id = base_snap_id + suffix = 2 + while (root / snap_id).exists(): + snap_id = f"{base_snap_id}-{suffix}" + suffix += 1 snap_dir = root / snap_id - snap_dir.mkdir(parents=True, exist_ok=True) + staging_dir = root / f".{snap_id}.{os.getpid()}.partial" + shutil.rmtree(staging_dir, ignore_errors=True) + staging_dir.mkdir(parents=True, exist_ok=False) + logger.info("quick snapshot phase=copy status=started id=%s", snap_id) manifest: Dict[str, int] = {} # rel_path -> file size failed_dbs: list[str] = [] # present *.db that could not be snapshotted @@ -1092,7 +1236,7 @@ def _too_large(path: Path, rel_name: str) -> bool: if sub.suffix == ".db": oversized_skipped.append(sub_rel) continue - dst = snap_dir / sub_rel + dst = staging_dir / sub_rel dst.parent.mkdir(parents=True, exist_ok=True) try: # Route SQLite DBs through the WAL-safe backup() path so a @@ -1126,7 +1270,7 @@ def _too_large(path: Path, rel_name: str) -> bool: oversized_skipped.append(rel) continue - dst = snap_dir / rel + dst = staging_dir / rel dst.parent.mkdir(parents=True, exist_ok=True) try: @@ -1167,7 +1311,7 @@ def _too_large(path: Path, rel_name: str) -> bool: ) if not manifest: - shutil.rmtree(snap_dir, ignore_errors=True) + shutil.rmtree(staging_dir, ignore_errors=True) if failed_dbs: # Distinguish "nothing to snapshot" from "state.db present but unreadable" print( @@ -1187,9 +1331,11 @@ def _too_large(path: Path, rel_name: str) -> bool: "failed_dbs": failed_dbs, "oversized_skipped": oversized_skipped, } - with open(snap_dir / "manifest.json", "w", encoding="utf-8") as f: + with open(staging_dir / "manifest.json", "w", encoding="utf-8") as f: json.dump(meta, f, indent=2) + os.replace(staging_dir, snap_dir) + # Auto-prune. Defaults preserve historical manual /snapshot behavior; callers # with known high-churn safety snapshots (for example pre-update) can pass a # smaller keep value so large state.db copies do not accumulate indefinitely. @@ -1216,7 +1362,12 @@ def _too_large(path: Path, rel_name: str) -> bool: len(failed_dbs), len(oversized_skipped), ) - logger.info("State snapshot created: %s (%d files)", snap_id, len(manifest)) + logger.info( + "quick snapshot phase=copy status=complete id=%s files=%d bytes=%d", + snap_id, + len(manifest), + sum(manifest.values()), + ) return snap_id @@ -1231,7 +1382,7 @@ def list_quick_snapshots( results = [] for d in sorted(root.iterdir(), reverse=True): - if not d.is_dir(): + if not d.is_dir() or d.name.startswith(".") or d.name.endswith(".partial"): continue manifest_path = d / "manifest.json" if manifest_path.exists(): @@ -1440,7 +1591,11 @@ def _prune_quick_snapshots(root: Path, keep: int = _QUICK_DEFAULT_KEEP) -> int: return 0 dirs = sorted( - (d for d in root.iterdir() if d.is_dir()), + ( + d + for d in root.iterdir() + if d.is_dir() and not d.name.startswith(".") and not d.name.endswith(".partial") + ), key=lambda d: d.name, reverse=True, ) @@ -1482,12 +1637,24 @@ def run_quick_backup(args) -> None: # --------------------------------------------------------------------------- def _write_full_zip_backup(out_path: Path, hermes_root: Path) -> Optional[Path]: + """Single-flight wrapper for automatic full zip backups.""" + try: + with _backup_operation_lock(hermes_root): + return _write_full_zip_backup_locked(out_path, hermes_root) + except BackupInProgressError as exc: + logger.warning("Full-zip backup skipped: %s", exc) + return None + + +def _write_full_zip_backup_locked(out_path: Path, hermes_root: Path) -> Optional[Path]: """Write a full zip snapshot of ``hermes_root`` to ``out_path``. Uses the same exclusion rules and SQLite safe-copy as :func:`run_backup`. Returns the output path on success, None on failure (nothing to back up, or write error — caller should surface the outcome but not raise). """ + scan_started = time.monotonic() + logger.info("automatic backup phase=scan status=started") files_to_add: list[tuple[Path, Path]] = [] try: for dirpath, dirnames, filenames in os.walk(hermes_root, followlinks=False): @@ -1513,10 +1680,18 @@ def _write_full_zip_backup(out_path: Path, hermes_root: Path) -> Optional[Path]: if not files_to_add: return None - sqlite_snapshot_failed = False + logger.info( + "automatic backup phase=scan status=complete duration_ms=%.1f files=%d", + (time.monotonic() - scan_started) * 1000, + len(files_to_add), + ) + + archive_started = time.monotonic() try: - with zipfile.ZipFile(out_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as zf: - for abs_path, rel_path in files_to_add: + with _atomic_output_path(out_path) as archive_path, zipfile.ZipFile( + archive_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6 + ) as zf: + for index, (abs_path, rel_path) in enumerate(files_to_add, 1): try: if abs_path.suffix == ".db": # Stage the snapshot alongside the output zip so that the @@ -1533,8 +1708,7 @@ def _write_full_zip_backup(out_path: Path, hermes_root: Path) -> Optional[Path]: "Full-zip backup aborted: SQLite snapshot failed for %s", rel_path, ) - sqlite_snapshot_failed = True - break + raise _SQLiteSnapshotError(str(rel_path)) zf.write(tmp_db, arcname=str(rel_path)) finally: tmp_db.unlink(missing_ok=True) @@ -1543,21 +1717,25 @@ def _write_full_zip_backup(out_path: Path, hermes_root: Path) -> Optional[Path]: except (PermissionError, OSError, ValueError) as exc: logger.debug("Skipping %s in zip backup: %s", rel_path, exc) continue - except OSError as exc: + if index % 500 == 0: + logger.info( + "automatic backup phase=archive status=progress completed=%d total=%d", + index, + len(files_to_add), + ) + except (OSError, _SQLiteSnapshotError) as exc: logger.warning("Full-zip backup: zip write failed: %s", exc) - # Best-effort cleanup of partial file - try: - out_path.unlink(missing_ok=True) - except OSError: - pass + # ``_atomic_output_path`` already removed the hidden partial. Do not + # unlink ``out_path`` here: it may be a previous valid backup that the + # atomic publisher deliberately preserved. return None - if sqlite_snapshot_failed: - try: - out_path.unlink(missing_ok=True) - except OSError: - pass - return None + logger.info( + "automatic backup phase=archive status=complete duration_ms=%.1f files=%d bytes=%d", + (time.monotonic() - archive_started) * 1000, + len(files_to_add), + out_path.stat().st_size, + ) return out_path diff --git a/tests/hermes_cli/test_backup_stability.py b/tests/hermes_cli/test_backup_stability.py new file mode 100644 index 000000000000..d461d507220b --- /dev/null +++ b/tests/hermes_cli/test_backup_stability.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from hermes_cli.backup import ( + BackupInProgressError, + _atomic_output_path, + _backup_operation_lock, + _write_full_zip_backup, + create_quick_snapshot, + list_quick_snapshots, +) + + +def test_backup_lock_rejects_a_second_operation(tmp_path) -> None: + home = tmp_path / ".hermes" + home.mkdir() + + with _backup_operation_lock(home): + with pytest.raises(BackupInProgressError): + with _backup_operation_lock(home, timeout_seconds=0): + raise AssertionError("second backup unexpectedly acquired the lock") + + +def test_atomic_output_publishes_only_after_clean_close(tmp_path) -> None: + final = tmp_path / "backup.zip" + final.write_bytes(b"previous") + + with _atomic_output_path(final) as partial: + partial.write_bytes(b"complete") + assert final.read_bytes() == b"previous" + + assert final.read_bytes() == b"complete" + assert not partial.exists() + + +def test_atomic_output_keeps_previous_file_after_failure(tmp_path) -> None: + final = tmp_path / "backup.zip" + final.write_bytes(b"previous") + + with pytest.raises(RuntimeError): + with _atomic_output_path(final) as partial: + partial.write_bytes(b"incomplete") + raise RuntimeError("compression failed") + + assert final.read_bytes() == b"previous" + assert not partial.exists() + + +def test_quick_snapshot_is_published_with_manifest(tmp_path, monkeypatch) -> None: + home = tmp_path / ".hermes" + home.mkdir() + (home / "config.yaml").write_text("model: {}\n", encoding="utf-8") + published: list[tuple[Path, Path]] = [] + + from hermes_cli import backup + + real_replace = backup.os.replace + + def replace(source, destination) -> None: + source_path = Path(source) + destination_path = Path(destination) + if destination_path.parent == home / "state-snapshots": + assert source_path.name.endswith(".partial") + assert (source_path / "manifest.json").is_file() + assert not destination_path.exists() + published.append((source_path, destination_path)) + real_replace(source, destination) + + monkeypatch.setattr(backup.os, "replace", replace) + snapshot_id = create_quick_snapshot(hermes_home=home) + + assert snapshot_id is not None + assert len(published) == 1 + manifest = json.loads( + (home / "state-snapshots" / snapshot_id / "manifest.json").read_text(encoding="utf-8") + ) + assert manifest["id"] == snapshot_id + assert manifest["files"] == {"config.yaml": 10} + + +def test_quick_snapshot_listing_ignores_partial_directories(tmp_path) -> None: + home = tmp_path / ".hermes" + partial = home / "state-snapshots" / ".unfinished.1.partial" + partial.mkdir(parents=True) + (partial / "manifest.json").write_text('{"id":"unfinished"}', encoding="utf-8") + + assert list_quick_snapshots(hermes_home=home) == [] + + +def test_failed_automatic_backup_preserves_previous_archive(tmp_path, monkeypatch) -> None: + home = tmp_path / ".hermes" + home.mkdir() + (home / "state.db").write_bytes(b"not-a-database") + archive = tmp_path / "automatic.zip" + archive.write_bytes(b"previous-valid-backup") + + monkeypatch.setattr("hermes_cli.backup._safe_copy_db", lambda _src, _dst: False) + + assert _write_full_zip_backup(archive, home) is None + assert archive.read_bytes() == b"previous-valid-backup" + assert list(tmp_path.glob(".*.partial")) == []