[DO NOT MERGE] Assets: integration branch for #16358 + #16393

alembic_db/versions/0007_record_content_split.py

@@ -67,19 +67,6 @@ def upgrade() -> None:
     op.create_index("ix_assets_name", "assets", ["name"])
     op.create_index("ix_assets_created_at", "assets", ["created_at"])
     op.create_index("ix_assets_preview_id", "assets", ["preview_id"])
-    op.create_table(
-        "asset_meta",
-        sa.Column("asset_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True),
-        sa.Column("key", sa.String(256), primary_key=True),
-        sa.Column("ordinal", sa.Integer(), primary_key=True),
-        sa.Column("val_str", sa.String(2048)), sa.Column("val_num", sa.Numeric(38, 10)),
-        sa.Column("val_bool", sa.Boolean()), sa.Column("val_json", sa.JSON()),
-        sa.CheckConstraint("val_str IS NOT NULL OR val_num IS NOT NULL OR val_bool IS NOT NULL OR val_json IS NOT NULL", name="ck_asset_meta_has_value"),
-    )
-    op.create_index("ix_asset_meta_key", "asset_meta", ["key"])
-    op.create_index("ix_asset_meta_key_val_str", "asset_meta", ["key", "val_str"])
-    op.create_index("ix_asset_meta_key_val_num", "asset_meta", ["key", "val_num"])
-    op.create_index("ix_asset_meta_key_val_bool", "asset_meta", ["key", "val_bool"])
     op.create_table("asset_tags", sa.Column("asset_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True), sa.Column("tag_name", sa.String(512), sa.ForeignKey("tags.name", ondelete="RESTRICT"), primary_key=True), sa.Column("origin", sa.String(32), nullable=False), sa.Column("added_at", sa.DateTime(), nullable=False))
     op.create_index("ix_asset_tags_tag_name", "asset_tags", ["tag_name"])
     op.create_index("ix_asset_tags_asset_id", "asset_tags", ["asset_id"])
@@ -89,7 +76,7 @@ def upgrade() -> None:
 def downgrade() -> None:
     op.drop_table("asset_system_state")
     op.drop_table("asset_tags")
-    op.drop_table("asset_meta")
+    op.execute("DROP TABLE IF EXISTS asset_meta")
     op.drop_table("assets")
     op.drop_table("asset_contents")
     op.create_table(

app/assets/api/routes.py

@@ -133,6 +133,12 @@ def register_assets_routes(
     app.add_routes(ROUTES)
 
 
+def close_assets_feature_gate() -> None:
+    """Answer 503 from routes that are already registered."""
+    global _ASSETS_ENABLED
+    _ASSETS_ENABLED = False
+
+
 def _build_error_response(
     status: int, code: str, message: str, details: dict | None = None
 ) -> web.Response:
@@ -1126,6 +1132,8 @@ async def mark_missing_assets(request: web.Request) -> web.Response:
     Returns:
         200 OK with count of marked assets
         409 Conflict if a scan is currently running
+        500 Internal Server Error with PRUNE_FAILED if the marking failed, so a
+            prune that did not run is never reported as a completed one
     """
     try:
         marked = asset_seeder.mark_missing_outside_prefixes()
@@ -1134,4 +1142,10 @@ async def mark_missing_assets(request: web.Request) -> web.Response:
             {"status": "scan_running", "marked": 0},
             status=409,
         )
+    if marked is None:
+        return _build_error_response(
+            500,
+            "PRUNE_FAILED",
+            "Failed to mark missing assets.",
+        )
     return web.json_response({"status": "completed", "marked": marked}, status=200)

app/assets/api/upload.py

@@ -1,9 +1,7 @@
 """Reads a multipart upload off the wire and lands its bytes in a temporary file
 the ingest service can hash and move. The body is consumed in chunks so a large
-model never has to fit in memory, and the temporary file is removed on every
-failure path so an abandoned upload leaves nothing behind. Field values are
-validated as they arrive, letting a bad request fail before its bytes are
-written.
+model never has to fit in memory. Validation failures remove any temporary file
+before they are reported to the caller.
 """
 
 import logging
@@ -65,101 +63,115 @@ async def parse_multipart_upload(
 
     file_written = 0
     tmp_path: str | None = None
+    parse_succeeded = False
 
-    while True:
-        field = await reader.next()
-        if field is None:
-            break
-
-        fname = getattr(field, "name", "") or ""
+    try:
+        while True:
+            field = await reader.next()
+            if field is None:
+                break
 
-        if fname == "hash":
-            try:
-                s = ((await field.text()) or "").strip().lower()
-            except Exception:
-                raise UploadError(
-                    400, "INVALID_HASH", "hash must be like 'blake3:<hex>'"
-                )
+            fname = getattr(field, "name", "") or ""
 
-            if s:
-                provided_hash = normalize_and_validate_hash(s)
+            if fname == "hash":
                 try:
-                    provided_hash_exists = check_hash_exists(provided_hash)
-                except Exception as e:
-                    logging.exception(
-                        "check_hash_exists failed for hash=%s: %s", provided_hash, e
+                    s = ((await field.text()) or "").strip().lower()
+                except Exception:
+                    raise UploadError(
+                        400, "INVALID_HASH", "hash must be like 'blake3:<hex>'"
+                    )
+
+                if s:
+                    provided_hash = normalize_and_validate_hash(s)
+                    try:
+                        provided_hash_exists = check_hash_exists(provided_hash)
+                    except Exception as e:
+                        logging.exception(
+                            "check_hash_exists failed for hash=%s: %s", provided_hash, e
+                        )
+                        raise UploadError(
+                            500,
+                            "HASH_CHECK_FAILED",
+                            "Backend error while checking asset hash.",
+                        )
+
+            elif fname == "file":
+                if file_present:
+                    raise UploadError(
+                        400,
+                        "UNSUPPORTED_FIELD",
+                        "Multiple 'file' parts are not supported.",
                     )
+                file_present = True
+                file_client_name = (field.filename or "").strip()
+
+                uploads_root = os.path.join(folder_paths.get_temp_directory(), "uploads")
+                unique_dir = os.path.join(uploads_root, uuid.uuid4().hex)
+                os.makedirs(unique_dir, exist_ok=True)
+                tmp_path = os.path.join(unique_dir, ".upload.part")
+
+                try:
+                    with open(tmp_path, "wb") as f:
+                        while True:
+                            chunk = await field.read_chunk(8 * 1024 * 1024)
+                            if not chunk:
+                                break
+                            f.write(chunk)
+                            file_written += len(chunk)
+                except Exception:
                     raise UploadError(
-                        500,
-                        "HASH_CHECK_FAILED",
-                        "Backend error while checking asset hash.",
+                        500, "UPLOAD_IO_ERROR", "Failed to receive and store uploaded file."
                     )
 
-        elif fname == "file":
-            file_present = True
-            file_client_name = (field.filename or "").strip()
-
-            uploads_root = os.path.join(folder_paths.get_temp_directory(), "uploads")
-            unique_dir = os.path.join(uploads_root, uuid.uuid4().hex)
-            os.makedirs(unique_dir, exist_ok=True)
-            tmp_path = os.path.join(unique_dir, ".upload.part")
-
-            try:
-                with open(tmp_path, "wb") as f:
-                    while True:
-                        chunk = await field.read_chunk(8 * 1024 * 1024)
-                        if not chunk:
-                            break
-                        f.write(chunk)
-                        file_written += len(chunk)
-            except Exception:
-                delete_temp_file_if_exists(tmp_path)
+            elif fname == "tags":
+                tags_raw.append((await field.text()) or "")
+            elif fname == "name":
+                provided_name = (await field.text()) or None
+            elif fname == "user_metadata":
+                user_metadata_raw = (await field.text()) or None
+            elif fname == "id":
                 raise UploadError(
-                    500, "UPLOAD_IO_ERROR", "Failed to receive and store uploaded file."
+                    400,
+                    "UNSUPPORTED_FIELD",
+                    "Client-provided 'id' is not supported. Asset IDs are assigned by the server.",
                 )
+            elif fname == "mime_type":
+                provided_mime_type = ((await field.text()) or "").strip() or None
+            elif fname == "preview_id":
+                provided_preview_id = ((await field.text()) or "").strip() or None
 
-        elif fname == "tags":
-            tags_raw.append((await field.text()) or "")
-        elif fname == "name":
-            provided_name = (await field.text()) or None
-        elif fname == "user_metadata":
-            user_metadata_raw = (await field.text()) or None
-        elif fname == "id":
+        if not file_present and not (provided_hash and provided_hash_exists):
             raise UploadError(
                 400,
-                "UNSUPPORTED_FIELD",
-                "Client-provided 'id' is not supported. Asset IDs are assigned by the server.",
+                "MISSING_FILE",
+                "Form must include a 'file' part or a known 'hash'.",
             )
-        elif fname == "mime_type":
-            provided_mime_type = ((await field.text()) or "").strip() or None
-        elif fname == "preview_id":
-            provided_preview_id = ((await field.text()) or "").strip() or None
-    if not file_present and not (provided_hash and provided_hash_exists):
-        raise UploadError(
-            400, "MISSING_FILE", "Form must include a 'file' part or a known 'hash'."
-        )
 
-    if (
-        file_present
-        and file_written == 0
-        and not (provided_hash and provided_hash_exists)
-    ):
-        delete_temp_file_if_exists(tmp_path)
-        raise UploadError(400, "EMPTY_UPLOAD", "Uploaded file is empty.")
-
-    return ParsedUpload(
-        file_present=file_present,
-        file_written=file_written,
-        file_client_name=file_client_name,
-        tmp_path=tmp_path,
-        tags_raw=tags_raw,
-        provided_name=provided_name,
-        user_metadata_raw=user_metadata_raw,
-        provided_hash=provided_hash,
-        provided_hash_exists=provided_hash_exists,
-        provided_mime_type=provided_mime_type,
-        provided_preview_id=provided_preview_id,
-    )
+        if (
+            file_present
+            and file_written == 0
+            and not (provided_hash and provided_hash_exists)
+        ):
+            raise UploadError(400, "EMPTY_UPLOAD", "Uploaded file is empty.")
+
+        parsed = ParsedUpload(
+            file_present=file_present,
+            file_written=file_written,
+            file_client_name=file_client_name,
+            tmp_path=tmp_path,
+            tags_raw=tags_raw,
+            provided_name=provided_name,
+            user_metadata_raw=user_metadata_raw,
+            provided_hash=provided_hash,
+            provided_hash_exists=provided_hash_exists,
+            provided_mime_type=provided_mime_type,
+            provided_preview_id=provided_preview_id,
+        )
+        parse_succeeded = True
+        return parsed
+    finally:
+        if not parse_succeeded:
+            delete_temp_file_if_exists(tmp_path)
 
 
 def delete_temp_file_if_exists(tmp_path: str | None) -> None:

app/assets/database/models.py

@@ -1,5 +1,5 @@
 """Declares the asset schema: content rows describing bytes on disk, records
-describing what a user sees, and the tag and metadata tables hanging off them.
+describing what a user sees, and the asset_tags table hanging off them.
 The split is the point — many records can name one content row, and retiring
 content by marking it missing rather than deleting it is what keeps a path's
 history intact. Constraints declared here, not application code, are what make
@@ -10,7 +10,6 @@
 
 import uuid
 from datetime import datetime
-from decimal import Decimal
 from typing import Any
 
 from sqlalchemy import (
@@ -21,8 +20,6 @@
     DateTime,
     ForeignKey,
     Index,
-    Integer,
-    Numeric,
     String,
     Text,
     text,
@@ -103,9 +100,6 @@ class Asset(Base):
     preview: Mapped[Asset | None] = relationship(
         "Asset", foreign_keys=[preview_id], remote_side=lambda: [Asset.id]
     )
-    metadata_entries: Mapped[list[AssetMeta]] = relationship(
-        back_populates="asset", cascade="all,delete-orphan", passive_deletes=True
-    )
     tag_links: Mapped[list[AssetTag]] = relationship(
         back_populates="asset", cascade="all,delete-orphan", passive_deletes=True
     )
@@ -121,33 +115,6 @@ class Asset(Base):
     )
 
 
-class AssetMeta(Base):
-    __tablename__ = "asset_meta"
-
-    asset_id: Mapped[str] = mapped_column(
-        String(36), ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True
-    )
-    key: Mapped[str] = mapped_column(String(256), primary_key=True)
-    ordinal: Mapped[int] = mapped_column(Integer, primary_key=True, default=0)
-    val_str: Mapped[str | None] = mapped_column(String(2048))
-    val_num: Mapped[Decimal | None] = mapped_column(Numeric(38, 10))
-    val_bool: Mapped[bool | None] = mapped_column(Boolean)
-    val_json: Mapped[Any | None] = mapped_column(JSON)
-
-    asset: Mapped[Asset] = relationship(back_populates="metadata_entries")
-
-    __table_args__ = (
-        Index("ix_asset_meta_key", "key"),
-        Index("ix_asset_meta_key_val_str", "key", "val_str"),
-        Index("ix_asset_meta_key_val_num", "key", "val_num"),
-        Index("ix_asset_meta_key_val_bool", "key", "val_bool"),
-        CheckConstraint(
-            "val_str IS NOT NULL OR val_num IS NOT NULL OR val_bool IS NOT NULL OR val_json IS NOT NULL",
-            name="ck_asset_meta_has_value",
-        ),
-    )
-
-
 class AssetTag(Base):
     __tablename__ = "asset_tags"
 

app/assets/database/queries/__init__.py

@@ -13,6 +13,7 @@
     delete_record,
     fetch_record_tags,
     get_record_by_id,
+    is_live_path_conflict,
     list_records_page,
     mark_content_missing,
     rename_record,
@@ -27,6 +28,7 @@
     "delete_record",
     "fetch_record_tags",
     "get_record_by_id",
+    "is_live_path_conflict",
     "list_records_page",
     "mark_content_missing",
     "rename_record",

app/assets/database/queries/records.py

@@ -1,5 +1,5 @@
-"""Owns every write to content rows, records and their tag links, plus the paged
-reads that list them. Inserts that can lose a race — a content row at a path, a
+"""Provides shared writes for content rows, records and tag links, plus the paged
+reads that list records. Inserts that can lose a race — a content row at a path, a
 tag, a tag link — run inside a savepoint and re-read the conflicting row, so a
 concurrent writer settles the call instead of raising, while a genuine
 constraint failure still surfaces. This is the sole writer of a content row's
@@ -47,7 +47,7 @@ class RecordPageSpec(NamedTuple):
 _LIVE_PATH_UNIQUE_INDEX = "uq_asset_contents_path_live"
 
 
-def _is_live_path_conflict(error: IntegrityError) -> bool:
+def is_live_path_conflict(error: IntegrityError) -> bool:
     orig = error.orig
     message = str(orig)
     postgres_names_the_index = getattr(getattr(orig, "diag", None), "constraint_name", None) == _LIVE_PATH_UNIQUE_INDEX
@@ -66,7 +66,7 @@ def create_content_reporting_insert(session: Session, path: str, hash: str | Non
             session.flush()
             return content, True
     except IntegrityError as error:
-        if not _is_live_path_conflict(error):
+        if not is_live_path_conflict(error):
             raise
         winner = session.execute(sa.select(AssetContent).where(AssetContent.path == path, AssetContent.is_missing.is_(False))).scalar_one()
         return winner, False
@@ -299,8 +299,9 @@ def rename_record(session: Session, id: str, name: str) -> Asset:
     record = session.get(Asset, id)
     if record is None:
         raise LookupError(id)
-    record.name = name
-    record.updated_at = get_utc_now()
+    if record.name != name:
+        record.name = name
+        record.updated_at = get_utc_now()
     session.flush()
     return record