From 62641f9d12993cad53a178224d4c3c1d62e535c1 Mon Sep 17 00:00:00 2001 From: Simon Pinfold Date: Wed, 26 Aug 2026 11:46:00 -0700 Subject: [PATCH] fix(assets): clean orphan content on failed ingest (review-11) --- app/assets/scanner.py | 97 ++++++----- app/assets/services/ingest.py | 56 ++++--- .../services/test_ingest_orphan_content.py | 152 ++++++++++++++++++ 3 files changed, 243 insertions(+), 62 deletions(-) create mode 100644 tests-unit/assets_test/services/test_ingest_orphan_content.py diff --git a/app/assets/scanner.py b/app/assets/scanner.py index 0666e87a9..805200ab7 100644 --- a/app/assets/scanner.py +++ b/app/assets/scanner.py @@ -41,6 +41,7 @@ from app.assets.services.path_utils import ( get_comfy_models_folders, get_name_and_tags_from_asset_path, ) +from app.assets.services.ingest import _discard_unreferenced_content from app.assets.services.snapshot_hash import snapshot_hash from app.database.db import create_session @@ -321,49 +322,65 @@ def build_asset_specs( def seed_asset_specs(session: Session, specs: list[SeedAssetSpec]) -> int: - """Create one B content row and one birth-classified record per new path.""" + """Create one B content row and one birth-classified record per new path. + + ``create_content`` inserts inside a SAVEPOINT that survives an outer rollback + under pysqlite, so a mid-batch ``create_record`` failure would leak the live + content rows created so far as unreferenced orphans (and a live orphan at a + path makes later scans skip it indefinitely). On any failure we roll the + aborted batch back and discard every content row we created, mirroring the + executed-output path, before letting the error propagate. + """ created = 0 - for spec in specs: - path = os.path.abspath(spec["abs_path"]) - try: - stat_result = os.stat(path, follow_symlinks=True) - except OSError: - logging.warning("Skipping vanished asset during scan: %s", path) - continue - try: - recovery = recover_missing_content( + created_content_ids: list[str] = [] + try: + for spec in specs: + path = os.path.abspath(spec["abs_path"]) + try: + stat_result = os.stat(path, follow_symlinks=True) + except OSError: + logging.warning("Skipping vanished asset during scan: %s", path) + continue + try: + recovery = recover_missing_content( + session, + path, + stat_result, + hashing_is_enabled=mode.hashing_enabled(), + ) + except OSError: + logging.warning("Skipping vanished asset during scan: %s", path) + continue + if recovery != "no_match": + continue + content = create_content( session, - path, - stat_result, - hashing_is_enabled=mode.hashing_enabled(), + path=path, + hash=None, + size_bytes=spec["size_bytes"], + mtime_ns=spec["mtime_ns"], ) - except OSError: - logging.warning("Skipping vanished asset during scan: %s", path) - continue - if recovery != "no_match": - continue - content = create_content( - session, - path=path, - hash=None, - size_bytes=spec["size_bytes"], - mtime_ns=spec["mtime_ns"], - ) - existing_record = session.scalar( - sa.select(Asset.id).where(Asset.content_id == content.id).limit(1) - ) - if existing_record is not None: - continue - create_record( - session, - content_id=content.id, - name=spec["info_name"], - mime_type=spec["mime_type"], - job_id=spec["job_id"], - loader_path=spec["fname"], - tags=spec["tags"], - ) - created += 1 + created_content_ids.append(content.id) + existing_record = session.scalar( + sa.select(Asset.id).where(Asset.content_id == content.id).limit(1) + ) + if existing_record is not None: + continue + create_record( + session, + content_id=content.id, + name=spec["info_name"], + mime_type=spec["mime_type"], + job_id=spec["job_id"], + loader_path=spec["fname"], + tags=spec["tags"], + ) + created += 1 + except Exception: + session.rollback() + for content_id in created_content_ids: + _discard_unreferenced_content(session, content_id) + raise return created diff --git a/app/assets/services/ingest.py b/app/assets/services/ingest.py index 0bd93fe7b..406a43202 100644 --- a/app/assets/services/ingest.py +++ b/app/assets/services/ingest.py @@ -302,17 +302,23 @@ def upload_from_temp_path( content = create_content( session, dest_abs, stored_hash, size_bytes, mtime_ns ) - record = _create_upload_record( - session, - content.id, - display_name, - dest_abs, - [*(tags or []), "uploaded"], - content_type, - user_metadata, - preview_id, - ) - session.commit() + created_content_id = content.id + try: + record = _create_upload_record( + session, + content.id, + display_name, + dest_abs, + [*(tags or []), "uploaded"], + content_type, + user_metadata, + preview_id, + ) + session.commit() + except Exception: + session.rollback() + _discard_unreferenced_content(session, created_content_id) + raise return _record_to_upload_result(session, record, created_new=True) @@ -397,17 +403,23 @@ def register_file_in_place( content = create_content( session, locator, stored_hash, size_bytes, mtime_ns ) - record = _create_upload_record( - session, - content.id, - display_name, - locator, - merged_tags, - content_type, - None, - None, - ) - session.commit() + created_content_id = content.id + try: + record = _create_upload_record( + session, + content.id, + display_name, + locator, + merged_tags, + content_type, + None, + None, + ) + session.commit() + except Exception: + session.rollback() + _discard_unreferenced_content(session, created_content_id) + raise return _record_to_upload_result(session, record, created_new=True) diff --git a/tests-unit/assets_test/services/test_ingest_orphan_content.py b/tests-unit/assets_test/services/test_ingest_orphan_content.py new file mode 100644 index 000000000..786e6f72c --- /dev/null +++ b/tests-unit/assets_test/services/test_ingest_orphan_content.py @@ -0,0 +1,152 @@ +"""Orphan-content regression: a failed record insert must not leak content. + +``create_content`` inserts inside a SAVEPOINT (``begin_nested``). Under pysqlite +that insert survives the enclosing ``rollback`` because pysqlite has no real +nested transaction, so a follow-on ``create_record`` failure would otherwise +leave a live, unreferenced ``AssetContent`` row behind. A live row occupying a +path makes later scans skip it indefinitely, so every ingest path that creates +content before its record must discard the orphan on failure. + +These tests inject a ``create_record`` failure AFTER ``create_content`` has +succeeded, then assert no live ``AssetContent`` row is left behind, for each of +the three claimed paths: ``upload_from_temp_path``, ``register_file_in_place`` +(ingest) and ``seed_asset_specs`` (scanner). +""" +import os +import uuid + +import pytest +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +import app.assets.mode as mode_module +import folder_paths +from app.assets.database.models import Asset, AssetContent +from app.assets.database.queries import create_record as create_record_query +from app.assets.scanner import SeedAssetSpec, seed_asset_specs +from app.assets.services import ingest +from app.assets.services.ingest import register_file_in_place, upload_from_temp_path + + +@pytest.fixture +def hashing_off(): + class FakeArgs: + enable_asset_hashing = False + + mode_module.init(FakeArgs()) + yield + mode_module.init(None) + + +def _write_temp(content: bytes) -> str: + uploads_root = os.path.join( + folder_paths.get_temp_directory(), "uploads", uuid.uuid4().hex + ) + os.makedirs(uploads_root, exist_ok=True) + path = os.path.join(uploads_root, ".upload.part") + with open(path, "wb") as file: + file.write(content) + return path + + +def _raise_create_record(*_args, **_kwargs): + raise RuntimeError("forced create_record failure") + + +def _live_content_count(session: Session) -> int: + return session.scalar( + select(func.count()) + .select_from(AssetContent) + .where(AssetContent.is_missing.is_(False)) + ) + + +def test_upload_from_temp_path_discards_content_on_record_failure( + mock_create_session, hashing_off, monkeypatch: pytest.MonkeyPatch +) -> None: + # Given a fresh upload whose record insert will fail after content creation + content = b"orphan-upload-" + uuid.uuid4().bytes + temp_path = _write_temp(content) + monkeypatch.setattr(ingest, "create_record", _raise_create_record) + + # When the upload runs and create_record blows up + with pytest.raises(RuntimeError, match="forced create_record failure"): + upload_from_temp_path( + temp_path=temp_path, + name="orphan.bin", + tags=["output"], + client_filename="orphan.bin", + ) + + # Then no live content row is left orphaned + with mock_create_session() as session: + assert _live_content_count(session) == 0 + + +def test_register_file_in_place_discards_content_on_record_failure( + mock_create_session, hashing_off, monkeypatch: pytest.MonkeyPatch +) -> None: + # Given a fresh on-disk file whose record insert will fail after content creation + output_dir = folder_paths.get_output_directory() + os.makedirs(output_dir, exist_ok=True) + path = os.path.join(output_dir, f"orphan_inplace_{uuid.uuid4().hex}.png") + with open(path, "wb") as file: + file.write(b"orphan-inplace-" + uuid.uuid4().bytes) + monkeypatch.setattr(ingest, "create_record", _raise_create_record) + + try: + # When registration runs and create_record blows up + with pytest.raises(RuntimeError, match="forced create_record failure"): + register_file_in_place(abs_path=path, name="orphan_inplace.png", tags=["output"]) + + # Then no live content row is left orphaned + with mock_create_session() as session: + assert _live_content_count(session) == 0 + finally: + if os.path.exists(path): + os.unlink(path) + + +def _seed_spec(path: str, size_bytes: int, mtime_ns: int, name: str) -> SeedAssetSpec: + return { + "abs_path": path, + "size_bytes": size_bytes, + "mtime_ns": mtime_ns, + "info_name": name, + "tags": ["input"], + "fname": name, + "metadata": None, + "mime_type": None, + "job_id": None, + } + + +def test_seed_asset_specs_discards_content_on_record_failure( + session: Session, tmp_path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Given three specs where the second record insert fails after content creation + specs: list[SeedAssetSpec] = [] + fail_name = "vanished.bin" + for name in ("first.bin", fail_name, "last.bin"): + file_path = tmp_path / name + file_path.write_bytes(name.encode()) + stat_result = file_path.stat() + specs.append( + _seed_spec(str(file_path), stat_result.st_size, stat_result.st_mtime_ns, name) + ) + + def _create_record_or_raise(session_arg, content_id, name, *args, **kwargs): + if name == fail_name: + raise RuntimeError("forced create_record failure") + return create_record_query(session_arg, content_id, name, *args, **kwargs) + + monkeypatch.setattr("app.assets.scanner.create_record", _create_record_or_raise) + + # When the batch seed runs and the second record insert blows up + with pytest.raises(RuntimeError, match="forced create_record failure"): + seed_asset_specs(session, specs) + session.rollback() + + # Then no live content row is left orphaned by the aborted batch + assert _live_content_count(session) == 0 + assert session.scalar(select(func.count()).select_from(Asset)) == 0