diff --git a/src/docforge/_fs_safety.py b/src/docforge/_fs_safety.py index 81c9f3b..7ad9c9a 100644 --- a/src/docforge/_fs_safety.py +++ b/src/docforge/_fs_safety.py @@ -14,6 +14,7 @@ from typing import Protocol, cast from .errors import DocForgeError +RENAME_NOREPLACE = 1 RENAME_EXCHANGE = 2 @@ -32,9 +33,13 @@ class _RenameAt2(Protocol): ) -> int: ... -def rename_exchange_at(directory_fd: int, first: str, second: str) -> None: - """Atomically exchange two names inside one already bound directory.""" - +def _rename_at2( + old_directory_fd: int, + old_name: str, + new_directory_fd: int, + new_name: str, + flags: int, +) -> int: library = ctypes.CDLL(None, use_errno=True) try: rename_at2 = cast(_RenameAt2, library.renameat2) @@ -53,20 +58,40 @@ def rename_exchange_at(directory_fd: int, first: str, second: str) -> None: rename_at2.restype = ctypes.c_int ctypes.set_errno(0) result = rename_at2( - directory_fd, - os.fsencode(first), - directory_fd, - os.fsencode(second), - RENAME_EXCHANGE, + old_directory_fd, + os.fsencode(old_name), + new_directory_fd, + os.fsencode(new_name), + flags, ) if result == 0: - return + return 0 error_number = ctypes.get_errno() if error_number in {errno.ENOSYS, errno.EINVAL, errno.EOPNOTSUPP}: raise DocForgeError( "atomic_exchange_unavailable", "Atomic exchange is unavailable on this filesystem", ) + return error_number + + +def rename_exchange_between_at( + first_directory_fd: int, + first: str, + second_directory_fd: int, + second: str, +) -> None: + """Atomically exchange names between two bound directories on one filesystem.""" + + error_number = _rename_at2( + first_directory_fd, + first, + second_directory_fd, + second, + RENAME_EXCHANGE, + ) + if error_number == 0: + return raise DocForgeError( "publication_failure", "Could not exchange atomic publication paths", @@ -74,6 +99,38 @@ def rename_exchange_at(directory_fd: int, first: str, second: str) -> None: ) from OSError(error_number, os.strerror(error_number)) +def rename_exchange_at(directory_fd: int, first: str, second: str) -> None: + """Atomically exchange two names inside one already bound directory.""" + + rename_exchange_between_at(directory_fd, first, directory_fd, second) + + +def rename_noreplace_between_at( + source_directory_fd: int, + source: str, + target_directory_fd: int, + target: str, +) -> bool: + """Atomically move one name without replacing a target that appeared.""" + + error_number = _rename_at2( + source_directory_fd, + source, + target_directory_fd, + target, + RENAME_NOREPLACE, + ) + if error_number == 0: + return True + if error_number == errno.EEXIST: + return False + raise DocForgeError( + "publication_failure", + "Could not move an atomic publication path without replacement", + error_number=error_number, + ) from OSError(error_number, os.strerror(error_number)) + + def open_bound_directory(path: Path) -> int: """Open one real directory and bind its current inode for later operations.""" diff --git a/src/docforge/application.py b/src/docforge/application.py index b9292da..d6130cc 100644 --- a/src/docforge/application.py +++ b/src/docforge/application.py @@ -4,16 +4,22 @@ from __future__ import annotations import json import os +import secrets import stat import tempfile from collections import defaultdict from collections.abc import Mapping, Sequence from contextlib import suppress -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Literal, Protocol, cast -from ._fs_safety import open_bound_directory, rename_exchange_at, require_bound_directory +from ._fs_safety import ( + open_bound_directory, + rename_exchange_between_at, + rename_noreplace_between_at, + require_bound_directory, +) from .changesets import ChangesetStore from .errors import DocForgeError from .index import ProjectIndex @@ -83,6 +89,9 @@ class _CanonicalPublication: backup_snapshot: _CanonicalFile | None = None published_snapshot: _CanonicalFile | None = None committed: bool = False + cleanup_conflicts: list[dict[str, object]] = field( + default_factory=lambda: list[dict[str, object]]() + ) class GenericCanonicalApplier: @@ -105,8 +114,10 @@ class GenericCanonicalApplier: projected_by_source = self._nodes_by_source(projected) publications: list[_CanonicalPublication] = [] created_directories: list[Path] = [] + transaction_root: Path | None = None targets = {relative: self._target(relative) for relative in sorted(changed_sources)} try: + transaction_root = self._prepare_transaction_root() for relative, target in targets.items(): self._prepare_parent(target.parent, created_directories) expected = self._capture(target, base.descriptor.limits.max_source_bytes) @@ -126,9 +137,12 @@ class GenericCanonicalApplier: source=relative, ) staged, staged_snapshot = self._stage( - target, + transaction_root, raw, mode=(stat.S_IMODE(expected.mode) if expected is not None else 0o600), + owner=(expected.owner if expected is not None else None), + group=(expected.group if expected is not None else None), + source=relative, ) publications.append( _CanonicalPublication( @@ -169,7 +183,9 @@ class GenericCanonicalApplier: ) except Exception as error: recovery = self._rollback(publications) - self._discard_unowned_staging(publications) + recovery.extend(self._discard_unowned_staging(publications)) + if transaction_root is not None: + recovery.extend(self._finish_transaction(transaction_root)) self._remove_empty_directories(created_directories) if recovery: cause = error.code if isinstance(error, DocForgeError) else type(error).__name__ @@ -192,18 +208,37 @@ class GenericCanonicalApplier: "Canonical target raced publication but was restored without data loss", ) from error raise - retained = self._discard_backups(publications) + retained = [ + conflict for publication in publications for conflict in publication.cleanup_conflicts + ] + retained.extend(self._discard_backups(publications)) + assert transaction_root is not None + retained.extend(self._finish_transaction(transaction_root)) + recovery_status = "cleanup_required" if retained else "clean" return { "applied_sources": sorted(changed_sources), "removed_sources": sorted( source for source in changed_sources if source not in projected_by_source ), "retained_recovery_files": retained, + "application_recovery": { + "status": recovery_status, + "retained": retained, + "remediation": ( + ( + "Canonical content is committed. Preserve and inspect the retained " + "transaction files, then remove only confirmed DocForge-owned artifacts." + ) + if retained + else None + ), + }, } def _publish(self, publication: _CanonicalPublication) -> None: target = publication.target parent_fd = open_bound_directory(target.parent) + staging_fd = open_bound_directory(publication.staged.parent) try: current = self._capture_at( target.parent, @@ -225,8 +260,8 @@ class GenericCanonicalApplier: source=publication.relative, ) staged = self._capture_at( - target.parent, - parent_fd, + publication.staged.parent, + staging_fd, publication.staged.name, self.project.descriptor.limits.max_source_bytes, ) @@ -237,22 +272,24 @@ class GenericCanonicalApplier: source=publication.relative, ) if publication.action == "create": - self._publish_create(publication, parent_fd) + self._publish_create(publication, parent_fd, staging_fd) else: - self._publish_exchange(publication, parent_fd) + self._publish_exchange(publication, parent_fd, staging_fd) finally: + os.close(staging_fd) os.close(parent_fd) def _publish_create( self, publication: _CanonicalPublication, parent_fd: int, + staging_fd: int, ) -> None: try: os.link( publication.staged.name, publication.target.name, - src_dir_fd=parent_fd, + src_dir_fd=staging_fd, dst_dir_fd=parent_fd, follow_symlinks=False, ) @@ -281,8 +318,40 @@ class GenericCanonicalApplier: "Published canonical create target does not match its staging file", source=publication.relative, ) + assert published is not None publication.published_snapshot = published - os.unlink(publication.staged.name, dir_fd=parent_fd) + private_copy = self._capture_at( + publication.staged.parent, + staging_fd, + publication.staged.name, + self.project.descriptor.limits.max_source_bytes, + ) + if private_copy is None or not publication.staged_snapshot.renamed_to(private_copy): + publication.cleanup_conflicts.append( + self._recovery_conflict(publication, "private_create_link_changed") + ) + else: + conflict = self._remove_private( + publication, + publication.staged, + private_copy, + "private_create_link_cleanup_failed", + ) + if conflict is not None: + publication.cleanup_conflicts.append(conflict) + refreshed = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + if not published.renamed_to(refreshed): + raise DocForgeError( + "application_mismatch", + "Published canonical create target changed during private cleanup", + source=publication.relative, + ) + publication.published_snapshot = refreshed os.fsync(parent_fd) require_bound_directory(publication.target.parent, parent_fd) @@ -290,12 +359,18 @@ class GenericCanonicalApplier: self, publication: _CanonicalPublication, parent_fd: int, + staging_fd: int, ) -> None: - rename_exchange_at(parent_fd, publication.staged.name, publication.target.name) + rename_exchange_between_at( + staging_fd, + publication.staged.name, + parent_fd, + publication.target.name, + ) publication.committed = True displaced = self._capture_at( - publication.target.parent, - parent_fd, + publication.staged.parent, + staging_fd, publication.staged.name, self.project.descriptor.limits.max_source_bytes, ) @@ -312,7 +387,7 @@ class GenericCanonicalApplier: or not publication.expected.renamed_to(displaced) or not publication.staged_snapshot.renamed_to(published) ): - if not self._exchange_back(publication, parent_fd): + if not self._exchange_back(publication, parent_fd, staging_fd): raise DocForgeError( "application_recovery_required", "Canonical target raced atomic publication and displaced data was retained", @@ -325,22 +400,98 @@ class GenericCanonicalApplier: source=publication.relative, ) if publication.action == "delete": - os.unlink(publication.target.name, dir_fd=parent_fd) - publication.published_snapshot = None + assert published is not None + conflict = self._detach_canonical( + publication, + parent_fd, + staging_fd, + published, + ) + if conflict is not None: + if publication.published_snapshot is None: + publication.cleanup_conflicts.append(conflict) + else: + raise DocForgeError( + "application_recovery_required", + "Canonical delete target raced final detachment and was retained", + source=publication.relative, + conflict=conflict, + ) os.fsync(parent_fd) + os.fsync(staging_fd) require_bound_directory(publication.target.parent, parent_fd) + require_bound_directory(publication.staged.parent, staging_fd) + + def _detach_canonical( + self, + publication: _CanonicalPublication, + parent_fd: int, + staging_fd: int, + expected: _CanonicalFile, + ) -> dict[str, object] | None: + detached_name: str | None = None + for _ in range(8): + candidate = f".detached-{secrets.token_hex(12)}" + if rename_noreplace_between_at( + parent_fd, + publication.target.name, + staging_fd, + candidate, + ): + detached_name = candidate + break + if detached_name is None: + return self._recovery_conflict(publication, "private_name_collisions") + detached = publication.staged.parent / detached_name + moved: _CanonicalFile | None = None + if self._same_inode_at(staging_fd, detached_name, expected): + with suppress(DocForgeError, OSError): + moved = self._capture_at( + detached.parent, + staging_fd, + detached.name, + self.project.descriptor.limits.max_source_bytes, + ) + if moved is None or not expected.renamed_to(moved): + restored = rename_noreplace_between_at( + staging_fd, + detached_name, + parent_fd, + publication.target.name, + ) + os.fsync(staging_fd) + os.fsync(parent_fd) + conflict = self._recovery_conflict(publication, "canonical_detach_raced") + conflict["raced_data"] = None if restored else self._relative(detached) + conflict["foreign_target_restored"] = restored + return conflict + publication.published_snapshot = None + conflict = self._remove_private( + publication, + detached, + moved, + "detached_cleanup_failed", + ) + os.fsync(parent_fd) + return conflict def _exchange_back( self, publication: _CanonicalPublication, parent_fd: int, + staging_fd: int, ) -> bool: backup = publication.backup_snapshot published = publication.published_snapshot if backup is None or published is None: return False try: - rename_exchange_at(parent_fd, publication.staged.name, publication.target.name) + rename_exchange_between_at( + staging_fd, + publication.staged.name, + parent_fd, + publication.target.name, + ) restored = self._capture_at( publication.target.parent, parent_fd, @@ -348,21 +499,36 @@ class GenericCanonicalApplier: self.project.descriptor.limits.max_source_bytes, ) staged = self._capture_at( - publication.target.parent, - parent_fd, + publication.staged.parent, + staging_fd, publication.staged.name, self.project.descriptor.limits.max_source_bytes, ) if not backup.renamed_to(restored) or not published.renamed_to(staged): return False + assert staged is not None + publication.staged_snapshot = staged publication.backup_snapshot = None publication.published_snapshot = None publication.committed = False os.fsync(parent_fd) + os.fsync(staging_fd) return True except (DocForgeError, OSError): return False + @staticmethod + def _same_inode_at( + directory_fd: int, + name: str, + expected: _CanonicalFile, + ) -> bool: + try: + current = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) + except OSError: + return False + return current.st_dev == expected.device and current.st_ino == expected.inode + def _rollback( self, publications: Sequence[_CanonicalPublication], @@ -392,6 +558,7 @@ class GenericCanonicalApplier: publication: _CanonicalPublication, ) -> dict[str, object] | None: parent_fd = open_bound_directory(publication.target.parent) + staging_fd = open_bound_directory(publication.staged.parent) try: current = self._capture_at( publication.target.parent, @@ -400,8 +567,8 @@ class GenericCanonicalApplier: self.project.descriptor.limits.max_source_bytes, ) backup = self._capture_at( - publication.target.parent, - parent_fd, + publication.staged.parent, + staging_fd, publication.staged.name, self.project.descriptor.limits.max_source_bytes, ) @@ -412,12 +579,16 @@ class GenericCanonicalApplier: or not publication.backup_snapshot.unchanged(backup) ): return self._recovery_conflict(publication, "target_or_backup_changed") - if not self._exchange_back(publication, parent_fd): + if not self._exchange_back(publication, parent_fd, staging_fd): return self._recovery_conflict(publication, "atomic_restore_unconfirmed") - os.unlink(publication.staged.name, dir_fd=parent_fd) - os.fsync(parent_fd) - return None + return self._remove_private( + publication, + publication.staged, + publication.staged_snapshot, + "rollback_staging_cleanup_failed", + ) finally: + os.close(staging_fd) os.close(parent_fd) def _rollback_delete( @@ -425,6 +596,7 @@ class GenericCanonicalApplier: publication: _CanonicalPublication, ) -> dict[str, object] | None: parent_fd = open_bound_directory(publication.target.parent) + staging_fd = open_bound_directory(publication.staged.parent) try: current = self._capture_at( publication.target.parent, @@ -433,8 +605,8 @@ class GenericCanonicalApplier: self.project.descriptor.limits.max_source_bytes, ) backup = self._capture_at( - publication.target.parent, - parent_fd, + publication.staged.parent, + staging_fd, publication.staged.name, self.project.descriptor.limits.max_source_bytes, ) @@ -444,15 +616,12 @@ class GenericCanonicalApplier: backup ): return self._recovery_conflict(publication, "backup_changed") - try: - os.link( - publication.staged.name, - publication.target.name, - src_dir_fd=parent_fd, - dst_dir_fd=parent_fd, - follow_symlinks=False, - ) - except FileExistsError: + if not rename_noreplace_between_at( + staging_fd, + publication.staged.name, + parent_fd, + publication.target.name, + ): return self._recovery_conflict(publication, "deleted_target_reappeared") restored = self._capture_at( publication.target.parent, @@ -462,12 +631,13 @@ class GenericCanonicalApplier: ) if not publication.backup_snapshot.renamed_to(restored): return self._recovery_conflict(publication, "restore_unconfirmed") - os.unlink(publication.staged.name, dir_fd=parent_fd) publication.backup_snapshot = None publication.committed = False os.fsync(parent_fd) + os.fsync(staging_fd) return None finally: + os.close(staging_fd) os.close(parent_fd) def _rollback_create( @@ -475,7 +645,7 @@ class GenericCanonicalApplier: publication: _CanonicalPublication, ) -> dict[str, object] | None: parent_fd = open_bound_directory(publication.target.parent) - tombstone: Path | None = None + staging_fd = open_bound_directory(publication.staged.parent) try: current = self._capture_at( publication.target.parent, @@ -491,36 +661,19 @@ class GenericCanonicalApplier: or not publication.published_snapshot.unchanged(current) ): return self._recovery_conflict(publication, "created_target_changed") - tombstone, tombstone_snapshot = self._stage(publication.target, b"", mode=0o600) - rename_exchange_at(parent_fd, tombstone.name, publication.target.name) - displaced = self._capture_at( - publication.target.parent, + conflict = self._detach_canonical( + publication, parent_fd, - tombstone.name, - self.project.descriptor.limits.max_source_bytes, + staging_fd, + current, ) - published_tombstone = self._capture_at( - publication.target.parent, - parent_fd, - publication.target.name, - self.project.descriptor.limits.max_source_bytes, - ) - if not publication.published_snapshot.renamed_to( - displaced - ) or not tombstone_snapshot.renamed_to(published_tombstone): - with suppress(DocForgeError): - rename_exchange_at(parent_fd, tombstone.name, publication.target.name) - return self._recovery_conflict(publication, "create_rollback_raced") - os.unlink(publication.target.name, dir_fd=parent_fd) - os.unlink(tombstone.name, dir_fd=parent_fd) - tombstone = None + if conflict is not None: + return conflict publication.committed = False publication.published_snapshot = None - os.fsync(parent_fd) return None finally: - if tombstone is not None: - tombstone.unlink(missing_ok=True) + os.close(staging_fd) os.close(parent_fd) def _discard_backups( @@ -529,55 +682,123 @@ class GenericCanonicalApplier: ) -> list[dict[str, object]]: retained: list[dict[str, object]] = [] for publication in publications: - if publication.action != "create" and publication.staged.exists(): - parent_fd = open_bound_directory(publication.target.parent) + if publication.action != "create" and publication.backup_snapshot is not None: try: - backup = self._capture_at( - publication.target.parent, - parent_fd, - publication.staged.name, - self.project.descriptor.limits.max_source_bytes, - ) - if ( - publication.backup_snapshot is None - or not publication.backup_snapshot.unchanged(backup) - ): - retained.append( - self._recovery_conflict(publication, "backup_cleanup_raced") + staging_fd = open_bound_directory(publication.staged.parent) + try: + backup = self._capture_at( + publication.staged.parent, + staging_fd, + publication.staged.name, + self.project.descriptor.limits.max_source_bytes, ) - continue - os.unlink(publication.staged.name, dir_fd=parent_fd) - os.fsync(parent_fd) - finally: - os.close(parent_fd) + finally: + os.close(staging_fd) + except Exception as error: + conflict = self._recovery_conflict( + publication, + "backup_cleanup_inspection_failed", + ) + conflict["error"] = ( + error.code if isinstance(error, DocForgeError) else type(error).__name__ + ) + retained.append(conflict) + continue + if not publication.backup_snapshot.unchanged(backup): + retained.append(self._recovery_conflict(publication, "backup_cleanup_raced")) + continue + conflict = self._remove_private( + publication, + publication.staged, + publication.backup_snapshot, + "backup_cleanup_failed", + ) + if conflict is not None: + retained.append(conflict) + continue publication.committed = False publication.backup_snapshot = None return retained - @staticmethod - def _discard_unowned_staging(publications: Sequence[_CanonicalPublication]) -> None: + def _discard_unowned_staging( + self, + publications: Sequence[_CanonicalPublication], + ) -> list[dict[str, object]]: + retained: list[dict[str, object]] = [] for publication in publications: - if not publication.committed and publication.staged.exists(): - publication.staged.unlink() + if publication.committed: + continue + conflict = self._remove_private( + publication, + publication.staged, + publication.staged_snapshot, + "staging_cleanup_failed", + ) + if conflict is not None: + retained.append(conflict) + return retained - def _stage(self, target: Path, content: bytes, *, mode: int) -> tuple[Path, _CanonicalFile]: + def _remove_private( + self, + publication: _CanonicalPublication, + path: Path, + expected: _CanonicalFile, + reason: str, + ) -> dict[str, object] | None: + try: + directory_fd = open_bound_directory(path.parent) + try: + backup = self._capture_at( + path.parent, + directory_fd, + path.name, + self.project.descriptor.limits.max_source_bytes, + ) + if backup is None: + return None + if not expected.unchanged(backup): + return self._private_recovery_conflict(publication, path, reason) + os.unlink(path.name, dir_fd=directory_fd) + os.fsync(directory_fd) + return None + finally: + os.close(directory_fd) + except Exception as error: + conflict = self._private_recovery_conflict(publication, path, reason) + conflict["error"] = ( + error.code if isinstance(error, DocForgeError) else type(error).__name__ + ) + return conflict + + def _stage( + self, + staging_root: Path, + content: bytes, + *, + mode: int, + owner: int | None, + group: int | None, + source: str, + ) -> tuple[Path, _CanonicalFile]: descriptor, temporary_name = tempfile.mkstemp( - prefix=".docforge-apply-", - dir=target.parent, + prefix="staged-", + dir=staging_root, ) temporary = Path(temporary_name) try: - os.fchmod(descriptor, mode) with os.fdopen(descriptor, "wb") as handle: handle.write(content) handle.flush() + if owner is not None and group is not None: + os.fchown(handle.fileno(), owner, group) + os.fchmod(handle.fileno(), mode) os.fsync(handle.fileno()) snapshot = self._capture(temporary, max(1, len(content))) if snapshot is None: raise DocForgeError( "application_mismatch", "Canonical staging file disappeared", - source=self._relative(target), + source=source, ) return temporary, snapshot except Exception: @@ -680,6 +901,19 @@ class GenericCanonicalApplier: ), } + def _private_recovery_conflict( + self, + publication: _CanonicalPublication, + path: Path, + reason: str, + ) -> dict[str, object]: + return { + "source": publication.relative, + "reason": reason, + "target": self._relative(publication.target), + "retained": self._relative(path) if path.exists() else None, + } + def _relative(self, path: Path) -> str: return path.relative_to(self.project.descriptor.root).as_posix() @@ -707,33 +941,106 @@ class GenericCanonicalApplier: def _prepare_parent(self, parent: Path, created: list[Path]) -> None: root = self.project.descriptor.root + created.extend(self._create_directories_durable(parent, root=root, mode=0o755)) + + def _prepare_transaction_root(self) -> Path: + root = self.project.descriptor.root + namespace = root / ".docforge/application" + self._create_directories_durable(namespace, root=root, mode=0o700) + transaction = Path(tempfile.mkdtemp(prefix="transaction-", dir=namespace)) + transaction.chmod(0o700) + self._fsync_directory(transaction) + self._fsync_directory(namespace) + return transaction + + def _create_directories_durable( + self, + path: Path, + *, + root: Path, + mode: int, + ) -> list[Path]: missing: list[Path] = [] - cursor = parent + cursor = path while not cursor.exists(): missing.append(cursor) cursor = cursor.parent if cursor.is_symlink() or cursor.resolve(strict=True) != cursor or not cursor.is_dir(): raise DocForgeError("path_escape", "Canonical target parent is unsafe") - if not cursor.is_relative_to(root): + if not cursor.is_relative_to(root) or not path.is_relative_to(root): raise DocForgeError("path_escape", "Canonical target parent escaped the project root") - parent.mkdir(parents=True, exist_ok=True) - if parent.resolve(strict=True) != parent: + created: list[Path] = [] + for directory in reversed(missing): + try: + directory.mkdir(mode=mode) + except FileExistsError as error: + if ( + directory.is_symlink() + or not directory.is_dir() + or directory.resolve(strict=True) != directory + ): + raise DocForgeError( + "path_escape", + "Canonical target parent became unsafe during creation", + ) from error + continue + self._fsync_directory(directory) + self._fsync_directory(directory.parent) + created.append(directory) + if path.resolve(strict=True) != path: raise DocForgeError("path_escape", "Canonical target parent resolves unexpectedly") - created.extend(reversed(missing)) + return created + + def _finish_transaction(self, transaction: Path) -> list[dict[str, object]]: + try: + retained = sorted( + (self._relative(path) for path in transaction.iterdir()), + ) + except Exception as error: + return [ + { + "reason": "transaction_inspection_failed", + "retained": self._relative(transaction), + "error": ( + error.code if isinstance(error, DocForgeError) else type(error).__name__ + ), + } + ] + if retained: + return [ + { + "reason": "transaction_files_retained", + "retained": retained, + } + ] + try: + transaction.rmdir() + self._fsync_directory(transaction.parent) + except OSError as error: + return [ + { + "reason": "transaction_cleanup_failed", + "retained": self._relative(transaction), + "error": type(error).__name__, + } + ] + return [] @staticmethod def _fsync_directory(path: Path) -> None: - descriptor = os.open(path, os.O_RDONLY) + descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: os.fsync(descriptor) finally: os.close(descriptor) - @staticmethod - def _remove_empty_directories(paths: Sequence[Path]) -> None: + def _remove_empty_directories(self, paths: Sequence[Path]) -> None: for path in reversed(paths): - with suppress(OSError): + try: path.rmdir() + self._fsync_directory(path.parent) + except OSError: + pass @staticmethod def _nodes_by_source(snapshot: ProjectSnapshot) -> dict[str, tuple[Node, ...]]: diff --git a/src/docforge/changesets.py b/src/docforge/changesets.py index d4c878e..9885c44 100644 --- a/src/docforge/changesets.py +++ b/src/docforge/changesets.py @@ -649,14 +649,28 @@ class ChangesetStore: tuple(cast(Mapping[str, object], item) for item in document["operations"]), ) current = self.project.load() + lifecycle_payload: dict[str, object] = { + "status": "applied", + "changeset_hash": actual_hash, + "revision": current.revision, + "source_hash": current.source_hash, + } + application_recovery = payload.get("application_recovery") + if isinstance(application_recovery, Mapping): + recovery_payload = cast(Mapping[str, object], application_recovery) + if recovery_payload.get("status") != "clean": + retained = recovery_payload.get("retained") + lifecycle_payload["application_recovery"] = { + "status": recovery_payload.get("status"), + "retained_count": ( + len(cast(list[object], retained)) if isinstance(retained, list) else 0 + ), + "retained_root": ".docforge/application", + "remediation": recovery_payload.get("remediation"), + } lifecycle = self._write_state( changeset_id, - { - "status": "applied", - "changeset_hash": actual_hash, - "revision": current.revision, - "source_hash": current.source_hash, - }, + lifecycle_payload, ) return self._result( current, diff --git a/tests/test_changesets.py b/tests/test_changesets.py index bfd878f..c956990 100644 --- a/tests/test_changesets.py +++ b/tests/test_changesets.py @@ -293,19 +293,24 @@ class DocForgeChangesetTests(unittest.TestCase): proposal_path = root / ".docforge/changesets/update-race.json" proposal_bytes = proposal_path.read_bytes() target = root / "docs/content/workflow.md" - exchange = application_module.rename_exchange_at + exchange = application_module.rename_exchange_between_at raced = False - def race(directory_fd: int, first: str, second: str) -> None: + def race( + first_directory_fd: int, + first: str, + second_directory_fd: int, + second: str, + ) -> None: nonlocal raced if second == target.name and not raced: raced = True target.write_bytes(target.read_bytes() + b"\nExternal edit at exchange.\n") - exchange(directory_fd, first, second) + exchange(first_directory_fd, first, second_directory_fd, second) with ( mock.patch( - "docforge.application.rename_exchange_at", + "docforge.application.rename_exchange_between_at", side_effect=race, ), self.assertRaises(DocForgeError) as captured, @@ -403,21 +408,26 @@ class DocForgeChangesetTests(unittest.TestCase): ], ) delete_target = delete_root / "docs/content/proof.toml" - exchange = application_module.rename_exchange_at + exchange = application_module.rename_exchange_between_at deleted_race = False - def race_delete(directory_fd: int, first: str, second: str) -> None: + def race_delete( + first_directory_fd: int, + first: str, + second_directory_fd: int, + second: str, + ) -> None: nonlocal deleted_race if second == delete_target.name and not deleted_race: deleted_race = True delete_target.write_bytes( delete_target.read_bytes() + b"\n# foreign delete edit\n" ) - exchange(directory_fd, first, second) + exchange(first_directory_fd, first, second_directory_fd, second) with ( mock.patch( - "docforge.application.rename_exchange_at", + "docforge.application.rename_exchange_between_at", side_effect=race_delete, ), self.assertRaises(DocForgeError) as delete_error, @@ -432,6 +442,167 @@ class DocForgeChangesetTests(unittest.TestCase): self.assertIn("# foreign delete edit", delete_target.read_text(encoding="utf-8")) self.assertFalse(tuple(delete_target.parent.glob(".docforge-apply-*"))) + def test_delete_detach_race_restores_foreign_target_and_retains_original(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + project = Project.open(root) + store = ChangesetStore(project, "alpha-editor") + proposal = store.register( + "delete-detach-race", + [ + { + "operation": "delete", + "node_id": "proof.validation", + "relationship_changes": [ + { + "action": "remove", + "source_id": "proof.validation", + "relation": "proves", + "target_id": "guide.workflow", + } + ], + "rationale": "Race the final no-replace canonical detachment.", + } + ], + ) + target = root / "docs/content/proof.toml" + original = target.read_bytes() + move = application_module.rename_noreplace_between_at + raced = False + + def race_detach( + source_directory_fd: int, + source: str, + target_directory_fd: int, + destination: str, + ) -> bool: + nonlocal raced + if source == target.name and destination.startswith(".detached-") and not raced: + raced = True + replacement = target.with_name(".foreign-delete") + replacement.write_bytes(b"foreign replacement at delete detach\n") + os.replace(replacement, target) + return move( + source_directory_fd, + source, + target_directory_fd, + destination, + ) + + with ( + mock.patch( + "docforge.application.rename_noreplace_between_at", + side_effect=race_detach, + ), + self.assertRaises(DocForgeError) as captured, + ): + store.apply( + changeset_id="delete-detach-race", + expected_changeset_hash=str(proposal["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(project).apply, + ) + + self.assertTrue(raced) + self.assertEqual("application_recovery_required", captured.exception.code) + self.assertEqual(b"foreign replacement at delete detach\n", target.read_bytes()) + conflicts = captured.exception.details["conflicts"] + retained = root / conflicts[0]["retained"] + self.assertEqual(original, retained.read_bytes()) + self.assertFalse( + (root / ".docforge/changesets/.state/delete-detach-race.json").exists() + ) + + def test_create_rollback_detach_race_never_unlinks_foreign_target(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + project = Project.open(root) + store = ChangesetStore(project, "alpha-editor") + proposal = store.register( + "create-rollback-detach-race", + [ + { + "operation": "create", + "node_id": "guide.created", + "target_source": "docs/content/a-created.md", + "metadata": self.new_metadata(), + "content": "Approved content that publishes first.", + "rationale": "Exercise create rollback detachment.", + }, + { + "operation": "update", + "node_id": "guide.workflow", + "metadata": {"summary": "Synthetic failing second publication."}, + "rationale": "Trigger rollback after create publication.", + }, + ], + ) + target = root / "docs/content/a-created.md" + publish = GenericCanonicalApplier._publish + move = application_module.rename_noreplace_between_at + publish_calls = 0 + raced = False + + def fail_second( + applier: GenericCanonicalApplier, + publication: Any, + ) -> None: + nonlocal publish_calls + publish_calls += 1 + if publish_calls == 1: + publish(applier, publication) + return + raise DocForgeError("application_failure", "Synthetic second publication failure") + + def race_rollback_detach( + source_directory_fd: int, + source: str, + target_directory_fd: int, + destination: str, + ) -> bool: + nonlocal raced + if source == target.name and destination.startswith(".detached-") and not raced: + raced = True + replacement = target.with_name(".foreign-create-rollback") + replacement.write_bytes(b"foreign replacement during create rollback\n") + os.replace(replacement, target) + return move( + source_directory_fd, + source, + target_directory_fd, + destination, + ) + + with ( + mock.patch.object( + GenericCanonicalApplier, + "_publish", + autospec=True, + side_effect=fail_second, + ), + mock.patch( + "docforge.application.rename_noreplace_between_at", + side_effect=race_rollback_detach, + ), + self.assertRaises(DocForgeError) as captured, + ): + store.apply( + changeset_id="create-rollback-detach-race", + expected_changeset_hash=str(proposal["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(project).apply, + ) + + self.assertTrue(raced) + self.assertEqual("application_recovery_required", captured.exception.code) + self.assertEqual( + b"foreign replacement during create rollback\n", + target.read_bytes(), + ) + self.assertFalse( + (root / ".docforge/changesets/.state/create-rollback-detach-race.json").exists() + ) + def test_rollback_never_clobbers_a_foreign_edit_and_retains_original_bytes(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) @@ -505,7 +676,8 @@ class DocForgeChangesetTests(unittest.TestCase): with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) target = root / "docs/content/workflow.md" - target.chmod(0o640) + target.chmod(0o6750) + before = target.stat() project = Project.open(root) store = ChangesetStore(project, "alpha-editor") proposal = store.register( @@ -528,8 +700,138 @@ class DocForgeChangesetTests(unittest.TestCase): ) self.assertTrue(result["applied"]) - self.assertEqual(0o640, stat.S_IMODE(target.stat().st_mode)) + after = target.stat() + self.assertEqual(0o6750, stat.S_IMODE(after.st_mode)) + self.assertEqual(before.st_uid, after.st_uid) + self.assertEqual(before.st_gid, after.st_gid) self.assertEqual([], result["retained_recovery_files"]) + self.assertEqual("clean", result["application_recovery"]["status"]) + self.assertFalse(tuple((root / ".docforge/application").glob("transaction-*"))) + + def test_nested_creation_fsyncs_each_new_directory_and_parent_entry(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + project = Project.open(root) + store = ChangesetStore(project, "alpha-editor") + proposal = store.register( + "nested-durable", + [ + { + "operation": "create", + "node_id": "guide.nested", + "target_source": "docs/content/nested/deeper/guide.md", + "metadata": self.new_metadata(), + "content": "Nested canonical content.", + "rationale": "Prove durable nested-directory creation.", + } + ], + ) + real_fsync = os.fsync + fsynced_directories: set[Path] = set() + + def record_fsync(descriptor: int) -> None: + try: + path = Path(os.readlink(f"/proc/self/fd/{descriptor}")) + if path.is_dir(): + fsynced_directories.add(path) + except OSError: + pass + real_fsync(descriptor) + + with mock.patch( + "docforge.application.os.fsync", + side_effect=record_fsync, + ): + result = store.apply( + changeset_id="nested-durable", + expected_changeset_hash=str(proposal["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(project).apply, + ) + + self.assertTrue(result["applied"]) + for path in ( + root / "docs/content", + root / "docs/content/nested", + root / "docs/content/nested/deeper", + ): + self.assertIn(path, fsynced_directories) + + def test_post_commit_cleanup_failure_closes_proposal_with_recovery_record( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + project = Project.open(root) + store = ChangesetStore(project, "alpha-editor") + proposal = store.register( + "cleanup-recovery", + [ + { + "operation": "update", + "node_id": "guide.workflow", + "metadata": {"summary": "Committed despite private cleanup failure."}, + "rationale": "Persist actionable post-commit recovery evidence.", + } + ], + ) + real_unlink = os.unlink + failed = False + + def fail_private_cleanup( + path: str | bytes, + *, + dir_fd: int | None = None, + ) -> None: + nonlocal failed + if ( + isinstance(path, str) + and path.startswith("staged-") + and dir_fd is not None + and not failed + ): + failed = True + raise PermissionError("synthetic private cleanup failure") + real_unlink(path, dir_fd=dir_fd) + + with mock.patch( + "docforge.application.os.unlink", + side_effect=fail_private_cleanup, + ): + result = store.apply( + changeset_id="cleanup-recovery", + expected_changeset_hash=str(proposal["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(project).apply, + ) + + self.assertTrue(failed) + self.assertTrue(result["applied"]) + self.assertEqual("applied", result["lifecycle"]["status"]) + self.assertEqual( + "cleanup_required", + result["application_recovery"]["status"], + ) + self.assertEqual( + "cleanup_required", + result["lifecycle"]["application_recovery"]["status"], + ) + retained = result["application_recovery"]["retained"] + self.assertTrue(retained) + lifecycle_path = root / ".docforge/changesets/.state/cleanup-recovery.json" + lifecycle = json.loads(lifecycle_path.read_text(encoding="utf-8")) + self.assertEqual( + "cleanup_required", + lifecycle["application_recovery"]["status"], + ) + with self.assertRaises(DocForgeError) as closed: + store.apply( + changeset_id="cleanup-recovery", + expected_changeset_hash=str(proposal["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(project).apply, + ) + self.assertEqual("changeset_closed", closed.exception.code) def test_changeset_rollback_fsyncs_the_parent_directory(self) -> None: with tempfile.TemporaryDirectory() as directory: