diff --git a/app/assets/seeder.py b/app/assets/seeder.py index d45fb954f..88722cbda 100644 --- a/app/assets/seeder.py +++ b/app/assets/seeder.py @@ -550,7 +550,15 @@ class _AssetSeeder: return # Must precede the prune and the scan: both read and extend these rows. - run_pending_semantics_steps(interrupt_check=self._is_cancelled) + # _check_pause_and_cancel, not _is_cancelled: a pause has to suspend + # the walk between batches, or it keeps statting the whole table + # while the seeder reports PAUSED and a prompt runs. + run_pending_semantics_steps(interrupt_check=self._check_pause_and_cancel) + + if self._is_cancelled(): + logging.info("Asset scan cancelled during semantics reset") + cancelled = True + return if self._prune_first: all_prefixes = get_owned_prefixes() diff --git a/app/assets/semantics/__init__.py b/app/assets/semantics/__init__.py index a25ff819e..222ac4796 100644 --- a/app/assets/semantics/__init__.py +++ b/app/assets/semantics/__init__.py @@ -19,6 +19,7 @@ from app.assets.semantics.step import ( InterruptCheck, SemanticsStep, SemanticsStepInterrupted, + StepResult, ) from app.database.db import can_create_session, create_session @@ -28,6 +29,7 @@ __all__ = [ "InterruptCheck", "SemanticsStep", "SemanticsStepInterrupted", + "StepResult", "run_pending_semantics_steps", ] @@ -88,6 +90,17 @@ def run_pending_semantics_steps(interrupt_check: InterruptCheck | None = None) - ) return applied + if not summary.complete: + logging.info( + "Asset semantics step %d (%s) left work unfinished, staying at " + "version %d so it runs again: %s", + step.version, + step.description, + stored_version + applied, + summary, + ) + return applied + with create_session() as session: set_semantics_version(session, step.version) session.commit() diff --git a/app/assets/semantics/reproject_derived.py b/app/assets/semantics/reproject_derived.py index 671d120d1..dfa286e6f 100644 --- a/app/assets/semantics/reproject_derived.py +++ b/app/assets/semantics/reproject_derived.py @@ -32,7 +32,11 @@ from app.assets.database.queries.semantics import ( get_tag_origins_by_reference, ) from app.assets.helpers import normalize_tags -from app.assets.semantics.step import InterruptCheck, SemanticsStepInterrupted +from app.assets.semantics.step import ( + InterruptCheck, + SemanticsStepInterrupted, + StepResult, +) from app.assets.services.file_utils import verify_file_unchanged from app.assets.services.path_utils import ( compute_loader_path, @@ -45,7 +49,7 @@ _ROWS_PER_TRANSACTION = 500 @dataclass -class ReprojectionSummary: +class ReprojectionSummary(StepResult): scanned: int = 0 unchanged_files: int = 0 @@ -58,6 +62,17 @@ class ReprojectionSummary: missing_flags_cleared: int = 0 verify_flags_set: int = 0 + @property + def complete(self) -> bool: + """A path under no known root today may be under one tomorrow. + + Its file is on disk -- an absent file is counted separately -- so the + root was configured away rather than deleted, and nothing else will ever + re-derive the row: the scan only builds specs for paths it has not seen. + Staying unstamped is what lets restoring the root repair them. + """ + return self.unclassified_paths == 0 + def __str__(self) -> str: return ( f"scanned={self.scanned} unchanged={self.unchanged_files} " diff --git a/app/assets/semantics/step.py b/app/assets/semantics/step.py index 61d6a3208..a2ee542f3 100644 --- a/app/assets/semantics/step.py +++ b/app/assets/semantics/step.py @@ -5,6 +5,15 @@ from typing import Callable InterruptCheck = Callable[[], bool] +@dataclass +class StepResult: + """What a step did. A falsy ``complete`` withholds the stamp, so the step runs again.""" + + @property + def complete(self) -> bool: + return True + + class SemanticsStepInterrupted(Exception): """A step stopped early on request. @@ -19,4 +28,4 @@ class SemanticsStep: version: int description: str - apply: Callable[[InterruptCheck | None], object] + apply: Callable[[InterruptCheck | None], StepResult] diff --git a/tests-unit/assets_test/test_semantics_reset.py b/tests-unit/assets_test/test_semantics_reset.py index 04afe4b7f..503270fdc 100644 --- a/tests-unit/assets_test/test_semantics_reset.py +++ b/tests-unit/assets_test/test_semantics_reset.py @@ -23,7 +23,11 @@ from app.assets.database.models import ( 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.semantics.step import ( + SemanticsStep, + SemanticsStepInterrupted, + StepResult, +) from app.assets.services.file_utils import get_mtime_ns @@ -419,6 +423,9 @@ class TestFileState: assert ref.is_missing is True assert ref.loader_path == "stale/gone.safetensors" assert summary.absent_files == 1 + assert summary.complete, ( + "a file that is simply gone is the scanner's business, not unfinished work" + ) def test_path_outside_every_known_root_is_left_alone(self, session, comfy_dirs): path = _write(comfy_dirs["elsewhere"], "model.safetensors") @@ -439,6 +446,9 @@ class TestFileState: "model_type:checkpoints": "automatic", }, "a root missing from the config must not strip tags off its assets" assert summary.unclassified_paths == 1 + assert not summary.complete, ( + "the row is unrepaired, so the step has not finished its work" + ) class TestIdempotence: @@ -542,6 +552,70 @@ class TestRunner: assert get_semantics_version(session) == semantics.CURRENT_SEMANTICS_VERSION assert session.get(AssetReference, "ref-1").loader_path == "model.safetensors" + def test_unclassified_row_withholds_the_stamp(self, session, comfy_dirs): + """Skipping a row must not also stamp past it -- nothing else repairs one.""" + _register( + session, + _write(comfy_dirs["elsewhere"], "unconfigured.safetensors"), + "ref-outside", + loader_path=None, + ) + _register( + session, + _write(comfy_dirs["checkpoints"], "configured.safetensors"), + "ref-inside", + loader_path=None, + ) + + assert run_pending_semantics_steps() == 0 + + session.expire_all() + assert get_semantics_version(session) == 0, ( + "a pass that could not classify every row must run again" + ) + assert ( + session.get(AssetReference, "ref-inside").loader_path + == "configured.safetensors" + ), "the rows it could classify are still repaired" + + def test_restoring_a_root_repairs_the_rows_it_had_hidden( + self, session, comfy_dirs + ): + """The point of withholding the stamp: a re-added root gets reprojected.""" + hidden = _write(comfy_dirs["elsewhere"], "unconfigured.safetensors") + _register(session, hidden, "ref-outside", loader_path=None) + + run_pending_semantics_steps() + + with patch( + "app.assets.services.path_utils.get_comfy_models_folders", + return_value=[ + ("checkpoints", [str(comfy_dirs["checkpoints"])], {".safetensors"}), + ("loras", [str(comfy_dirs["loras"])], {".safetensors"}), + ("vae", [str(comfy_dirs["elsewhere"])], {".safetensors"}), + ], + ): + assert run_pending_semantics_steps() == 1 + + session.expire_all() + assert get_semantics_version(session) == 1 + assert ( + session.get(AssetReference, "ref-outside").loader_path + == "unconfigured.safetensors" + ) + + def test_fully_classified_pass_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) == 1 + def test_stamped_database_does_not_walk_again(self, session, comfy_dirs): run_pending_semantics_steps() @@ -586,7 +660,7 @@ class TestRunner: self, session, comfy_dirs ): def _ok(_interrupt_check): - return "fine" + return StepResult() def _explode(_interrupt_check): raise RuntimeError("step failed") @@ -610,7 +684,7 @@ class TestRunner: def _record(version): def _apply(_interrupt_check): applied.append(version) - return version + return StepResult() return _apply diff --git a/tests-unit/seeder_test/test_seeder.py b/tests-unit/seeder_test/test_seeder.py index f6e49bfa4..a43c7c39b 100644 --- a/tests-unit/seeder_test/test_seeder.py +++ b/tests-unit/seeder_test/test_seeder.py @@ -563,6 +563,93 @@ class TestSeederSemanticsReset: assert captured["check"]() is True + def test_semantics_reset_suspends_on_pause(self, fresh_seeder: _AssetSeeder): + """A pause must stop the walk between batches, not just a cancel. + + The seeder is paused while a prompt runs, so a check that ignores pause + keeps statting the whole asset table during generation. + """ + 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) + + check = captured.get("check") + assert check is not None + + fresh_seeder._run_gate.clear() + returned = threading.Event() + + def _call_check(): + check() + returned.set() + + threading.Thread(target=_call_check, daemon=True).start() + assert not returned.wait(timeout=0.5), ( + "the interrupt check must block while paused, not report 'keep going'" + ) + + fresh_seeder._run_gate.set() + assert returned.wait(timeout=5.0) + + def test_cancel_during_semantics_reset_stops_before_pruning( + self, fresh_seeder: _AssetSeeder + ): + """A cancelled reset must not fall through into the prune's writes.""" + calls = [] + + def _cancel_during_reset(interrupt_check=None): + calls.append("reset") + fresh_seeder._cancel_event.set() + + with ( + patch("app.assets.seeder.dependencies_available", return_value=True), + patch( + "app.assets.seeder.run_pending_semantics_steps", + side_effect=_cancel_during_reset, + ), + patch("app.assets.seeder.get_owned_prefixes", return_value=["/models"]), + patch( + "app.assets.seeder.mark_missing_outside_prefixes_safely", + side_effect=lambda prefixes: calls.append("prune") or 0, + ), + patch( + "app.assets.seeder.sync_temp_references_safely", + side_effect=lambda: calls.append("sync_temp"), + ), + patch( + "app.assets.seeder.sync_root_safely", + side_effect=lambda root: calls.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 calls == ["reset"], ( + f"cancel must stop the scan before the prune's writes, got {calls}" + ) + + class TestSeederPhases: """Test phased scanning behavior."""