From e7eb7832b9cd0c46167ff5b2fca88cb564d00ce8 Mon Sep 17 00:00:00 2001 From: Simon Pinfold Date: Tue, 18 Aug 2026 11:34:34 -0700 Subject: [PATCH] Do not stamp past references the reset could not classify A reference whose file is on disk but whose path is under no configured root was skipped so that a missing extra_model_paths.yaml could not strip tags off assets that were merely out of view. But the step still finished successfully and the version was still stamped, so restoring the root repaired nothing: the step never ran again, and no other path re-derives an existing reference's loader_path -- the scan only builds specs for paths it has not seen, and the enricher computes a loader path without writing it back. The guard turned destroying the data into never repairing it. A step now reports whether it finished its work, and an unfinished one is not stamped. A file that is simply gone still does not count as unfinished; that is the scan's business under its own missing semantics. Two ordering fixes in the scan around the same call: The walk is handed the seeder's pause-aware checkpoint rather than a cancel-only check. The seeder is paused while a prompt runs, so a check that ignores pause left the reset statting the whole asset table during generation while the status reported PAUSED. Cancellation is checked immediately after the reset instead of after the prune. Cancelling during a long reprojection otherwise fell through into the prune and temp reconciliation, doing state-mutating work on a scan that had been told to stop. --- app/assets/seeder.py | 10 ++- app/assets/semantics/__init__.py | 13 +++ app/assets/semantics/reproject_derived.py | 19 +++- app/assets/semantics/step.py | 11 ++- .../assets_test/test_semantics_reset.py | 80 ++++++++++++++++- tests-unit/seeder_test/test_seeder.py | 87 +++++++++++++++++++ 6 files changed, 213 insertions(+), 7 deletions(-) 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."""