orchestrata_ultimate.refactor.codemod.txn

Reference › orchestrata_ultimate › refactor › codemod › txn

The write path of the codemod engine: it puts a ChangeSet onto disk atomically, and puts the old bytes back if anything fails. workspace is its only importer; it uses report and the event types from observability.

commit_changes(root, changes, dry_run, on_event) returns the lists of files written and rolled back, and returns two empty lists on a dry run. Otherwise it validates that every path stays inside the repository (PathEscapeError), snapshots each target and checks that it has not been edited since load (ConcurrentEditError, also raised when a target has vanished; symlinks are refused with SymlinkAttackError and locked files with FileLockError). It then writes each file through _write_atomic and _replace_file, which write a temporary file, keep the target's mode and group, fsync and rename it over the target. After _batch_fsync it re-parses what was written (_verify_post_write, failing with CommitVerificationError). If any step raises, _rollback restores only the files this commit already wrote, byte for byte with their mode and group, emits ROLLBACK_TRIGGERED with any per-file restore failures, and re-raises the original exception. A group change that the filesystem rejects is logged and the commit continues.

At a glance

Class diagram

Dependency diagram

API

Classes

CommitVerificationError

Bases: RuntimeError · Decorators: traces(spec='110', story='US-110-001', reqs=['REQ-110-014']) · Line 31

Raised when post-write verification fails.

ConcurrentEditError

Bases: RuntimeError · Decorators: traces(spec='110', story='US-110-001', reqs=['REQ-110-015']) · Line 41

Raised when a file was modified after workspace load.

SymlinkAttackError

Bases: RuntimeError · Decorators: traces(spec='110', story='US-110-001', reqs=['REQ-110-015']) · Line 60

Raised when a symlink is detected where a regular file is expected.

FileLockError

Bases: RuntimeError · Decorators: traces(spec='110', story='US-110-001', reqs=['REQ-110-015']) · Line 71

Raised when a file is locked by another process.

PathEscapeError

Bases: RuntimeError · Decorators: traces(spec='110', story='US-110-001', reqs=['REQ-110-021']) · Line 80

Raised when a path escapes the repository root.

_FileSnapshot

Decorators: traces(spec='110', story='US-110-001', reqs=['REQ-110-014']) · Line 90

Pre-commit snapshot of a single file (for rollback).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/txn.py

"""Transactional write-back with atomic commits and rollback.

Implements SPEC-110 FR-005: atomic per-file writes via tempfile + fsync + rename,
batch directory fsync, post-write re-parse verification, full rollback on any
failure, concurrent-edit guards (mtime + xxhash3_64 content hash), symlink
attack prevention (O_NOFOLLOW), and advisory locking (fcntl.flock).
"""

from __future__ import annotations

import ast
import errno
import fcntl
import logging
import os
import tempfile
from collections.abc import Callable
from pathlib import Path
from typing import Any

import xxhash

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.observability.events import CodemodEventType
from orchestrata_ultimate.refactor.codemod.report import ChangeEntry, ChangeSet

logger = logging.getLogger(__name__)


@traces(spec="110", story="US-110-001", reqs=["REQ-110-014"])
class CommitVerificationError(RuntimeError):
    """Raised when post-write verification fails."""

    def __init__(self, path: str, reason: str) -> None:
        self.path = path
        self.reason = reason
        super().__init__(f"Post-write verification failed for {path}: {reason}")


@traces(spec="110", story="US-110-001", reqs=["REQ-110-015"])
class ConcurrentEditError(RuntimeError):
    """Raised when a file was modified after workspace load."""

    def __init__(self, path: str, vanished: bool = False) -> None:
        self.path = path
        if vanished:
            super().__init__(
                f"File {path}: the file no longer exists on disk (deleted "
                "after the concurrent-edit check, before the write)"
            )
            return
        super().__init__(
            f"File {path}: the bytes on disk differ from the tree this change "
            "was computed against (modified after it was loaded, or a "
            "previous change to it was not committed to disk)"
        )


@traces(spec="110", story="US-110-001", reqs=["REQ-110-015"])
class SymlinkAttackError(RuntimeError):
    """Raised when a symlink is detected where a regular file is expected."""

    def __init__(self, path: str) -> None:
        self.path = path
        super().__init__(
            f"Security alert: {path} is a symlink — refusing to follow (O_NOFOLLOW)"
        )


