Class MutationSnapshot

openkb/mutation.py:129–257

Snapshot of final KB paths touched by a mutation attempt.

@dataclass
class MutationSnapshot:
    """Snapshot of final KB paths touched by a mutation attempt."""

    kb_dir: Path
    backup_dir: Path
    journal_path: Path
    operation: str
    details: dict = field(default_factory=dict)
    entries: dict[Path, Path | None] = field(default_factory=dict)
    attempts: int = 0
    # Dirs whose backup was hardlinked (in-process only; not persisted, so a
    # crash-rebuilt snapshot leaves this empty and rollback falls back to the
    # safe full-copy path). Drives O(touched) rollback via inode-diff restore.
    hardlinked_dirs: set[Path] = field(default_factory=set)

    def _journal_data(self, status: str) -> dict:
        return {
            "version": 1,
            "operation": self.operation,
            "status": status,
            "kb_dir": str(self.kb_dir),
            "backup_dir": str(self.backup_dir),
            "details": self.details,
            "attempts": self.attempts,
            "entries": [
                {
                    "target": str(target),
                    "backup": str(backup) if backup is not None else None,
                }
                for target, backup in self.entries.items()
            ],
        }

    def write_journal(self, status: str) -> None:
        self.journal_path.parent.mkdir(parents=True, exist_ok=True)
        atomic_write_json(self.journal_path, self._journal_data(status))

    def mark_committed(self) -> None:
        """Mark the journal committed without removing the backup.

        Call this the instant the mutation is durably applied (e.g. the
        registry write has landed) so a subsequent
        :func:`recover_pending_journals` discards the journal instead of
        rolling it back. This is the commit signal; :meth:`discard` is the
        post-commit cleanup that also removes the backup dir and journal
        file and must itself be best-effort — it runs *after* the commit
        point and its failure must never trigger a rollback.
        """
        self.write_journal("committed")

    def track_new(self, paths: list[Path]) -> None:
        """Register paths created *after* the snapshot for removal on rollback.

        Some artifacts get their final name only once the mutation runs — the
        PageIndex ``{doc_id}`` blob under ``.openkb/files`` is named by indexing,
        which happens after :func:`snapshot_paths`. Rather than eagerly
        snapshotting the whole append-only blob store up front (an ``os.link``
        per existing blob on *every* add — O(total blobs), not O(this doc)),