diff --git a/backend/packages/harness/deerflow/uploads/__init__.py b/backend/packages/harness/deerflow/uploads/__init__.py index e335c2fb8..028be2c46 100644 --- a/backend/packages/harness/deerflow/uploads/__init__.py +++ b/backend/packages/harness/deerflow/uploads/__init__.py @@ -1,20 +1,23 @@ from .conversion import convert_uploaded_file_to_markdown +from .errors import AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError from .layout import ( UPLOAD_CONVERSIONS_DIRNAME, + UPLOAD_LOCKS_DIRNAME, UnsafeConversionPathError, artifact_url_for_virtual_path, conversion_dir_for_uploads, conversion_path_for_upload, conversion_virtual_path, ensure_conversion_dir, + ensure_upload_lock_dir, existing_conversion_path_for_upload, validate_conversion_dir, ) +from .lease import UploadIdentity, UploadNameLease from .manager import ( UPLOAD_STAGING_PREFIX, UPLOAD_STAGING_SUFFIX, - AtomicUploadPublishError, - PathTraversalError, + PublishedUpload, StagedUpload, abort_staged_upload, claim_unique_filename, @@ -28,9 +31,13 @@ from .manager import ( list_files_in_dir, normalize_filename, publish_staged_upload, + publish_staged_upload_leased, publish_upload_bytes, + publish_upload_bytes_leased, publish_upload_copy, + publish_upload_copy_leased, replace_system_owned_staged_file, + rollback_published_upload, upload_artifact_url, upload_virtual_path, validate_path_traversal, @@ -40,6 +47,7 @@ from .manager import ( __all__ = [ "get_uploads_dir", "UPLOAD_CONVERSIONS_DIRNAME", + "UPLOAD_LOCKS_DIRNAME", "UnsafeConversionPathError", "conversion_dir_for_uploads", "conversion_path_for_upload", @@ -47,21 +55,30 @@ __all__ = [ "artifact_url_for_virtual_path", "validate_conversion_dir", "ensure_conversion_dir", + "ensure_upload_lock_dir", "existing_conversion_path_for_upload", "convert_uploaded_file_to_markdown", "ensure_uploads_dir", "normalize_filename", "PathTraversalError", + "UnsafeUploadPathError", "AtomicUploadPublishError", + "UploadIdentity", + "UploadNameLease", "StagedUpload", + "PublishedUpload", "UPLOAD_STAGING_PREFIX", "UPLOAD_STAGING_SUFFIX", "claim_unique_filename", "create_upload_staging_file", "abort_staged_upload", "publish_staged_upload", + "publish_staged_upload_leased", "publish_upload_bytes", + "publish_upload_bytes_leased", "publish_upload_copy", + "publish_upload_copy_leased", + "rollback_published_upload", "replace_system_owned_staged_file", "cleanup_stale_upload_staging_files", "is_upload_staging_file", diff --git a/backend/packages/harness/deerflow/uploads/errors.py b/backend/packages/harness/deerflow/uploads/errors.py new file mode 100644 index 000000000..153e9b479 --- /dev/null +++ b/backend/packages/harness/deerflow/uploads/errors.py @@ -0,0 +1,13 @@ +"""Shared upload validation and publication errors.""" + + +class PathTraversalError(ValueError): + """Raised when a path escapes its allowed base directory.""" + + +class UnsafeUploadPathError(ValueError): + """Raised when an upload destination is not a safe regular file path.""" + + +class AtomicUploadPublishError(UnsafeUploadPathError): + """Raised when storage cannot honor atomic no-replace publication.""" diff --git a/backend/packages/harness/deerflow/uploads/layout.py b/backend/packages/harness/deerflow/uploads/layout.py index de68f0466..788bf994f 100644 --- a/backend/packages/harness/deerflow/uploads/layout.py +++ b/backend/packages/harness/deerflow/uploads/layout.py @@ -8,6 +8,7 @@ from urllib.parse import quote from deerflow.config.paths import VIRTUAL_PATH_PREFIX UPLOAD_CONVERSIONS_DIRNAME = ".upload-conversions" +UPLOAD_LOCKS_DIRNAME = ".locks" class UnsafeConversionPathError(ValueError): @@ -49,6 +50,28 @@ def ensure_conversion_dir(uploads_dir: Path) -> Path: return validated +def upload_lock_dir_for_uploads(uploads_dir: Path) -> Path: + """Return the stable per-upload lock directory.""" + return conversion_dir_for_uploads(uploads_dir) / UPLOAD_LOCKS_DIRNAME + + +def ensure_upload_lock_dir(uploads_dir: Path) -> Path: + """Create and validate the system-owned upload lock directory.""" + conversion_dir = ensure_conversion_dir(uploads_dir) + lock_dir = conversion_dir / UPLOAD_LOCKS_DIRNAME + try: + lock_dir.mkdir(mode=0o700) + except FileExistsError: + pass + try: + lock_stat = os.lstat(lock_dir) + except FileNotFoundError as exc: + raise UnsafeConversionPathError("Upload lock directory disappeared") from exc + if stat.S_ISLNK(lock_stat.st_mode) or not stat.S_ISDIR(lock_stat.st_mode): + raise UnsafeConversionPathError("Unsafe upload lock directory") + return lock_dir + + def existing_conversion_path_for_upload(upload_path: Path) -> Path | None: """Return an existing safe generated file owned by ``upload_path``.""" if validate_conversion_dir(upload_path.parent) is None: diff --git a/backend/packages/harness/deerflow/uploads/lease.py b/backend/packages/harness/deerflow/uploads/lease.py new file mode 100644 index 000000000..109e44ae8 --- /dev/null +++ b/backend/packages/harness/deerflow/uploads/lease.py @@ -0,0 +1,159 @@ +"""Cross-process leases and inode identities for published uploads.""" + +import hashlib +import os +import stat +import threading +from dataclasses import dataclass, field +from pathlib import Path +from typing import BinaryIO + +from deerflow.uploads.errors import UnsafeUploadPathError +from deerflow.uploads.layout import ensure_upload_lock_dir + +try: + import fcntl +except ImportError: # pragma: no cover - Windows only + fcntl = None # type: ignore[assignment] + import msvcrt + + +_LOCK_STRIPES = tuple(threading.Lock() for _ in range(64)) + + +@dataclass(frozen=True, slots=True) +class UploadIdentity: + """Filesystem identity of one published upload generation.""" + + device: int + inode: int + + @classmethod + def from_path(cls, path: Path) -> "UploadIdentity": + upload_stat = os.lstat(path) + if not stat.S_ISREG(upload_stat.st_mode): + raise UnsafeUploadPathError("Published upload is not a regular file") + return cls(device=upload_stat.st_dev, inode=upload_stat.st_ino) + + def matches(self, path: Path) -> bool: + """Return whether *path* still names this generation.""" + try: + upload_stat = os.lstat(path) + except FileNotFoundError: + return False + return stat.S_ISREG(upload_stat.st_mode) and (upload_stat.st_dev, upload_stat.st_ino) == ( + self.device, + self.inode, + ) + + +def _open_lock_file(lock_path: Path) -> BinaryIO: + flags = os.O_RDWR | os.O_CREAT + if hasattr(os, "O_BINARY"): + flags |= os.O_BINARY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + fd = os.open(lock_path, flags, 0o600) + try: + descriptor_stat = os.fstat(fd) + path_stat = os.lstat(lock_path) + if not stat.S_ISREG(descriptor_stat.st_mode) or descriptor_stat.st_nlink != 1 or not stat.S_ISREG(path_stat.st_mode) or (descriptor_stat.st_dev, descriptor_stat.st_ino) != (path_stat.st_dev, path_stat.st_ino): + raise UnsafeUploadPathError("Unsafe upload lock file") + if descriptor_stat.st_size == 0: + os.write(fd, b"\0") + os.lseek(fd, 0, os.SEEK_SET) + return os.fdopen(fd, "r+b", buffering=0) + except BaseException: + os.close(fd) + raise + + +def _lock_file(lock_file: BinaryIO) -> None: + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + return + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1) + + +def _unlock_file(lock_file: BinaryIO) -> None: + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + return + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) + + +@dataclass(slots=True) +class UploadNameLease: + """Exclusive thread-and-process lease for one actual upload filename.""" + + uploads_dir: Path + filename: str + lock_path: Path + _lock_file: BinaryIO + _stripe: threading.Lock + _active: bool = True + _state_lock: threading.Lock = field(default_factory=threading.Lock) + + @classmethod + def acquire(cls, uploads_dir: Path, filename: str) -> "UploadNameLease": + """Acquire the stable name lease, blocking until it is available.""" + if not filename or Path(filename).name != filename or "\\" in filename: + raise UnsafeUploadPathError(f"Unsafe upload lease filename: {filename!r}") + if len(filename.encode("utf-8")) > 255: + raise UnsafeUploadPathError("Upload lease filename is too long") + + digest = hashlib.sha256(filename.encode("utf-8")).hexdigest() + stripe = _LOCK_STRIPES[int(digest[:2], 16) % len(_LOCK_STRIPES)] + stripe.acquire() + lock_file: BinaryIO | None = None + try: + lock_path = ensure_upload_lock_dir(Path(uploads_dir)) / f"{digest}.lock" + lock_file = _open_lock_file(lock_path) + _lock_file(lock_file) + return cls( + uploads_dir=Path(uploads_dir), + filename=filename, + lock_path=lock_path, + _lock_file=lock_file, + _stripe=stripe, + ) + except BaseException: + if lock_file is not None: + lock_file.close() + stripe.release() + raise + + @property + def is_active(self) -> bool: + """Return whether this object still owns the name lease.""" + with self._state_lock: + return self._active + + def release(self) -> None: + """Release the OS lock and process stripe; repeated calls are harmless.""" + with self._state_lock: + if not self._active: + return + error: BaseException | None = None + try: + _unlock_file(self._lock_file) + except BaseException as exc: # pragma: no cover - exceptional OS failure + error = exc + try: + self._lock_file.close() + except BaseException as exc: # pragma: no cover - exceptional OS failure + if error is None: + error = exc + finally: + self._active = False + self._stripe.release() + if error is not None: + raise error + + def __enter__(self) -> "UploadNameLease": + return self + + def __exit__(self, exc_type, exc_value, traceback) -> None: + self.release() diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index 3b4046d18..f5c05ab58 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -17,27 +17,16 @@ from typing import BinaryIO from deerflow.config.paths import get_paths from deerflow.runtime.user_context import get_effective_user_id +from deerflow.uploads.errors import AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError from deerflow.uploads.layout import ( UPLOAD_CONVERSIONS_DIRNAME, artifact_url_for_virtual_path, existing_conversion_path_for_upload, upload_virtual_path, ) +from deerflow.uploads.lease import UploadIdentity, UploadNameLease from deerflow.utils.thread_id import validate_thread_id - -class PathTraversalError(ValueError): - """Raised when a path escapes its allowed base directory.""" - - -class UnsafeUploadPathError(ValueError): - """Raised when an upload destination is not a safe regular file path.""" - - -class AtomicUploadPublishError(UnsafeUploadPathError): - """Raised when storage cannot honor atomic no-replace publication.""" - - logger = logging.getLogger(__name__) UPLOAD_STAGING_PREFIX = ".upload-" @@ -53,6 +42,23 @@ class StagedUpload: handle: BinaryIO +@dataclass(slots=True) +class PublishedUpload: + """A published upload whose actual filename remains exclusively leased.""" + + path: Path + identity: UploadIdentity + lease: UploadNameLease + + @property + def is_active(self) -> bool: + return self.lease.is_active + + def release(self) -> None: + """Release the publication's name lease.""" + self.lease.release() + + def get_uploads_dir(thread_id: str, *, user_id: str | None = None) -> Path: """Return the uploads directory path for a thread (no side effects).""" validate_thread_id(thread_id) @@ -91,6 +97,8 @@ def normalize_filename(filename: str) -> str: raise ValueError(f"Filename contains backslash: {filename!r}") if len(safe.encode("utf-8")) > 255: raise ValueError(f"Filename too long: {len(safe)} chars") + if is_upload_staging_file(safe): + raise ValueError(f"Filename uses reserved upload staging pattern: {filename!r}") return safe @@ -155,9 +163,20 @@ def create_upload_staging_file(base_dir: Path) -> StagedUpload: def abort_staged_upload(staged: StagedUpload) -> None: """Close and remove a staging file, tolerating repeated cleanup.""" - if not staged.handle.closed: - staged.handle.close() - staged.path.unlink(missing_ok=True) + close_error: BaseException | None = None + try: + if not staged.handle.closed: + staged.handle.close() + except BaseException as exc: + close_error = exc + try: + staged.path.unlink(missing_ok=True) + except BaseException as unlink_error: + if close_error is not None: + raise close_error from unlink_error + raise + if close_error is not None: + raise close_error def _validate_staged_upload(staged: StagedUpload) -> None: @@ -197,57 +216,146 @@ def _filename_candidates(name: str) -> Iterator[str]: counter += 1 -def publish_staged_upload(staged: StagedUpload, preferred_filename: str) -> Path: - """Atomically publish a complete staging file without replacing an entry.""" +def _unlink_matching_upload(path: Path, identity: UploadIdentity) -> None: + """Remove *path* only while it still names *identity*.""" + if identity.matches(path): + path.unlink(missing_ok=True) + + +def _rollback_link_without_masking(path: Path, identity: UploadIdentity) -> None: + try: + _unlink_matching_upload(path, identity) + except BaseException: + logger.warning("Failed to roll back partially published upload: %s", path, exc_info=True) + + +def _release_lease_without_masking(lease: UploadNameLease) -> None: + try: + lease.release() + except BaseException: + logger.warning("Failed to release upload name lease: %s", lease.lock_path, exc_info=True) + + +def publish_staged_upload_leased(staged: StagedUpload, preferred_filename: str) -> PublishedUpload: + """Atomically publish a staging file and retain its actual-name lease.""" safe_name = normalize_filename(preferred_filename) if not staged.handle.closed: staged.handle.close() - try: - _validate_staged_upload(staged) - for candidate_name in _filename_candidates(safe_name): - candidate = staged.base_dir / candidate_name + _validate_staged_upload(staged) + staged_identity = UploadIdentity.from_path(staged.path) + for candidate_name in _filename_candidates(safe_name): + lease = UploadNameLease.acquire(staged.base_dir, candidate_name) + candidate = staged.base_dir / candidate_name + linked = False + try: try: os.link(staged.path, candidate, follow_symlinks=False) + linked = True except FileExistsError: + lease.release() continue except (NotImplementedError, TypeError) as exc: raise AtomicUploadPublishError("Storage does not support atomic no-replace publication") from exc except OSError as exc: if exc.errno == errno.EEXIST: + lease.release() continue raise AtomicUploadPublishError(f"Storage does not support atomic no-replace publication: {exc}") from exc + if not staged_identity.matches(candidate): + raise AtomicUploadPublishError("Published upload identity changed during publication") try: staged.path.unlink() - except OSError: - logger.warning("Failed to remove published upload staging link: %s", staged.path, exc_info=True) - return candidate - except Exception: - staged.path.unlink(missing_ok=True) + except OSError as exc: + _rollback_link_without_masking(candidate, staged_identity) + raise AtomicUploadPublishError("Failed to remove upload staging link after publication") from exc + + try: + candidate_stat = os.lstat(candidate) + except FileNotFoundError as exc: + raise AtomicUploadPublishError("Published upload disappeared") from exc + if not stat.S_ISREG(candidate_stat.st_mode) or candidate_stat.st_nlink != 1 or not staged_identity.matches(candidate): + _rollback_link_without_masking(candidate, staged_identity) + raise AtomicUploadPublishError("Published upload did not become an exclusive regular file") + return PublishedUpload(path=candidate, identity=staged_identity, lease=lease) + except BaseException: + if linked: + _rollback_link_without_masking(candidate, staged_identity) + _release_lease_without_masking(lease) + raise + + +def publish_staged_upload(staged: StagedUpload, preferred_filename: str) -> Path: + """Atomically publish a complete staging file without replacing an entry.""" + publication = publish_staged_upload_leased(staged, preferred_filename) + try: + return publication.path + finally: + publication.release() + + +def _abort_staged_upload_without_masking(staged: StagedUpload) -> None: + try: + abort_staged_upload(staged) + except BaseException: + logger.warning("Failed to clean up upload staging file: %s", staged.path, exc_info=True) + + +def publish_upload_bytes_leased(base_dir: Path, preferred_filename: str, data: bytes) -> PublishedUpload: + """Stage bytes, publish them atomically, and retain the actual-name lease.""" + safe_name = normalize_filename(preferred_filename) + staged = create_upload_staging_file(base_dir) + try: + staged.handle.write(data) + return publish_staged_upload_leased(staged, safe_name) + except BaseException: + _abort_staged_upload_without_masking(staged) raise def publish_upload_bytes(base_dir: Path, preferred_filename: str, data: bytes) -> Path: """Stage and atomically publish an in-memory upload payload.""" + publication = publish_upload_bytes_leased(base_dir, preferred_filename, data) + try: + return publication.path + finally: + publication.release() + + +def publish_upload_copy_leased(base_dir: Path, preferred_filename: str, source_path: Path) -> PublishedUpload: + """Copy a source into staging, publish it, and retain the actual-name lease.""" + safe_name = normalize_filename(preferred_filename) staged = create_upload_staging_file(base_dir) try: - staged.handle.write(data) - return publish_staged_upload(staged, preferred_filename) - except Exception: - abort_staged_upload(staged) + with Path(source_path).open("rb") as source: + shutil.copyfileobj(source, staged.handle) + return publish_staged_upload_leased(staged, safe_name) + except BaseException: + _abort_staged_upload_without_masking(staged) raise def publish_upload_copy(base_dir: Path, preferred_filename: str, source_path: Path) -> Path: """Copy a local source into staging and atomically publish it.""" - staged = create_upload_staging_file(base_dir) + publication = publish_upload_copy_leased(base_dir, preferred_filename, source_path) try: - with Path(source_path).open("rb") as source: - shutil.copyfileobj(source, staged.handle) - return publish_staged_upload(staged, preferred_filename) - except Exception: - abort_staged_upload(staged) - raise + return publication.path + finally: + publication.release() + + +def rollback_published_upload(publication: PublishedUpload) -> None: + """Remove only the still-leased upload generation represented by *publication*.""" + if not publication.is_active: + raise RuntimeError("Cannot roll back a publication after releasing its lease") + if publication.lease.filename != publication.path.name: + raise UnsafeUploadPathError("Publication lease does not match its upload path") + if not publication.identity.matches(publication.path): + return + owned_conversion = existing_conversion_path_for_upload(publication.path) + publication.path.unlink() + if owned_conversion is not None: + owned_conversion.unlink(missing_ok=True) def replace_system_owned_staged_file(staged: StagedUpload, filename: str) -> Path: @@ -393,18 +501,22 @@ def delete_file_safe(base_dir: Path, filename: str) -> dict: if isinstance(exc.__cause__, FileNotFoundError): raise FileNotFoundError(f"File not found: {filename}") from exc raise - file_path = base_dir / safe_name + lease = UploadNameLease.acquire(base_dir, safe_name) try: - file_stat = os.lstat(file_path) - except FileNotFoundError: - raise FileNotFoundError(f"File not found: {filename}") - if not stat.S_ISREG(file_stat.st_mode) or file_stat.st_nlink != 1: - raise UnsafeUploadPathError(f"Unsafe upload file: {safe_name}") + file_path = base_dir / safe_name + try: + file_stat = os.lstat(file_path) + except FileNotFoundError: + raise FileNotFoundError(f"File not found: {filename}") from None + if not stat.S_ISREG(file_stat.st_mode) or file_stat.st_nlink != 1: + raise UnsafeUploadPathError(f"Unsafe upload file: {safe_name}") - owned_conversion = existing_conversion_path_for_upload(file_path) - file_path.unlink() - if owned_conversion is not None: - owned_conversion.unlink(missing_ok=True) + owned_conversion = existing_conversion_path_for_upload(file_path) + file_path.unlink() + if owned_conversion is not None: + owned_conversion.unlink(missing_ok=True) + finally: + lease.release() return {"success": True, "message": f"Deleted {filename}"} diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py index 6f2612233..a866b1dae 100644 --- a/backend/tests/test_uploads_manager.py +++ b/backend/tests/test_uploads_manager.py @@ -1,8 +1,13 @@ """Tests for deerflow.uploads.manager — shared upload management logic.""" import errno +import multiprocessing import os +import threading from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from queue import Empty +from typing import Any from unittest.mock import patch import pytest @@ -16,17 +21,39 @@ from deerflow.uploads.manager import ( AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError, + abort_staged_upload, claim_unique_filename, cleanup_stale_upload_staging_files, + create_upload_staging_file, delete_file_safe, list_files_in_dir, normalize_filename, + publish_staged_upload, publish_upload_bytes, + publish_upload_bytes_leased, publish_upload_copy, + rollback_published_upload, validate_path_traversal, write_upload_file_no_symlink, ) + +def _delete_upload_in_process( + uploads_dir: str, + filename: str, + started: Any, + finished: Any, + errors: Any, +) -> None: + try: + started.set() + delete_file_safe(Path(uploads_dir), filename) + except BaseException as exc: # pragma: no cover - surfaced in the parent + errors.put(repr(exc)) + finally: + finished.set() + + # --------------------------------------------------------------------------- # normalize_filename # --------------------------------------------------------------------------- @@ -119,6 +146,120 @@ class TestValidatePathTraversal: class TestUploadPublication: + def test_reserved_staging_name_is_rejected_before_stage_creation(self, tmp_path): + with patch("deerflow.uploads.manager.create_upload_staging_file") as create_stage: + with pytest.raises(ValueError, match="reserved"): + publish_upload_bytes(tmp_path, ".upload-user.part", b"payload") + + create_stage.assert_not_called() + + def test_leased_publication_blocks_delete_until_release(self, tmp_path): + publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old") + started = threading.Event() + finished = threading.Event() + + def delete(): + started.set() + delete_file_safe(tmp_path, "report.pdf") + finished.set() + + worker = threading.Thread(target=delete) + worker.start() + try: + assert started.wait(1) + assert not finished.wait(0.1) + finally: + publication.release() + worker.join(2) + + assert finished.is_set() + + def test_leased_publication_blocks_delete_across_processes(self, tmp_path): + publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old") + context = multiprocessing.get_context("spawn") + started = context.Event() + finished = context.Event() + errors = context.Queue() + worker = context.Process( + target=_delete_upload_in_process, + args=(str(tmp_path), "report.pdf", started, finished, errors), + ) + worker.start() + try: + assert started.wait(10) + assert not finished.wait(0.2) + finally: + publication.release() + worker.join(10) + if worker.is_alive(): + worker.terminate() + worker.join(2) + + assert worker.exitcode == 0 + try: + child_error = errors.get_nowait() + except Empty: + child_error = None + assert child_error is None + assert finished.is_set() + + def test_lease_for_one_filename_does_not_block_another(self, tmp_path): + publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old") + try: + with ThreadPoolExecutor(max_workers=1) as pool: + other = pool.submit(publish_upload_bytes, tmp_path, "notes.txt", b"new") + assert other.result(timeout=1) == tmp_path / "notes.txt" + finally: + publication.release() + + def test_rollback_does_not_remove_reused_path(self, tmp_path): + publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old") + try: + publication.path.unlink() + publication.path.write_bytes(b"new") + rollback_published_upload(publication) + finally: + publication.release() + + assert (tmp_path / "report.pdf").read_bytes() == b"new" + + def test_staging_unlink_failure_is_not_reported_as_success(self, tmp_path): + staged = create_upload_staging_file(tmp_path) + staged.handle.write(b"payload") + real_unlink = Path.unlink + + def fail_only_for_stage(path, *args, **kwargs): + if path == staged.path: + raise OSError("cannot unlink stage") + return real_unlink(path, *args, **kwargs) + + with patch.object(Path, "unlink", autospec=True, side_effect=fail_only_for_stage): + with pytest.raises(AtomicUploadPublishError, match="staging"): + publish_staged_upload(staged, "report.pdf") + + assert not (tmp_path / "report.pdf").exists() + + def test_abort_unlinks_stage_when_close_raises(self, tmp_path): + staged = create_upload_staging_file(tmp_path) + + class CloseFailingHandle: + def __init__(self, wrapped): + self._wrapped = wrapped + + @property + def closed(self): + return self._wrapped.closed + + def close(self): + self._wrapped.close() + raise OSError("close failed") + + staged.handle = CloseFailingHandle(staged.handle) + with pytest.raises(OSError, match="close failed"): + abort_staged_upload(staged) + + assert not staged.path.exists() + def test_compatibility_wrapper_writes_new_file(self, tmp_path): dest = write_upload_file_no_symlink(tmp_path, "notes.txt", b"hello")