@traces(spec="110", story="US-110-001", reqs=["REQ-110-015"])
class FileLockError(RuntimeError):
    """Raised when a file is locked by another process."""

    def __init__(self, path: str) -> None:
        self.path = path
        super().__init__(f"File {path} is locked by another process")


@traces(spec="110", story="US-110-001", reqs=["REQ-110-021"])
class PathEscapeError(RuntimeError):
    """Raised when a path escapes the repository root."""

    def __init__(self, path: str, root: str) -> None:
        self.path = path
        self.root = root
        super().__init__(f"Path {path} escapes repository root {root}")


@traces(spec="110", story="US-110-001", reqs=["REQ-110-014"])
class _FileSnapshot:
    """Pre-commit snapshot of a single file (for rollback)."""

    __slots__ = ("path", "original_bytes", "mtime_ns", "content_hash", "mode", "gid")

    def __init__(
        self,
        path: str,
        original_bytes: bytes,
        mtime_ns: int,
        content_hash: str,
        mode: int,
        gid: int = -1,
    ) -> None:
        self.path = path
        self.original_bytes = original_bytes
        self.mtime_ns = mtime_ns
        self.content_hash = content_hash
        # st_mode permission bits (incl. setuid/setgid/sticky) at snapshot time
        self.mode = mode
        # st_gid at snapshot time (-1 = unknown, leave the group alone)
        self.gid = gid


@traces(
    spec="110",
    story="US-110-001",
    reqs=[
        "REQ-110-014",
        "REQ-110-015",
        "REQ-110-016",
        "REQ-110-021",
        "REQ-110-025",
    ],
)
def commit_changes(
    root: Path,
    changes: ChangeSet,
    dry_run: bool = True,
    on_event: Callable[[CodemodEventType, dict[str, Any]], None] | None = None,
) -> tuple[list[str], list[str]]:
    """Commit changes to disk with atomic writes and rollback.

    Args:
        root: Repository root path.
        changes: ChangeSet containing entries to write.
        dry_run: If True, return empty lists without writing.
        on_event: Optional lifecycle sink (REQ-110-025). Called with
            ``ROLLBACK_TRIGGERED`` when a failed write restores the
            snapshot, before the original exception is re-raised.

    Returns:
        Tuple of (files_written, files_rolled_back).

    Raises:
        ConcurrentEditError: If a file was modified after workspace load.
        SymlinkAttackError: If a file is a symlink.
        FileLockError: If a file is locked by another process.
        CommitVerificationError: If post-write re-parse fails.
        PathEscapeError: If a path escapes the repository root.
    """
    if dry_run:
        return ([], [])

    resolved_root = root.resolve()
    _validate_paths(resolved_root, changes)

    snapshot = _check_concurrent_edits(resolved_root, changes)

    files_written: list[str] = []
    try:
        for entry in changes.entries:
            _write_atomic(resolved_root, entry)
            files_written.append(entry.path)

        _batch_fsync(resolved_root, files_written)
        _verify_post_write(resolved_root, files_written)

        return (files_written, [])

    except Exception as exc:
        # Only entries that completed their rename are restored: every other
        # snapshot entry is unchanged on disk by construction.
        files_rolled_back, rollback_failures = _rollback(
            resolved_root,
            {p: snapshot[p] for p in files_written},
        )
        for failure in rollback_failures:
            exc.add_note(f"rollback failed: {failure}")
        if on_event is not None:
            try:
                on_event(
                    CodemodEventType.ROLLBACK_TRIGGERED,
                    {
                        "files_rolled_back": len(files_rolled_back),
                        "error_type": type(exc).__name__,
                        "message": str(exc),
                        "rollback_failures": rollback_failures,
                    },
                )
            except Exception:
                # A raising sink must not mask the commit failure.
                logger.exception("on_event sink raised; original error kept")
        logger.error(
            "Commit failed, rolled back %d files",
            len(files_rolled_back),
            exc_info=True,
        )
        raise


@traces(spec="110", story="US-110-001", reqs=["REQ-110-021"])
def _validate_paths(root: Path, changes: ChangeSet) -> None:
    """Validate all change paths are within the repository root.

    Args:
        root: Resolved repository root.
        changes: ChangeSet to validate.

    Raises:
        PathEscapeError: If any path escapes the root.
    """
    for entry in changes.entries:
        resolved = (root / entry.path).resolve()
        try:
            resolved.relative_to(root)
        except ValueError:
            raise PathEscapeError(entry.path, str(root))


@traces(
    spec="110",
    story="US-110-001",
    reqs=["REQ-110-015", "REQ-110-021"],
)
def _check_concurrent_edits(
    root: Path,
    changes: ChangeSet,
) -> dict[str, _FileSnapshot]:
    """Check if files were modified since workspace load and build rollback snapshot.

    Compares the current file on disk against the original_tree.code from
    the ChangeEntry. If they differ, the file was modified after the
    workspace was loaded.

    Uses O_NOFOLLOW to prevent symlink attacks and fcntl.flock for
    advisory locking.

    Args:
        root: Resolved repository root.
        changes: ChangeSet with original_tree for comparison.

    Returns:
        Mapping of relative path to _FileSnapshot (for rollback).

    Raises:
        SymlinkAttackError: If a file is a symlink.
        FileLockError: If a file is locked by another process.
        ConcurrentEditError: If a file was modified concurrently.
    """
    snapshot: dict[str, _FileSnapshot] = {}
    for entry in changes.entries:
        file_path = root / entry.path
        str_path = str(file_path)
        expected_bytes = entry.original_tree.code
        expected_hash = xxhash.xxh3_64(expected_bytes).hexdigest()

        if os.path.islink(str_path):
            raise SymlinkAttackError(entry.path)

        fd = -1
        try:
            fd = os.open(str_path, os.O_RDONLY | os.O_NOFOLLOW)
        except OSError as exc:
            if exc.errno == 40:
                raise SymlinkAttackError(entry.path)
            raise

        try:
            try:
                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except OSError:
                raise FileLockError(entry.path)

            try:
                stat = os.fstat(fd)
                file_size = stat.st_size
                current_bytes = os.read(fd, file_size) if file_size > 0 else b""
                current_hash = xxhash.xxh3_64(current_bytes).hexdigest()
                if current_hash != expected_hash:
                    raise ConcurrentEditError(entry.path)

                snapshot[entry.path] = _FileSnapshot(
                    path=entry.path,
                    original_bytes=current_bytes,
                    mtime_ns=stat.st_mtime_ns,
                    content_hash=current_hash,
                    mode=stat.st_mode & 0o7777,
                    gid=stat.st_gid,
                )
            finally:
                fcntl.flock(fd, fcntl.LOCK_UN)
        finally:
            os.close(fd)

    return snapshot


@traces(spec="110", story="US-110-001", reqs=["REQ-110-014"])
def _replace_file(file_path: Path, data: bytes, mode: int, gid: int = -1) -> None:
    """Replace ``file_path`` with ``data`` via tempfile + fsync + rename.

    mkstemp creates the temp file 0600 and the rename would carry that over
    the target, so ``mode`` is applied with ``fchmod``. The kernel clears
    setuid/setgid on write, so ``fchmod`` runs AFTER the write. The rename
    swaps in an inode owned by the writer, so the target's group is restored
    with ``fchown(fd, -1, gid)`` BEFORE ``fchmod`` (chown also clears
    setuid/setgid). An ``OSError`` from fchown with ``EPERM`` (not allowed)
    or ``EINVAL`` (gid unmapped in a user namespace) means the group cannot
    be preserved: it is logged and the commit continues; the uid cannot be
    preserved without root, so the file ends up owned by the writer. A failure at any
    step unlinks the temp file before the exception propagates.

    Args:
        file_path: Target path (its directory receives the temp file).
        data: Bytes the target holds after the rename.
        mode: Permission bits (``st_mode & 0o7777``) the target ends with.
        gid: Group the target keeps (-1 leaves the writer's group).
    """
    fd = -1
    tmp_path = ""
    try:
        fd, tmp_path = tempfile.mkstemp(dir=str(file_path.parent), suffix=".tmp")
        view = memoryview(data)
        while view:
            # os.write may be short (e.g. RLIMIT_FSIZE); looping surfaces the
            # real error instead of renaming a truncated file over the target.
            view = view[os.write(fd, view) :]
        if gid >= 0:
            try:
                os.fchown(fd, -1, gid)
            except OSError as exc:
                if exc.errno not in (errno.EPERM, errno.EINVAL):
                    raise
                # EPERM: the writer may not hand the file to ``gid``. EINVAL:
                # ``gid`` is unmapped (user namespaces: rootless containers,
                # bubblewrap). Either way the group cannot be preserved.
                logger.warning(
                    "cannot preserve group of %s: wanted gid %d, file keeps gid %d",
                    file_path,
                    gid,
                    os.fstat(fd).st_gid,
                )
        os.fchmod(fd, mode)
        os.fsync(fd)
        os.close(fd)
        fd = -1
        os.rename(tmp_path, str(file_path))
    except BaseException:
        if fd >= 0:
            os.close(fd)
        if tmp_path:
            try:
                os.unlink(tmp_path)
            except FileNotFoundError:
                pass
        raise


