fix(assets): clean orphan content on failed ingest (review-11)

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
 
 

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)
 
 

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