Function snapshot_paths

openkb/mutation.py:308–368

Snapshot final KB paths before a mutation starts. ``hardlink_dirs`` marks directories whose backup may be hardlinks instead of copies (O(1), no per-file byte copy). A directory is only safe to list here if every writer into it is either atomic temp+replace (new inode, so the hardlin

(
    kb_dir: Path,
    paths: list[Path],
    *,
    operation: str,
    details: dict | None = None,
    hardlink_dirs: set[Path] | None = None,
)

def snapshot_paths(
    kb_dir: Path,
    paths: list[Path],
    *,
    operation: str,
    details: dict | None = None,
    hardlink_dirs: set[Path] | None = None,
) -> MutationSnapshot:
    """Snapshot final KB paths before a mutation starts.

    ``hardlink_dirs`` marks directories whose backup may be hardlinks instead
    of copies (O(1), no per-file byte copy). A directory is only safe to list
    here if every writer into it is either atomic temp+replace (new inode, so
    the hardlink backup keeps the old bytes) or append-only. This is the
    required caller contract for hardlinked dirs; any in-place writer into one
    of those trees would silently corrupt the backup and make rollback a no-op
    for that file.
    """
    kb_dir = kb_dir.resolve()
    hardlink_resolved = {p.resolve() for p in (hardlink_dirs or ())}
    journal_id = uuid.uuid4().hex
    backup_dir = kb_dir / ".openkb" / "staging" / f"rollback-{journal_id}"
    backup_dir.mkdir(parents=True, exist_ok=False)
    snapshot = MutationSnapshot(
        kb_dir=kb_dir,
        backup_dir=backup_dir,
        journal_path=kb_dir / ".openkb" / "journal" / f"{journal_id}.json",
        operation=operation,
        details=details or {},
    )
    try:
        for path in paths:
            target = path.resolve()
            if target in snapshot.entries:
                continue
            if not target.exists():
                snapshot.entries[target] = None
                continue
            rel = target.relative_to(kb_dir)
            backup = backup_dir / rel
            backup.parent.mkdir(parents=True, exist_ok=True)
            if target.is_dir():
                if target in hardlink_resolved:
                    shutil.copytree(target, backup, copy_function=_hardlink_or_copy)
                    snapshot.hardlinked_dirs.add(target)
                else:
                    shutil.copytree(target, backup)
            else:
                _copy_file_atomic(target, backup)
            snapshot.entries[target] = backup
        # The active journal is the recovery signal: once this exists, a future
        # process can restore every recorded target even if the current one exits.
        snapshot.write_journal("active")
    except Exception:
        # Partial snapshot: backup_dir exists on disk but no journal was
        # written. recover_pending_journals only scans journals, so remove the
        # orphan backup here — otherwise it leaks forever with nothing able to
        # reach or clean it.