@traces(spec="110", story="US-110-001", reqs=["REQ-110-014", "REQ-110-015"])
def _write_atomic(root: Path, entry: ChangeEntry) -> None:
    """Write a single file atomically via tempfile + fsync + rename.

    Writes the new content to a temporary file in the target's directory
    and renames it over the target, keeping the target's permission bits.
    The target must exist: ``commit_changes`` verified it in
    ``_check_concurrent_edits``, so a target that vanished since then was
    deleted concurrently and is refused rather than silently recreated.

    Args:
        root: Resolved repository root.
        entry: ChangeEntry containing the mutated tree.

    Raises:
        ConcurrentEditError: If the target no longer exists.
    """
    file_path = root / entry.path
    try:
        target_stat = os.stat(file_path)
    except FileNotFoundError:
        raise ConcurrentEditError(entry.path, vanished=True)

    _replace_file(
        file_path,
        entry.mutated_tree.code,
        target_stat.st_mode & 0o7777,
        target_stat.st_gid,
    )


@traces(spec="110", story="US-110-001", reqs=["REQ-110-014"])
def _batch_fsync(root: Path, files_written: list[str]) -> None:
    """Fsync directory file descriptors after all writes complete.

    Collects unique parent directories from all written files and fsyncs
    each directory fd once, reducing I/O overhead compared to per-file fsync.

    Args:
        root: Resolved repository root.
        files_written: List of relative paths that were written.
    """
    dirs_to_sync: set[Path] = set()
    for path_str in files_written:
        dir_path = (root / path_str).parent
        dirs_to_sync.add(dir_path)

    for dir_path in dirs_to_sync:
        fd = -1
        try:
            fd = os.open(str(dir_path), os.O_RDONLY)
            os.fsync(fd)
        finally:
            if fd >= 0:
                os.close(fd)


@traces(spec="110", story="US-110-001", reqs=["REQ-110-014"])
def _verify_post_write(root: Path, files_written: list[str]) -> None:
    """Re-parse written files to verify correctness.

    Reads each written file from disk and attempts an ``ast.parse`` to
    confirm the written bytes form valid Python.

    Args:
        root: Resolved repository root.
        files_written: List of relative paths that were written.

    Raises:
        CommitVerificationError: If any file fails to re-parse.
    """
    for path_str in files_written:
        file_path = root / path_str
        try:
            written_bytes = file_path.read_bytes()
            ast.parse(written_bytes.decode("utf-8"))
        except Exception as exc:
            raise CommitVerificationError(path_str, str(exc))


@traces(spec="110", story="US-110-001", reqs=["REQ-110-014", "REQ-110-021"])
def _rollback(
    root: Path,
    snapshot: dict[str, _FileSnapshot],
) -> tuple[list[str], list[str]]:
    """Restore files from pre-commit snapshot.

    Writes the original bytes back to each given file through the same
    tempfile + rename path as the forward write (so a read-only target is
    restored too) and re-applies the snapshot's recorded mode and group,
    providing byte-exact rollback on any commit failure. A restore that
    fails is logged and collected; it never stops the remaining restores.

    Args:
        root: Resolved repository root.
        snapshot: Snapshots of the files that were written (only these are
            restored).

    Returns:
        Tuple of (relative paths rolled back, ``"path: error"`` failures).
    """
    files_rolled_back: list[str] = []
    failures: list[str] = []
    for path_str, snap in snapshot.items():
        file_path = root / path_str
        try:
            _replace_file(file_path, snap.original_bytes, snap.mode, snap.gid)
        except Exception as exc:
            logger.error("Rollback of %s failed: %s", path_str, exc, exc_info=True)
            failures.append(f"{path_str}: {type(exc).__name__}: {exc}")
            continue
        files_rolled_back.append(path_str)
        logger.info("Rolled back %s", path_str)
    return files_rolled_back, failures