diff --git a/app/assets/services/ingest.py b/app/assets/services/ingest.py index c20cf943112..bbc5665b30e 100644 --- a/app/assets/services/ingest.py +++ b/app/assets/services/ingest.py @@ -1,16 +1,19 @@ """Turns incoming bytes into catalogued assets: multipart uploads moved into a hash-addressed destination, files registered where they already sit, and -records created from a hash the catalog already holds. Every path persists the -stat that hashing verified, so a row's recorded size and mtime describe the -same observation as its hash. A live row already at the destination is -reconciled before the write, so an upload never adopts a fresh hash onto -records created for bytes it just replaced. +records created from a hash the catalog already holds. Registration persists +the stat that hashing verified, so a row's recorded size and mtime describe the +same observation as its hash; an upload records the stat of the file once it is +in place, since a copy across volumes has its own mtime. A live row already at +the destination is reconciled before the write, so an upload never adopts a +fresh hash onto records created for bytes it just replaced. """ import contextlib +import errno import logging import mimetypes import os +import shutil from typing import Any, NamedTuple, Sequence from sqlalchemy import false, func, select @@ -187,7 +190,12 @@ def _guess_upload_mime_type( def _move_temp_to_dest(temp_path: str, dest_abs: str) -> None: os.makedirs(os.path.dirname(dest_abs), exist_ok=True) try: - os.replace(temp_path, dest_abs) + try: + os.replace(temp_path, dest_abs) + except OSError as e: # EXDEV: destination is on another volume + if e.errno != errno.EXDEV: + raise + shutil.copyfile(temp_path, dest_abs) except Exception as e: raise RuntimeError(f"failed to move uploaded file into place: {e}") from e @@ -448,7 +456,7 @@ def upload_from_temp_path( user_metadata = user_metadata or {} try: - digest, verified_stat = _snapshot_hash_with_retry(temp_path) + digest, _ = _snapshot_hash_with_retry(temp_path) except UploadUnstableError: _remove_temp_path(temp_path) raise @@ -489,7 +497,9 @@ def upload_from_temp_path( _move_temp_to_dest(temp_path, dest_abs) finally: _remove_temp_path(temp_path) - size_bytes, mtime_ns = verified_stat.st_size, verified_stat.st_mtime_ns + # A cross-volume copy gets a new mtime, so record the file on disk (a rename keeps it). + placed_stat = os.stat(dest_abs) + size_bytes, mtime_ns = placed_stat.st_size, placed_stat.st_mtime_ns system_metadata = _extract_system_metadata_sync(dest_abs, content_type) with create_session() as session: _reconcile_live_content_at_path( diff --git a/tests-unit/assets_test/services/test_upload_cross_device.py b/tests-unit/assets_test/services/test_upload_cross_device.py new file mode 100644 index 00000000000..d9e63cab69d --- /dev/null +++ b/tests-unit/assets_test/services/test_upload_cross_device.py @@ -0,0 +1,122 @@ +"""Uploads whose destination is on a different volume than the temp upload, where +the rename fails with EXDEV (WinError 17 on Windows) and the bytes are copied.""" + +import errno +import os +import uuid +from pathlib import Path + +import pytest +from sqlalchemy import select + +import app.assets.mode as mode_module +import app.assets.services.ingest as ingest_module +import folder_paths +from app.assets.database.models import AssetContent +from app.assets.services.ingest import upload_from_temp_path +from app.assets.services.snapshot_hash import snapshot_hash + +_CONTENT = b"cross-device upload bytes" +_OLD_MTIME_NS = 1_600_000_000_000_000_000 + + +@pytest.fixture +def cross_device_upload(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + class FakeArgs: + enable_asset_hashing = True + + mode_module.init(FakeArgs()) + monkeypatch.setattr(folder_paths, "get_temp_directory", lambda: str(tmp_path / "temp")) + monkeypatch.setattr(folder_paths, "get_input_directory", lambda: str(tmp_path / "input")) + + def exdev(src, dst): + raise OSError(errno.EXDEV, "Invalid cross-device link") + + monkeypatch.setattr(ingest_module.os, "replace", exdev) + temp = tmp_path / "temp" / "uploads" / uuid.uuid4().hex / ".upload.part" + temp.parent.mkdir(parents=True) + temp.write_bytes(_CONTENT) + os.utime(temp, ns=(_OLD_MTIME_NS, _OLD_MTIME_NS)) + dest = tmp_path / "input" / f"{snapshot_hash(str(temp))[0]}.png" + yield temp, dest + mode_module.init(None) + + +def _upload(temp: Path): + return upload_from_temp_path( + temp_path=str(temp), name="photo.png", tags=["input"], client_filename="photo.png" + ) + + +def _recorded_mtime_ns(session_factory, dest: Path) -> int: + with session_factory() as session: + return session.scalars( + select(AssetContent.mtime_ns).where(AssetContent.path == str(dest)) + ).one() + + +def test_cross_device_upload_is_copied_into_place(mock_create_session, cross_device_upload): + temp, dest = cross_device_upload + + result = _upload(temp) + + assert result.created_new is True + assert dest.read_bytes() == _CONTENT + assert not temp.exists() + # The copy has its own mtime, and that is what must be recorded. + assert dest.stat().st_mtime_ns != _OLD_MTIME_NS + assert _recorded_mtime_ns(mock_create_session, dest) == dest.stat().st_mtime_ns + + +@pytest.mark.parametrize( + "existing", [_CONTENT[::-1], _CONTENT[:5]], ids=["same-size-other-bytes", "truncated"] +) +def test_cross_device_upload_replaces_other_bytes_at_the_destination( + mock_create_session, cross_device_upload, existing +): + temp, dest = cross_device_upload + dest.parent.mkdir(parents=True) + dest.write_bytes(existing) + + _upload(temp) + + assert dest.read_bytes() == _CONTENT + + +def test_failed_cross_device_copy_fails_the_upload_and_registers_nothing( + mock_create_session, cross_device_upload, monkeypatch +): + temp, dest = cross_device_upload + + def disk_full(src, dst): + Path(dst).write_bytes(_CONTENT[:5]) + raise OSError(errno.ENOSPC, "No space left on device") + + monkeypatch.setattr(ingest_module.shutil, "copyfile", disk_full) + + with pytest.raises(RuntimeError, match="failed to move uploaded file into place"): + _upload(temp) + + assert not temp.exists() + with mock_create_session() as session: + assert session.scalars( + select(AssetContent).where(AssetContent.path == str(dest)) + ).first() is None + + +def test_other_move_errors_raise_without_copying( + mock_create_session, cross_device_upload, monkeypatch +): + temp, _dest = cross_device_upload + + def denied(src, dst): + raise PermissionError(errno.EACCES, "Access is denied") + + monkeypatch.setattr(ingest_module.os, "replace", denied) + copies: list = [] + monkeypatch.setattr(ingest_module.shutil, "copyfile", lambda *a: copies.append(a)) + + with pytest.raises(RuntimeError, match="failed to move uploaded file into place"): + _upload(temp) + + assert copies == []