diff --git a/alembic_db/versions/0007_add_asset_semantics_version.py b/alembic_db/versions/0007_add_asset_semantics_version.py new file mode 100644 index 000000000..4aa446b03 --- /dev/null +++ b/alembic_db/versions/0007_add_asset_semantics_version.py @@ -0,0 +1,34 @@ +""" +Add asset_semantics_version table. + +Alembic records the *shape* of the assets tables. This table records the +*meaning* of their contents: which generation of the derivation logic produced +the values currently stored in them. The two move independently, so they are +tracked independently. + +Revision ID: 0007_add_asset_semantics_version +Revises: 0006_add_loader_path +Create Date: 2026-08-18 +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0007_add_asset_semantics_version" +down_revision = "0006_add_loader_path" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "asset_semantics_version", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("version", sa.Integer(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=False), nullable=False), + sa.PrimaryKeyConstraint("id", name="pk_asset_semantics_version"), + ) + + +def downgrade() -> None: + op.drop_table("asset_semantics_version") diff --git a/app/assets/database/models.py b/app/assets/database/models.py index 329cd483d..df698a125 100644 --- a/app/assets/database/models.py +++ b/app/assets/database/models.py @@ -23,6 +23,28 @@ from app.assets.helpers import get_utc_now from app.database.models import Base +class AssetSemanticsVersion(Base): + """Which generation of the derivation logic produced this database's rows. + + Alembic tracks the *shape* of the assets tables; this tracks the *meaning* + of what is stored in them. A row can be structurally current and still hold + values a superseded rule computed, so the two versions move independently. + Reset steps in ``app.assets.semantics`` bring such rows forward and then + advance this stamp. Always a single row, keyed on id 1. + """ + + __tablename__ = "asset_semantics_version" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1) + version: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=False), nullable=False, default=get_utc_now + ) + + def __repr__(self) -> str: + return f"" + + class Asset(Base): __tablename__ = "assets" diff --git a/app/assets/database/queries/__init__.py b/app/assets/database/queries/__init__.py index 9949e84e1..50c4cfd10 100644 --- a/app/assets/database/queries/__init__.py +++ b/app/assets/database/queries/__init__.py @@ -52,6 +52,17 @@ from app.assets.database.queries.asset_reference import ( update_reference_updated_at, upsert_reference, ) +from app.assets.database.queries.semantics import ( + AUTOMATIC_TAG_ORIGIN, + DerivedStateRow, + bulk_add_automatic_tags, + bulk_remove_automatic_tags, + bulk_set_loader_paths, + get_file_backed_references_page, + get_semantics_version, + get_tags_by_reference, + set_semantics_version, +) from app.assets.database.queries.tags import ( AddTagsResult, RemoveTagsResult, @@ -71,7 +82,9 @@ from app.assets.database.queries.tags import ( __all__ = [ "AddTagsResult", + "AUTOMATIC_TAG_ORIGIN", "CacheStateRow", + "DerivedStateRow", "RemoveTagsResult", "SetTagsResult", "UnenrichedReferenceRow", @@ -81,6 +94,9 @@ __all__ = [ "bulk_insert_assets", "bulk_insert_references_ignore_conflicts", "bulk_insert_tags_and_meta", + "bulk_add_automatic_tags", + "bulk_remove_automatic_tags", + "bulk_set_loader_paths", "bulk_update_enrichment_level", "count_active_siblings", "create_stub_asset", @@ -96,6 +112,7 @@ __all__ = [ "fetch_reference_asset_and_tags", "get_asset_by_hash", "get_existing_asset_ids", + "get_file_backed_references_page", "get_or_create_reference", "get_reference_by_file_path", "get_reference_by_id", @@ -104,6 +121,8 @@ __all__ = [ "get_reference_tags", "get_references_by_paths_and_asset_ids", "get_references_for_prefixes", + "get_semantics_version", + "get_tags_by_reference", "get_unenriched_references", "get_unreferenced_unhashed_asset_ids", "insert_reference", @@ -123,6 +142,7 @@ __all__ = [ "set_reference_metadata", "set_reference_preview", "set_reference_system_metadata", + "set_semantics_version", "soft_delete_reference_by_id", "set_reference_tags", "update_asset_hash_and_mime", diff --git a/app/assets/database/queries/semantics.py b/app/assets/database/queries/semantics.py new file mode 100644 index 000000000..ff035114f --- /dev/null +++ b/app/assets/database/queries/semantics.py @@ -0,0 +1,195 @@ +"""Queries backing the asset semantics reset (see ``app.assets.semantics``).""" + +from dataclasses import dataclass + +import sqlalchemy as sa +from sqlalchemy.dialects import sqlite +from sqlalchemy.orm import Session + +from app.assets.database.models import ( + Asset, + AssetReference, + AssetReferenceTag, + AssetSemanticsVersion, +) +from app.assets.database.queries.common import ( + MAX_BIND_PARAMS, + iter_chunks, + iter_row_chunks, +) +from app.assets.database.queries.tags import ensure_tags_exist +from app.assets.helpers import get_utc_now + +# The version table holds exactly one row; the id is a constant, not a sequence. +_VERSION_ROW_ID = 1 + +# Tags this subsystem derives from the filesystem rather than being told. +AUTOMATIC_TAG_ORIGIN = "automatic" + + +def get_semantics_version(session: Session) -> int: + """Return the stored semantics version, or 0 if none has been stamped.""" + row = session.get(AssetSemanticsVersion, _VERSION_ROW_ID) + return int(row.version) if row is not None else 0 + + +def set_semantics_version(session: Session, version: int) -> None: + """Stamp the semantics version. Caller commits.""" + row = session.get(AssetSemanticsVersion, _VERSION_ROW_ID) + if row is None: + session.add( + AssetSemanticsVersion( + id=_VERSION_ROW_ID, version=int(version), updated_at=get_utc_now() + ) + ) + else: + row.version = int(version) + row.updated_at = get_utc_now() + session.flush() + + +@dataclass(frozen=True) +class DerivedStateRow: + """A file-backed reference and the state a reprojection may rewrite.""" + + reference_id: str + file_path: str + loader_path: str | None + mtime_ns: int | None + is_missing: bool + needs_verify: bool + size_bytes: int | None + + +def get_file_backed_references_page( + session: Session, + after_id: str | None, + limit: int, +) -> list[DerivedStateRow]: + """Return up to ``limit`` file-backed references ordered by id, after ``after_id``. + + Keyset pagination on the primary key so a reprojection can walk the whole + table in bounded transactions without holding one open across the walk. + Soft-deleted references are included: their derived columns are as stale as + anyone else's, and an undelete would serve them. + """ + query = ( + sa.select( + AssetReference.id, + AssetReference.file_path, + AssetReference.loader_path, + AssetReference.mtime_ns, + AssetReference.is_missing, + AssetReference.needs_verify, + Asset.size_bytes, + ) + .join(Asset, Asset.id == AssetReference.asset_id) + .where(AssetReference.file_path.isnot(None)) + .order_by(AssetReference.id.asc()) + .limit(limit) + ) + if after_id is not None: + query = query.where(AssetReference.id > after_id) + + return [ + DerivedStateRow( + reference_id=row[0], + file_path=row[1], + loader_path=row[2], + mtime_ns=row[3], + is_missing=bool(row[4]), + needs_verify=bool(row[5]), + size_bytes=int(row[6]) if row[6] is not None else None, + ) + for row in session.execute(query).all() + ] + + +def get_tags_by_reference( + session: Session, reference_ids: list[str] +) -> dict[str, dict[str, str]]: + """Return {reference_id: {tag_name: origin}} for the given references.""" + if not reference_ids: + return {} + + by_reference: dict[str, dict[str, str]] = {} + for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS): + rows = session.execute( + sa.select( + AssetReferenceTag.asset_reference_id, + AssetReferenceTag.tag_name, + AssetReferenceTag.origin, + ).where(AssetReferenceTag.asset_reference_id.in_(chunk)) + ).all() + for reference_id, tag_name, origin in rows: + by_reference.setdefault(reference_id, {})[tag_name] = origin + return by_reference + + +def bulk_set_loader_paths( + session: Session, loader_paths: dict[str, str | None] +) -> None: + """Set loader_path on the given references. Caller commits.""" + if not loader_paths: + return + session.execute( + sa.update(AssetReference), + [ + {"id": reference_id, "loader_path": loader_path} + for reference_id, loader_path in loader_paths.items() + ], + ) + + +def bulk_add_automatic_tags(session: Session, links: list[tuple[str, str]]) -> None: + """Attach (reference_id, tag_name) links with automatic origin. Caller commits.""" + if not links: + return + + now = get_utc_now() + ensure_tags_exist(session, {tag_name for _, tag_name in links}) + rows = [ + { + "asset_reference_id": reference_id, + "tag_name": tag_name, + "origin": AUTOMATIC_TAG_ORIGIN, + "added_at": now, + } + for reference_id, tag_name in links + ] + for chunk in iter_row_chunks(rows, cols_per_row=4): + session.execute( + sqlite.insert(AssetReferenceTag) + .values(chunk) + .on_conflict_do_nothing( + index_elements=[ + AssetReferenceTag.asset_reference_id, + AssetReferenceTag.tag_name, + ] + ) + ) + + +def bulk_remove_automatic_tags(session: Session, links: list[tuple[str, str]]) -> None: + """Detach (reference_id, tag_name) links, but only automatic ones. + + The origin is re-asserted in the statement rather than trusted from the + caller's snapshot, so a manual tag can never be removed by this path. + Caller commits. + """ + if not links: + return + + by_reference: dict[str, list[str]] = {} + for reference_id, tag_name in links: + by_reference.setdefault(reference_id, []).append(tag_name) + + for reference_id, tag_names in by_reference.items(): + for chunk in iter_chunks(tag_names, MAX_BIND_PARAMS): + session.execute( + sa.delete(AssetReferenceTag).where( + AssetReferenceTag.asset_reference_id == reference_id, + AssetReferenceTag.tag_name.in_(chunk), + AssetReferenceTag.origin == AUTOMATIC_TAG_ORIGIN, + ) + ) diff --git a/app/assets/seeder.py b/app/assets/seeder.py index 134fc98a8..977204a26 100644 --- a/app/assets/seeder.py +++ b/app/assets/seeder.py @@ -23,6 +23,7 @@ from app.assets.scanner import ( sync_root_safely, sync_temp_references_safely, ) +from app.assets.semantics import run_pending_semantics_steps from app.database.db import dependencies_available @@ -548,6 +549,11 @@ class _AssetSeeder: ) return + # Stale rows are brought forward before anything reads or extends + # them. A database already at the current semantics version costs a + # single row read here. + run_pending_semantics_steps(interrupt_check=self._is_cancelled) + if self._prune_first: all_prefixes = get_owned_prefixes() marked = mark_missing_outside_prefixes_safely(all_prefixes) diff --git a/app/assets/semantics/__init__.py b/app/assets/semantics/__init__.py new file mode 100644 index 000000000..7b2c850f1 --- /dev/null +++ b/app/assets/semantics/__init__.py @@ -0,0 +1,119 @@ +"""Reset steps that bring stale asset rows forward to the current semantics. + +Alembic migrates the *shape* of the assets tables. Nothing migrates their +*meaning*: a row whose columns are structurally current can still hold values +that a superseded rule computed, and the steady-state scan will not repair one, +because it only ever looks at paths it has not seen before. + +A reset step closes that gap. Each step is numbered, applied in order from the +version stamped in the database up to ``CURRENT_SEMANTICS_VERSION``, and stamped +only once it finishes, so an interrupted run resumes rather than half-applying. +Steps must be idempotent: re-running one on rows it has already fixed changes +nothing. + +The registry, not any individual step, is the durable part. A step is free to be +as broad or as surgical as its particular drift calls for. +""" + +import logging +import time + +from app.assets.database.queries.semantics import ( + get_semantics_version, + set_semantics_version, +) +from app.assets.semantics.reproject_derived import reproject_derived_state +from app.assets.semantics.step import ( + InterruptCheck, + SemanticsStep, + SemanticsStepInterrupted, +) +from app.database.db import can_create_session, create_session + +__all__ = [ + "CURRENT_SEMANTICS_VERSION", + "SEMANTICS_STEPS", + "InterruptCheck", + "SemanticsStep", + "SemanticsStepInterrupted", + "run_pending_semantics_steps", +] + +SEMANTICS_STEPS: tuple[SemanticsStep, ...] = ( + SemanticsStep( + version=1, + description="reproject path-derived reference state", + apply=reproject_derived_state, + ), +) + +CURRENT_SEMANTICS_VERSION = SEMANTICS_STEPS[-1].version + + +def run_pending_semantics_steps(interrupt_check: InterruptCheck | None = None) -> int: + """Apply every reset step this database has not been stamped for. + + A database already at ``CURRENT_SEMANTICS_VERSION`` costs one indexed row + read and nothing else -- no filesystem is touched to discover there is + nothing to do, which is the overwhelmingly common case. + + Returns the number of steps applied. + """ + if not can_create_session(): + return 0 + + try: + with create_session() as session: + stored_version = get_semantics_version(session) + except Exception: + logging.exception( + "Could not read the asset semantics version; skipping semantics reset" + ) + return 0 + + pending = [step for step in SEMANTICS_STEPS if step.version > stored_version] + if not pending: + return 0 + + logging.info( + "Asset semantics at version %d, current is %d: applying %d step(s)", + stored_version, + CURRENT_SEMANTICS_VERSION, + len(pending), + ) + + applied = 0 + for step in pending: + started = time.perf_counter() + try: + summary = step.apply(interrupt_check) + except SemanticsStepInterrupted: + logging.info( + "Asset semantics step %d (%s) interrupted; resuming on next start", + step.version, + step.description, + ) + return applied + except Exception: + logging.exception( + "Asset semantics step %d (%s) failed; database stays at version %d", + step.version, + step.description, + stored_version + applied, + ) + return applied + + with create_session() as session: + set_semantics_version(session, step.version) + session.commit() + + applied += 1 + logging.info( + "Asset semantics step %d (%s) applied in %.3fs: %s", + step.version, + step.description, + time.perf_counter() - started, + summary, + ) + + return applied diff --git a/app/assets/semantics/reproject_derived.py b/app/assets/semantics/reproject_derived.py new file mode 100644 index 000000000..177309118 --- /dev/null +++ b/app/assets/semantics/reproject_derived.py @@ -0,0 +1,215 @@ +"""Semantics step 1: re-derive the reference state that comes from the path. + +Three columns are pure functions of a file's location and whether it is there: +``loader_path``, the backend tags a path implies, and ``is_missing``. Every one +of them was computed once, when the reference was first written, by whatever +rules were current that day. ``loader_path`` in particular was added to existing +databases as a NULL column and no path has ever filled it in, so references +older than that column serve a null loader path forever. + +This step recomputes those three from the filesystem as it stands now. It reads +no file contents: ``verify_file_unchanged`` answers whether a file still matches +what the database recorded, and the only thing that answer is used for is +deciding *not* to touch content-derived state. Nothing here re-hashes, so a +library of untouched 500GB models costs one ``stat`` each. + +What the step will not touch: + +- ``hash`` and ``size_bytes`` -- content facts, unrecoverable without reading + the file. A file that has changed underneath its row is handed to the existing + ``needs_verify`` path instead, exactly as the scanner would flag it. +- Anything a person chose: manual and upload-origin tags, ``user_metadata``, + ``preview_id``, ``deleted_at``, ``job_id``, ``name``. +- References whose file is gone, or whose path is not under any root this + install currently knows about. Retiring the former is the scanner's job under + its own rules; the latter cannot be classified at all, and a misconfigured + ``extra_model_paths.yaml`` must not be able to strip tags off assets that are + merely out of view. +""" + +import logging +import os +from dataclasses import dataclass + +from app.assets.database.queries import ( + bulk_update_is_missing, + bulk_update_needs_verify, +) +from app.assets.database.queries.semantics import ( + AUTOMATIC_TAG_ORIGIN, + DerivedStateRow, + bulk_add_automatic_tags, + bulk_remove_automatic_tags, + bulk_set_loader_paths, + get_file_backed_references_page, + get_tags_by_reference, +) +from app.assets.helpers import normalize_tags +from app.assets.semantics.step import InterruptCheck, SemanticsStepInterrupted +from app.assets.services.file_utils import verify_file_unchanged +from app.assets.services.path_utils import ( + compute_loader_path, + get_path_derived_tag_vocabulary, + get_path_derived_tags_from_path, +) +from app.database.db import create_session + +# Bounds each transaction. Small enough that an interrupt loses little work, +# large enough that the walk is not dominated by session setup. +_BATCH_SIZE = 500 + + +@dataclass +class ReprojectionSummary: + """What a reprojection pass did, for the log line.""" + + scanned: int = 0 + unchanged_files: int = 0 + changed_files: int = 0 + absent_files: int = 0 + unclassified_paths: int = 0 + loader_paths_rewritten: int = 0 + tags_added: int = 0 + tags_removed: int = 0 + missing_flags_cleared: int = 0 + verify_flags_set: int = 0 + + def __str__(self) -> str: + return ( + f"scanned={self.scanned} unchanged={self.unchanged_files} " + f"changed={self.changed_files} absent={self.absent_files} " + f"unclassified={self.unclassified_paths} " + f"loader_paths={self.loader_paths_rewritten} " + f"tags+{self.tags_added}/-{self.tags_removed} " + f"unflagged_missing={self.missing_flags_cleared} " + f"flagged_verify={self.verify_flags_set}" + ) + + +def reproject_derived_state( + interrupt_check: InterruptCheck | None = None, +) -> ReprojectionSummary: + """Re-derive path-derived state for every file-backed reference. + + Walks the reference table in keyset-paginated batches, committing each one, + so a kill mid-walk leaves committed batches reprojected and the rest + untouched -- both states the step handles on its next run. + """ + summary = ReprojectionSummary() + vocabulary = get_path_derived_tag_vocabulary() + after_id: str | None = None + + while True: + if interrupt_check is not None and interrupt_check(): + raise SemanticsStepInterrupted( + f"interrupted after {summary.scanned} references" + ) + + with create_session() as session: + rows = get_file_backed_references_page( + session, after_id=after_id, limit=_BATCH_SIZE + ) + if not rows: + return summary + after_id = rows[-1].reference_id + _reproject_batch(session, rows, vocabulary, summary) + session.commit() + + +def _reproject_batch( + session, + rows: list[DerivedStateRow], + vocabulary: set[str], + summary: ReprojectionSummary, +) -> None: + stored_tags = get_tags_by_reference(session, [row.reference_id for row in rows]) + + loader_paths: dict[str, str | None] = {} + tags_to_add: list[tuple[str, str]] = [] + tags_to_remove: list[tuple[str, str]] = [] + clear_missing: list[str] = [] + set_needs_verify: list[str] = [] + + for row in rows: + summary.scanned += 1 + present, unchanged = _classify_file(row) + + if not present: + # The scanner owns retiring these, under its own missing semantics. + summary.absent_files += 1 + continue + + if unchanged: + summary.unchanged_files += 1 + else: + # The file moved on from what the row records, so its hash, size and + # mime no longer describe it. Re-deriving those means reading the + # file, which this step does not do; flag it for the path that does. + summary.changed_files += 1 + if not row.needs_verify: + set_needs_verify.append(row.reference_id) + + try: + derived_tags = set( + normalize_tags(get_path_derived_tags_from_path(row.file_path)) + ) + except ValueError: + summary.unclassified_paths += 1 + continue + + loader_path = compute_loader_path(row.file_path) + if loader_path != row.loader_path: + loader_paths[row.reference_id] = loader_path + summary.loader_paths_rewritten += 1 + + current_tags = stored_tags.get(row.reference_id, {}) + for tag_name in sorted(derived_tags - set(current_tags)): + tags_to_add.append((row.reference_id, tag_name)) + for tag_name, origin in sorted(current_tags.items()): + if ( + origin == AUTOMATIC_TAG_ORIGIN + and tag_name in vocabulary + and tag_name not in derived_tags + ): + tags_to_remove.append((row.reference_id, tag_name)) + + if row.is_missing: + # The file is right there. Nothing else un-flags this: the scanner + # excludes already-missing references from its temp reconciliation. + clear_missing.append(row.reference_id) + + bulk_set_loader_paths(session, loader_paths) + bulk_add_automatic_tags(session, tags_to_add) + bulk_remove_automatic_tags(session, tags_to_remove) + bulk_update_is_missing(session, clear_missing, value=False) + bulk_update_needs_verify(session, set_needs_verify, value=True) + + summary.tags_added += len(tags_to_add) + summary.tags_removed += len(tags_to_remove) + summary.missing_flags_cleared += len(clear_missing) + summary.verify_flags_set += len(set_needs_verify) + + +def _classify_file(row: DerivedStateRow) -> tuple[bool, bool]: + """Return (file is present, file still matches what the row recorded). + + Mirrors the scanner's stat handling so the two agree about what counts as a + present file: a permission error means the file is there but unreadable, any + other OS error means treat it as gone. + """ + try: + stat_result = os.stat(row.file_path, follow_symlinks=True) + except FileNotFoundError: + return False, False + except PermissionError: + logging.debug("Permission denied accessing %s", row.file_path) + return True, False + except OSError as error: + logging.debug("OSError checking %s: %s", row.file_path, error) + return False, False + + return True, verify_file_unchanged( + mtime_db=row.mtime_ns, + size_db=row.size_bytes, + stat_result=stat_result, + ) diff --git a/app/assets/semantics/step.py b/app/assets/semantics/step.py new file mode 100644 index 000000000..a3bf0fe03 --- /dev/null +++ b/app/assets/semantics/step.py @@ -0,0 +1,28 @@ +"""What a semantics reset step is. See ``app.assets.semantics`` for the why.""" + +from dataclasses import dataclass +from typing import Callable + +InterruptCheck = Callable[[], bool] + + +class SemanticsStepInterrupted(Exception): + """A step stopped early on request. + + Its partial work stands -- steps are idempotent, so a resumed run repeats it + harmlessly -- but the version is not stamped, so the step runs again. + """ + + +@dataclass(frozen=True) +class SemanticsStep: + """One numbered reset step. + + ``apply`` takes an optional interrupt check and returns a summary of what it + did, which is logged. It must be idempotent and must raise + ``SemanticsStepInterrupted`` rather than return if it stops early. + """ + + version: int + description: str + apply: Callable[[InterruptCheck | None], object] diff --git a/app/assets/services/path_utils.py b/app/assets/services/path_utils.py index 7c27c8878..7e46e6a03 100644 --- a/app/assets/services/path_utils.py +++ b/app/assets/services/path_utils.py @@ -325,6 +325,21 @@ def get_path_derived_tags_from_path(path: str) -> list[str]: return tags +def get_path_derived_tag_vocabulary() -> set[str]: + """Every tag get_path_derived_tags_from_path can emit under the current config. + + Bounds which tags a re-derivation may take away: a stored automatic tag + inside this vocabulary that the current rules no longer emit was produced + by a superseded rule, while one outside it came from somewhere else and is + none of the derivation's business. + """ + vocabulary = {"input", "output", "temp", "models"} + vocabulary.update(_KNOWN_SUBFOLDER_TAGS) + for folder_name, _bases, _extensions in get_comfy_models_folders(): + vocabulary.add(f"model_type:{folder_name}") + return vocabulary + + def get_name_and_tags_from_asset_path(file_path: str) -> tuple[str, list[str]]: """Return (name, tags) derived from a filesystem path. diff --git a/tests-unit/assets_test/test_semantics_reset.py b/tests-unit/assets_test/test_semantics_reset.py new file mode 100644 index 000000000..0eb6f3d1c --- /dev/null +++ b/tests-unit/assets_test/test_semantics_reset.py @@ -0,0 +1,672 @@ +"""Tests for the asset semantics reset (app/assets/semantics). + +Runs standalone against in-memory SQLite: + + pytest tests-unit/assets_test/test_semantics_reset.py --noconftest +""" + +import os +import tempfile +from datetime import datetime +from pathlib import Path +from unittest.mock import patch + +import pytest +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool + +import app.assets.semantics as semantics +from app.assets.database.models import ( + Asset, + AssetReference, + AssetReferenceTag, + AssetSemanticsVersion, + Base, + Tag, +) +from app.assets.database.queries.semantics import get_semantics_version +from app.assets.semantics import run_pending_semantics_steps +from app.assets.semantics.reproject_derived import reproject_derived_state +from app.assets.semantics.step import SemanticsStep, SemanticsStepInterrupted +from app.assets.services.file_utils import get_mtime_ns + + +@pytest.fixture(autouse=True) +def autoclean_unit_test_assets(): + """Override the package autouse fixture; these tests need no server.""" + yield + + +@pytest.fixture +def session_factory(): + """A session factory over one shared in-memory database. + + StaticPool keeps every session on the same connection, so the batching walk + sees its own committed writes across sessions. + """ + engine = create_engine( + "sqlite:///:memory:", + poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + Base.metadata.create_all(engine) + factory = sessionmaker(bind=engine) + with ( + patch("app.assets.semantics.reproject_derived.create_session", factory), + patch("app.assets.semantics.create_session", factory), + patch("app.assets.semantics.can_create_session", return_value=True), + ): + yield factory + + +@pytest.fixture +def session(session_factory) -> Session: + """A session for the test's own reads and writes.""" + with session_factory() as sess: + yield sess + + +@pytest.fixture +def comfy_dirs(): + """Point every asset root at a throwaway tree with one model category.""" + with tempfile.TemporaryDirectory() as base: + dirs = { + name: Path(base) / name + for name in ("checkpoints", "loras", "input", "output", "temp", "elsewhere") + } + for directory in dirs.values(): + directory.mkdir() + with ( + patch("folder_paths.get_input_directory", return_value=str(dirs["input"])), + patch( + "folder_paths.get_output_directory", return_value=str(dirs["output"]) + ), + patch("folder_paths.get_temp_directory", return_value=str(dirs["temp"])), + patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[ + ("checkpoints", [str(dirs["checkpoints"])], {".safetensors"}), + ("loras", [str(dirs["loras"])], {".safetensors"}), + ], + ), + ): + yield dirs + + +def _write(directory: Path, name: str, content: bytes = b"\x00" * 100) -> str: + path = directory / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + return os.path.abspath(str(path)) + + +def _register( + session: Session, + file_path: str, + ref_id: str, + *, + loader_path: str | None = None, + asset_hash: str | None = "", + size_bytes: int = 100, + mtime_ns: int | None = None, + is_missing: bool = False, + needs_verify: bool = False, + user_metadata: dict | None = None, + preview_id: str | None = None, + deleted_at: datetime | None = None, + job_id: str | None = None, + tags: dict[str, str] | None = None, +) -> AssetReference: + """Insert an Asset + AssetReference (+ tags as {name: origin}) and commit. + + ``asset_hash=""`` means "any unique hash"; assets.hash is unique-indexed. + """ + if asset_hash == "": + asset_hash = f"blake3:{ref_id}" + if mtime_ns is None: + mtime_ns = get_mtime_ns(os.stat(file_path, follow_symlinks=True)) + + session.add(Asset(id=f"asset-{ref_id}", hash=asset_hash, size_bytes=size_bytes)) + session.flush() + ref = AssetReference( + id=ref_id, + asset_id=f"asset-{ref_id}", + name=os.path.basename(file_path), + owner_id="", + file_path=file_path, + loader_path=loader_path, + mtime_ns=mtime_ns, + is_missing=is_missing, + needs_verify=needs_verify, + user_metadata=user_metadata, + preview_id=preview_id, + deleted_at=deleted_at, + job_id=job_id, + ) + session.add(ref) + session.flush() + + for tag_name, origin in (tags or {}).items(): + session.merge(Tag(name=tag_name)) + session.add( + AssetReferenceTag( + asset_reference_id=ref_id, tag_name=tag_name, origin=origin + ) + ) + session.commit() + return ref + + +def _tags(session: Session, ref_id: str) -> dict[str, str]: + rows = session.execute( + select(AssetReferenceTag.tag_name, AssetReferenceTag.origin).where( + AssetReferenceTag.asset_reference_id == ref_id + ) + ).all() + return {name: origin for name, origin in rows} + + +def _snapshot(session: Session) -> list[tuple]: + """Every field the reset could plausibly touch, for equality across runs.""" + session.expire_all() + refs = session.execute(select(AssetReference).order_by(AssetReference.id)).scalars() + rows = [ + ( + ref.id, + ref.file_path, + ref.loader_path, + ref.is_missing, + ref.needs_verify, + ref.mtime_ns, + ref.name, + ref.preview_id, + str(ref.user_metadata), + ref.deleted_at, + ref.job_id, + tuple(sorted(_tags(session, ref.id).items())), + ) + for ref in refs + ] + assets = session.execute(select(Asset).order_by(Asset.id)).scalars() + rows.extend((asset.id, asset.hash, asset.size_bytes) for asset in assets) + return rows + + +class TestLoaderPathReprojection: + def test_null_loader_path_is_backfilled(self, session, comfy_dirs): + """The drift 0006 left behind: a column added with no backfill.""" + path = _write(comfy_dirs["checkpoints"], "flux/model.safetensors") + _register(session, path, "ref-1", loader_path=None) + + reproject_derived_state() + + session.expire_all() + ref = session.get(AssetReference, "ref-1") + assert ref.loader_path == "flux/model.safetensors" + + def test_stale_loader_path_is_rewritten(self, session, comfy_dirs): + path = _write(comfy_dirs["input"], "sub/photo.png") + _register(session, path, "ref-1", loader_path="checkpoints/sub/photo.png") + + summary = reproject_derived_state() + + session.expire_all() + assert session.get(AssetReference, "ref-1").loader_path == "sub/photo.png" + assert summary.loader_paths_rewritten == 1 + + def test_correct_loader_path_is_left_alone(self, session, comfy_dirs): + path = _write(comfy_dirs["loras"], "style.safetensors") + _register(session, path, "ref-1", loader_path="style.safetensors") + + summary = reproject_derived_state() + + assert summary.loader_paths_rewritten == 0 + + def test_unloadable_extension_loses_its_loader_path(self, session, comfy_dirs): + """The current rule gives no loader path to a file its category cannot load.""" + path = _write(comfy_dirs["checkpoints"], "notes.txt") + _register(session, path, "ref-1", loader_path="notes.txt") + + reproject_derived_state() + + session.expire_all() + assert session.get(AssetReference, "ref-1").loader_path is None + + def test_reference_without_file_path_is_skipped(self, session, comfy_dirs): + session.add(Asset(id="asset-api", hash="blake3:abc", size_bytes=10)) + session.flush() + session.add( + AssetReference( + id="ref-api", asset_id="asset-api", name="api.png", owner_id="" + ) + ) + session.commit() + + summary = reproject_derived_state() + + assert summary.scanned == 0 + session.expire_all() + assert session.get(AssetReference, "ref-api").loader_path is None + + +class TestHashPreservation: + def test_unchanged_file_keeps_its_hash_and_is_not_read(self, session, comfy_dirs): + path = _write(comfy_dirs["checkpoints"], "big.safetensors") + _register(session, path, "ref-1", asset_hash="blake3:original", size_bytes=100) + + with patch( + "app.assets.services.hashing.compute_blake3_hash", + side_effect=AssertionError("the reset must never re-hash"), + ): + summary = reproject_derived_state() + + session.expire_all() + assert session.get(Asset, "asset-ref-1").hash == "blake3:original" + assert summary.unchanged_files == 1 + assert summary.changed_files == 0 + + def test_changed_file_keeps_its_hash_and_is_flagged_for_verify( + self, session, comfy_dirs + ): + """A file that moved on is handed to the existing verify path, not re-read.""" + path = _write(comfy_dirs["checkpoints"], "big.safetensors") + _register( + session, + path, + "ref-1", + asset_hash="blake3:original", + mtime_ns=1, + size_bytes=100, + ) + + summary = reproject_derived_state() + + session.expire_all() + assert session.get(Asset, "asset-ref-1").hash == "blake3:original" + assert session.get(Asset, "asset-ref-1").size_bytes == 100 + assert session.get(AssetReference, "ref-1").needs_verify is True + assert summary.changed_files == 1 + + def test_changed_file_still_gets_its_path_state_reprojected( + self, session, comfy_dirs + ): + path = _write(comfy_dirs["loras"], "style.safetensors") + _register(session, path, "ref-1", loader_path=None, mtime_ns=1) + + reproject_derived_state() + + session.expire_all() + assert session.get(AssetReference, "ref-1").loader_path == "style.safetensors" + + +class TestIntentIsPreserved: + def test_manual_tags_metadata_preview_and_deletion_survive( + self, session, comfy_dirs + ): + path = _write(comfy_dirs["checkpoints"], "curated.safetensors") + other = _write(comfy_dirs["input"], "thumb.png") + _register(session, other, "ref-preview") + deleted = datetime(2026, 1, 2, 3, 4, 5) + _register( + session, + path, + "ref-1", + loader_path=None, + user_metadata={"note": "hand written", "filename": "kept.safetensors"}, + preview_id="ref-preview", + deleted_at=deleted, + job_id="job-42", + tags={"favourite": "manual", "uploaded": "upload"}, + ) + + reproject_derived_state() + + session.expire_all() + ref = session.get(AssetReference, "ref-1") + assert ref.user_metadata == { + "note": "hand written", + "filename": "kept.safetensors", + } + assert ref.preview_id == "ref-preview" + assert ref.deleted_at == deleted + assert ref.job_id == "job-42" + assert ref.name == "curated.safetensors" + tags = _tags(session, "ref-1") + assert tags["favourite"] == "manual" + assert tags["uploaded"] == "upload" + # ...while the derived state around them was still brought forward. + assert ref.loader_path == "curated.safetensors" + + def test_manual_tag_inside_the_derived_vocabulary_is_not_removed( + self, session, comfy_dirs + ): + """A person may tag an input file 'models'; that is their business.""" + path = _write(comfy_dirs["input"], "photo.png") + _register(session, path, "ref-1", tags={"models": "manual"}) + + reproject_derived_state() + + assert _tags(session, "ref-1")["models"] == "manual" + + +class TestTagReprojection: + def test_missing_derived_tags_are_added(self, session, comfy_dirs): + path = _write(comfy_dirs["checkpoints"], "model.safetensors") + _register(session, path, "ref-1", tags={}) + + reproject_derived_state() + + assert _tags(session, "ref-1") == { + "models": "automatic", + "model_type:checkpoints": "automatic", + } + + def test_superseded_automatic_tag_is_removed(self, session, comfy_dirs): + """An older rule tagged by directory alone; extensions now gate the tag.""" + path = _write(comfy_dirs["checkpoints"], "model.safetensors") + _register( + session, + path, + "ref-1", + tags={ + "models": "automatic", + "model_type:checkpoints": "automatic", + "model_type:loras": "automatic", + }, + ) + + reproject_derived_state() + + assert "model_type:loras" not in _tags(session, "ref-1") + assert "model_type:checkpoints" in _tags(session, "ref-1") + + def test_automatic_tag_outside_the_vocabulary_is_left_alone( + self, session, comfy_dirs + ): + """'missing' is automatic but not path-derived; the scanner owns it.""" + path = _write(comfy_dirs["checkpoints"], "model.safetensors") + _register(session, path, "ref-1", tags={"missing": "automatic"}) + + reproject_derived_state() + + assert "missing" in _tags(session, "ref-1") + + +class TestFileState: + def test_present_file_is_unflagged_as_missing(self, session, comfy_dirs): + path = _write(comfy_dirs["temp"], "preview.png") + _register(session, path, "ref-1", is_missing=True) + + summary = reproject_derived_state() + + session.expire_all() + assert session.get(AssetReference, "ref-1").is_missing is False + assert summary.missing_flags_cleared == 1 + + def test_absent_file_is_left_to_the_scanner(self, session, comfy_dirs): + path = os.path.join(str(comfy_dirs["checkpoints"]), "gone.safetensors") + _register( + session, + path, + "ref-1", + mtime_ns=1, + loader_path="stale/gone.safetensors", + is_missing=True, + ) + + summary = reproject_derived_state() + + session.expire_all() + ref = session.get(AssetReference, "ref-1") + assert ref.is_missing is True + assert ref.loader_path == "stale/gone.safetensors" + assert summary.absent_files == 1 + + def test_path_outside_every_known_root_is_left_alone(self, session, comfy_dirs): + """A root missing from the config must not strip tags off its assets.""" + path = _write(comfy_dirs["elsewhere"], "model.safetensors") + _register( + session, + path, + "ref-1", + loader_path="model.safetensors", + tags={"models": "automatic", "model_type:checkpoints": "automatic"}, + ) + + summary = reproject_derived_state() + + session.expire_all() + assert session.get(AssetReference, "ref-1").loader_path == "model.safetensors" + assert _tags(session, "ref-1") == { + "models": "automatic", + "model_type:checkpoints": "automatic", + } + assert summary.unclassified_paths == 1 + + +class TestIdempotence: + def test_empty_database_is_a_no_op(self, session, comfy_dirs): + summary = reproject_derived_state() + + assert summary.scanned == 0 + assert summary.loader_paths_rewritten == 0 + assert summary.tags_added == 0 + + def test_second_run_changes_nothing(self, session, comfy_dirs): + _register( + session, + _write(comfy_dirs["checkpoints"], "a/model.safetensors"), + "ref-1", + loader_path=None, + tags={"model_type:loras": "automatic", "favourite": "manual"}, + ) + _register( + session, + _write(comfy_dirs["input"], "pasted/clip.png"), + "ref-2", + loader_path="wrong.png", + is_missing=True, + ) + _register( + session, + os.path.join(str(comfy_dirs["output"]), "gone.png"), + "ref-3", + mtime_ns=1, + ) + + reproject_derived_state() + after_first = _snapshot(session) + + second = reproject_derived_state() + assert _snapshot(session) == after_first + assert second.loader_paths_rewritten == 0 + assert second.tags_added == 0 + assert second.tags_removed == 0 + assert second.missing_flags_cleared == 0 + + def test_statements_chunked_by_bind_param_limit_stay_correct( + self, session, comfy_dirs + ): + """The bind-param chunking must not drop or duplicate a tag.""" + for index in range(4): + _register( + session, + _write(comfy_dirs["checkpoints"], f"m{index}.safetensors"), + f"ref-{index}", + tags={"model_type:loras": "automatic"}, + ) + + with ( + patch("app.assets.database.queries.semantics.MAX_BIND_PARAMS", 2), + patch("app.assets.database.queries.common.MAX_BIND_PARAMS", 2), + ): + reproject_derived_state() + + for index in range(4): + assert _tags(session, f"ref-{index}") == { + "models": "automatic", + "model_type:checkpoints": "automatic", + } + + def test_walk_crosses_batch_boundaries(self, session, comfy_dirs): + for index in range(7): + _register( + session, + _write(comfy_dirs["checkpoints"], f"m{index:02d}.safetensors"), + f"ref-{index:02d}", + loader_path=None, + ) + + with patch("app.assets.semantics.reproject_derived._BATCH_SIZE", 2): + summary = reproject_derived_state() + + assert summary.scanned == 7 + session.expire_all() + assert all( + session.get(AssetReference, f"ref-{index:02d}").loader_path + == f"m{index:02d}.safetensors" + for index in range(7) + ) + + +class TestRunner: + def test_pending_step_runs_and_stamps(self, session, comfy_dirs): + _register( + session, + _write(comfy_dirs["checkpoints"], "model.safetensors"), + "ref-1", + loader_path=None, + ) + + assert run_pending_semantics_steps() == 1 + + session.expire_all() + assert get_semantics_version(session) == semantics.CURRENT_SEMANTICS_VERSION + assert session.get(AssetReference, "ref-1").loader_path == "model.safetensors" + + def test_stamped_database_does_not_walk_again(self, session, comfy_dirs): + run_pending_semantics_steps() + + with patch( + "app.assets.semantics.reproject_derived.get_file_backed_references_page", + side_effect=AssertionError("a stamped database must not be walked"), + ): + assert run_pending_semantics_steps() == 0 + + def test_stamp_is_not_advanced_when_a_step_raises(self, session, comfy_dirs): + def _explode(_interrupt_check): + raise RuntimeError("step failed") + + with patch.object( + semantics, + "SEMANTICS_STEPS", + (SemanticsStep(version=1, description="explodes", apply=_explode),), + ): + assert run_pending_semantics_steps() == 0 + + session.expire_all() + assert get_semantics_version(session) == 0 + assert session.get(AssetSemanticsVersion, 1) is None + + def test_stamp_is_not_advanced_when_a_step_is_interrupted( + self, session, comfy_dirs + ): + def _interrupted(_interrupt_check): + raise SemanticsStepInterrupted("stopped") + + with patch.object( + semantics, + "SEMANTICS_STEPS", + (SemanticsStep(version=1, description="interrupts", apply=_interrupted),), + ): + assert run_pending_semantics_steps() == 0 + + session.expire_all() + assert get_semantics_version(session) == 0 + + def test_earlier_steps_stay_stamped_when_a_later_one_fails( + self, session, comfy_dirs + ): + def _ok(_interrupt_check): + return "fine" + + def _explode(_interrupt_check): + raise RuntimeError("step failed") + + with patch.object( + semantics, + "SEMANTICS_STEPS", + ( + SemanticsStep(version=1, description="ok", apply=_ok), + SemanticsStep(version=2, description="explodes", apply=_explode), + ), + ): + assert run_pending_semantics_steps() == 1 + + session.expire_all() + assert get_semantics_version(session) == 1 + + def test_steps_below_the_stored_version_are_skipped(self, session, comfy_dirs): + applied: list[int] = [] + + def _record(version): + def _apply(_interrupt_check): + applied.append(version) + return version + + return _apply + + steps = ( + SemanticsStep(version=1, description="one", apply=_record(1)), + SemanticsStep(version=2, description="two", apply=_record(2)), + ) + with patch.object(semantics, "SEMANTICS_STEPS", steps[:1]): + run_pending_semantics_steps() + with patch.object(semantics, "SEMANTICS_STEPS", steps): + run_pending_semantics_steps() + + assert applied == [1, 2] + + def test_interrupted_walk_leaves_committed_work_and_resumes( + self, session, comfy_dirs + ): + for index in range(4): + _register( + session, + _write(comfy_dirs["checkpoints"], f"m{index}.safetensors"), + f"ref-{index}", + loader_path=None, + ) + + calls = {"n": 0} + + def _stop_after_first_batch() -> bool: + calls["n"] += 1 + return calls["n"] > 1 + + with ( + patch("app.assets.semantics.reproject_derived._BATCH_SIZE", 2), + patch.object( + semantics, + "SEMANTICS_STEPS", + ( + SemanticsStep( + version=1, + description="reproject", + apply=reproject_derived_state, + ), + ), + ), + ): + assert run_pending_semantics_steps(_stop_after_first_batch) == 0 + session.expire_all() + assert get_semantics_version(session) == 0 + assert session.get(AssetReference, "ref-0").loader_path is not None + assert session.get(AssetReference, "ref-3").loader_path is None + + assert run_pending_semantics_steps() == 1 + + session.expire_all() + assert get_semantics_version(session) == 1 + assert session.get(AssetReference, "ref-3").loader_path == "m3.safetensors" diff --git a/tests-unit/seeder_test/test_seeder.py b/tests-unit/seeder_test/test_seeder.py index 55f813585..566069d41 100644 --- a/tests-unit/seeder_test/test_seeder.py +++ b/tests-unit/seeder_test/test_seeder.py @@ -23,6 +23,7 @@ def mock_dependencies(): """Mock all external dependencies for isolated testing.""" with ( patch("app.assets.seeder.dependencies_available", return_value=True), + patch("app.assets.seeder.run_pending_semantics_steps", return_value=0), patch("app.assets.seeder.sync_root_safely", return_value=set()), patch("app.assets.seeder.collect_paths_for_roots", return_value=[]), patch("app.assets.seeder.build_asset_specs", return_value=([], set(), 0)), @@ -454,6 +455,7 @@ class TestSeederMarkMissing: with ( patch("app.assets.seeder.dependencies_available", return_value=True), + patch("app.assets.seeder.run_pending_semantics_steps", return_value=0), patch("app.assets.seeder.get_owned_prefixes", return_value=["/models"]), patch("app.assets.seeder.mark_missing_outside_prefixes_safely", side_effect=track_mark), patch("app.assets.seeder.sync_temp_references_safely"), @@ -475,6 +477,7 @@ class TestSeederMarkMissing: ): with ( patch("app.assets.seeder.dependencies_available", return_value=True), + patch("app.assets.seeder.run_pending_semantics_steps", return_value=0), patch("app.assets.seeder.get_owned_prefixes", return_value=["/models"]), patch("app.assets.seeder.mark_missing_outside_prefixes_safely", return_value=0), patch("app.assets.seeder.sync_temp_references_safely") as sync_temp, @@ -494,6 +497,73 @@ class TestSeederMarkMissing: ) +class TestSeederSemanticsReset: + """The scan brings stale rows forward before it reads or extends them.""" + + def test_semantics_reset_runs_before_any_scan_work( + self, fresh_seeder: _AssetSeeder + ): + call_order = [] + + with ( + patch("app.assets.seeder.dependencies_available", return_value=True), + patch( + "app.assets.seeder.run_pending_semantics_steps", + side_effect=lambda interrupt_check=None: call_order.append("reset"), + ), + patch("app.assets.seeder.get_owned_prefixes", return_value=["/models"]), + patch( + "app.assets.seeder.mark_missing_outside_prefixes_safely", + side_effect=lambda prefixes: call_order.append("prune") or 0, + ), + patch( + "app.assets.seeder.sync_temp_references_safely", + side_effect=lambda: call_order.append("sync_temp"), + ), + patch( + "app.assets.seeder.sync_root_safely", + side_effect=lambda root: call_order.append("sync") or set(), + ), + patch("app.assets.seeder.collect_paths_for_roots", return_value=[]), + patch("app.assets.seeder.build_asset_specs", return_value=([], set(), 0)), + patch("app.assets.seeder.insert_asset_specs", return_value=0), + patch("app.assets.seeder.get_unenriched_assets_for_roots", return_value=[]), + patch("app.assets.seeder.enrich_assets_batch", return_value=(0, 0)), + ): + fresh_seeder.start(roots=("models",), prune_first=True) + fresh_seeder.wait(timeout=5.0) + + assert call_order[0] == "reset", ( + "reprojection must finish before the scan reads or extends those rows" + ) + + def test_semantics_reset_can_be_cancelled(self, fresh_seeder: _AssetSeeder): + captured = {} + + with ( + patch("app.assets.seeder.dependencies_available", return_value=True), + patch( + "app.assets.seeder.run_pending_semantics_steps", + side_effect=lambda interrupt_check=None: captured.update( + check=interrupt_check + ), + ), + patch("app.assets.seeder.sync_root_safely", return_value=set()), + patch("app.assets.seeder.collect_paths_for_roots", return_value=[]), + patch("app.assets.seeder.build_asset_specs", return_value=([], set(), 0)), + patch("app.assets.seeder.insert_asset_specs", return_value=0), + patch("app.assets.seeder.get_unenriched_assets_for_roots", return_value=[]), + patch("app.assets.seeder.enrich_assets_batch", return_value=(0, 0)), + ): + fresh_seeder.start(roots=("models",)) + fresh_seeder.wait(timeout=5.0) + + assert captured.get("check") is not None + assert captured["check"]() is False + fresh_seeder._cancel_event.set() + assert captured["check"]() is True + + class TestSeederPhases: """Test phased scanning behavior."""