diff --git a/py/api/__init__.py b/py/api/__init__.py index fbebf5c..ad5fb7a 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -32,6 +32,7 @@ from .saved_views import SavedViewRoutesMixin from .history import HistoryRoutesMixin from .images import ImageRoutesMixin from .admin import AdminRoutesMixin +from .settings_routes import SettingsRoutesMixin from .job_routes import JobRoutesMixin from .image_tag_jobs import ImageTagJobsMixin from .jobs import JobRegistry @@ -261,6 +262,7 @@ class PromptManagerAPI( PromptRoutesMixin, ImageRoutesMixin, AdminRoutesMixin, + SettingsRoutesMixin, JobRoutesMixin, ImageTagJobsMixin, LoggingRoutesMixin, @@ -361,6 +363,24 @@ class PromptManagerAPI( """409 envelope for a long job that is already running.""" return _json_error(f"{label} already running", 409) + async def _run_held(self, kind, title, admission, operation): + """Run a legacy-page operation under a registry hold (D80). + + While ``operation()`` runs, the registry lists it as a running, + non-cancellable job, so workspace jobs its admission class conflicts + with are refused, naming it. When the registry refuses the hold, the + legacy 409 envelope names the running job and nothing is done. A + response of 400 or above ends the hold ``failed``. + """ + busy = self._jobs.blocker(kind, admission=admission) + if busy is not None: + return self._long_job_busy_response(busy.title) + async with self._jobs.hold(kind, title, admission=admission) as job: + response = await operation() + if getattr(response, "status", 200) >= 400: + job.finish("failed") + return response + async def _run_in_executor(self, func, *args, **kwargs): """Run a blocking function in the default thread pool executor. @@ -456,6 +476,7 @@ class PromptManagerAPI( self._register_prompt_routes(routes) self._register_image_routes(routes) self._register_admin_routes(routes) + self._register_settings_routes(routes) self._register_job_routes(routes) self._register_logging_routes(routes) self._register_autotag_routes(routes) diff --git a/py/api/admin.py b/py/api/admin.py index f4a6af9..7ad05da 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -22,16 +22,17 @@ from .jobs import ( SCAN_GROUP_NAMES, SCAN_GROUPS, SCAN_SCOPES, + DUPLICATE_CLEANUP_JOB_KIND, + MAINTENANCE_JOB_KIND, + RESTORE_JOB_KIND, Job, JobBusy, ) try: from ...utils.parallel import map_parallel - from ...utils.validators import validate_result_timeout, validate_worker_threads except ImportError: from utils.parallel import map_parallel - from utils.validators import validate_result_timeout, validate_worker_threads try: from ...database.operations import PromptDatabase @@ -502,14 +503,6 @@ class AdminRoutesMixin: async def get_stats_route(request): return await self.get_statistics(request) - @routes.get("/prompt_manager/settings") - async def get_settings_route(request): - return await self.get_settings(request) - - @routes.post("/prompt_manager/settings") - async def save_settings_route(request): - return await self.save_settings(request) - @routes.get("/prompt_manager/backup") async def backup_database_route(request): return await self.backup_database(request) @@ -651,7 +644,15 @@ class AdminRoutesMixin: job.finish() async def cleanup_duplicates_endpoint(self, request): - """Cleanup duplicate prompts endpoint.""" + """Cleanup duplicate prompts endpoint (holds the registry: records).""" + return await self._run_held( + MAINTENANCE_JOB_KIND, + "Prompt cleanup (admin page)", + "records", + self._cleanup_duplicates, + ) + + async def _cleanup_duplicates(self): try: removed_count = await self._run_in_executor(self.db.cleanup_duplicates) @@ -766,26 +767,39 @@ class AdminRoutesMixin: return hash_sha256.hexdigest() async def delete_duplicate_images_endpoint(self, request): - """Delete duplicate image files from disk (single flight with the scan).""" - job = self._acquire_long_job(self.LONG_JOB_DUPLICATES) - if job is None: + """Delete duplicate image files from disk. + + Single flight with the duplicate scan, and a records hold on the + registry, so no scan, restore or tagging job runs meanwhile. A body + without paths is refused before the hold (it is no job). + """ + data, error_response = await _read_json(request) + if error_response is not None: + return error_response + image_paths = data.get("image_paths", []) + if not image_paths or not isinstance(image_paths, list): + return web.json_response( + {"success": False, "error": "No image paths provided"}, + status=400, + ) + guard = self._acquire_long_job(self.LONG_JOB_DUPLICATES) + if guard is None: return self._long_job_busy_response("Duplicate scan or delete") - async with job: - return await self._delete_duplicate_images(request) - async def _delete_duplicate_images(self, request): + async def delete(): + # No await since the check above: the lock is still free. + async with guard: + return await self._delete_duplicate_images(image_paths) + + return await self._run_held( + DUPLICATE_CLEANUP_JOB_KIND, + "Duplicate image delete (gallery page)", + "records", + delete, + ) + + async def _delete_duplicate_images(self, image_paths): try: - data, error_response = await _read_json(request) - if error_response is not None: - return error_response - image_paths = data.get("image_paths", []) - - if not image_paths or not isinstance(image_paths, list): - return web.json_response( - {"success": False, "error": "No image paths provided"}, - status=400, - ) - output_dir = await self._run_in_executor(self._find_comfyui_output_dir) result = await self._run_in_executor( self._delete_duplicate_images_sync, image_paths, output_dir @@ -882,208 +896,6 @@ class AdminRoutesMixin: status=500, ) - async def get_settings(self, request): - """Get current settings.""" - try: - from ..config import PromptManagerConfig, GalleryConfig - - root_paths = [ - self._public_path(d) for d in GalleryConfig.MONITORING_DIRECTORIES - ] - return web.json_response( - { - "success": True, - "settings": { - "result_timeout": PromptManagerConfig.RESULT_TIMEOUT, - "webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE, - "infinite_scroll": PromptManagerConfig.INFINITE_SCROLL, - "worker_threads": PromptManagerConfig.WORKER_THREADS, - "cpu_count": PromptManagerConfig.max_worker_threads(), - "gallery_root_paths": root_paths, - "gallery_root_path": root_paths[0] if root_paths else "", - "monitored_directories": [ - self._public_path(d) for d in self._monitored_directories() - ], - }, - } - ) - except Exception as e: - self.logger.error(f"Get settings error: {e}", exc_info=True) - return web.json_response( - {"success": False, "error": "Failed to get settings"}, - status=500, - ) - - def _monitored_directories(self): - """Directories the running image monitor watches, else the configured roots.""" - from ..config import GalleryConfig - - monitor = _current_image_monitor() - if monitor is not None: - return list(getattr(monitor, "monitored_directories", [])) - return list(GalleryConfig.MONITORING_DIRECTORIES) - - async def save_settings(self, request): - """Save settings: the whole payload is validated before anything changes.""" - try: - from ..config import PromptManagerConfig - - data, error_response = await _read_json(request) - if error_response is not None: - return error_response - - try: - updates = self._validate_settings(data) - except ValueError as ve: - return web.json_response( - {"success": False, "error": str(ve)}, status=400 - ) - - new_roots, error = self._parse_gallery_roots(data) - if error: - return web.json_response({"success": False, "error": error}, status=400) - - for name, value in updates.items(): - setattr(PromptManagerConfig, name, value) - restart_required = False - if new_roots is not None: - restart_required = self._apply_gallery_roots(new_roots) - - self._persist_settings() - - return web.json_response( - { - "success": True, - "message": "Settings saved successfully", - "restart_required": restart_required, - } - ) - except Exception as e: - self.logger.error(f"Save settings error: {e}", exc_info=True) - return web.json_response( - {"success": False, "error": "Failed to save settings"}, - status=500, - ) - - @staticmethod - def _validate_settings(data): - """Scalar settings of a payload as ``{CONFIG_ATTR: value}``. - - Raises ValueError with a user-facing message on the first bad value; - nothing is applied until every value passed. - """ - from ..config import PromptManagerConfig - - updates = {} - if "result_timeout" in data: - updates["RESULT_TIMEOUT"] = validate_result_timeout(data["result_timeout"]) - if "webui_display_mode" in data: - mode = data["webui_display_mode"] - if mode not in ("popup", "newtab"): - raise ValueError("webui_display_mode must be 'popup' or 'newtab'") - updates["WEBUI_DISPLAY_MODE"] = mode - if "infinite_scroll" in data: - if not isinstance(data["infinite_scroll"], bool): - raise ValueError("infinite_scroll must be true or false") - updates["INFINITE_SCROLL"] = data["infinite_scroll"] - if "worker_threads" in data: - updates["WORKER_THREADS"] = validate_worker_threads( - data["worker_threads"], PromptManagerConfig.max_worker_threads() - ) - return updates - - def _parse_gallery_roots(self, data): - """Extract and validate gallery roots from a settings payload. - - Accepts ``gallery_root_paths`` (list, preferred) or the legacy single - ``gallery_root_path`` string. Blank entries are dropped. - - Returns: - (roots, error): ``roots`` is the validated list, or ``None`` when - the payload carries no gallery root key; ``error`` is a message - when validation failed (and ``roots`` is then ``None``). - """ - if "gallery_root_paths" in data: - raw = data["gallery_root_paths"] - if not isinstance(raw, list): - return None, "gallery_root_paths must be a list" - elif "gallery_root_path" in data: - raw = [data["gallery_root_path"]] - else: - return None, None - - from ..config import GalleryConfig - - roots = [] - for entry in raw: - if not isinstance(entry, str): - return None, "Gallery root paths must be strings" - entry = entry.strip() - if not entry: - continue - ok, reason = GalleryConfig.validate_gallery_root(entry) - if not ok: - return ( - None, - f"Invalid gallery root '{os.path.basename(entry)}': {reason}", - ) - roots.append(GalleryConfig.resolve_gallery_root(entry)) - return roots, None - - def _apply_gallery_roots(self, roots): - """Install validated gallery roots; returns True when they changed.""" - from ..config import GalleryConfig - - if roots == list(GalleryConfig.MONITORING_DIRECTORIES): - return False - GalleryConfig.MONITORING_DIRECTORIES = roots - self._cached_output_dir = None - self.reset_gallery_roots_cache() - return True - - def _persist_settings(self): - """Merge the user-editable settings into the configured config.json. - - Only the keys this endpoint owns are rewritten; every other section - (integrations, database, performance, ...) and every other key in - ``web_ui``/``gallery`` is kept as it was. A config file that cannot - be parsed is replaced. - """ - from ..config import GalleryConfig, PromptManagerConfig, write_private_json - - config_file = PromptManagerConfig.get_config_path() - existing = self._read_config_file(config_file) - web_ui = dict(existing.get("web_ui") or {}) - web_ui["result_timeout"] = PromptManagerConfig.RESULT_TIMEOUT - web_ui["webui_display_mode"] = PromptManagerConfig.WEBUI_DISPLAY_MODE - web_ui["infinite_scroll"] = PromptManagerConfig.INFINITE_SCROLL - gallery = dict(existing.get("gallery") or {}) - monitoring = dict(gallery.get("monitoring") or {}) - monitoring["directories"] = list(GalleryConfig.MONITORING_DIRECTORIES) - gallery["monitoring"] = monitoring - performance = dict(existing.get("performance") or {}) - performance["worker_threads"] = PromptManagerConfig.WORKER_THREADS - config_data = { - **existing, - "web_ui": web_ui, - "gallery": gallery, - "performance": performance, - } - try: - write_private_json(config_file, config_data) - self.logger.info(f"Settings saved to {config_file}") - except OSError as save_err: - self.logger.warning(f"Could not save config file: {save_err}") - - def _read_config_file(self, config_file): - """Current contents of config.json as a dict, or {} when unusable.""" - try: - with open(config_file, "r", encoding="utf-8") as f: - data = json.load(f) - except (OSError, ValueError): - return {} - return data if isinstance(data, dict) else {} - async def run_diagnostics(self, request): """Run comprehensive system diagnostics and health checks.""" try: @@ -1176,6 +988,11 @@ class AdminRoutesMixin: "vacuum", "cleanup_orphaned_images", ) + # Operations that only read; a run of only these holds the registry as + # "read", every other run as "records". + READ_ONLY_MAINTENANCE_OPERATIONS = frozenset( + {"check_hash_duplicates", "statistics", "check_consistency"} + ) async def run_maintenance(self, request): """Perform database maintenance operations and report each outcome.""" @@ -1188,7 +1005,27 @@ class AdminRoutesMixin: operations = data.get( "operations", list(self.DEFAULT_MAINTENANCE_OPERATIONS) ) + except Exception as e: + self.logger.error(f"Maintenance error: {e}", exc_info=True) + return web.json_response( + {"success": False, "error": "Maintenance failed"}, status=500 + ) + writes = not ( + isinstance(operations, list) + and all( + isinstance(name, str) and name in self.READ_ONLY_MAINTENANCE_OPERATIONS + for name in operations + ) + ) + return await self._run_held( + MAINTENANCE_JOB_KIND, + "Maintenance (admin page)", + "records" if writes else "read", + lambda: self._run_maintenance(operations), + ) + async def _run_maintenance(self, operations): + try: results = await self._run_in_executor( self._run_maintenance_operations, operations ) @@ -1342,8 +1179,18 @@ class AdminRoutesMixin: The upload is streamed to a temp file (capped at restore_max_bytes), verified with PromptModel.verify_database_file and only then copied - over the live database, which is backed up first. + over the live database, which is backed up first. The whole request + holds the registry (records): it is refused while a scan or another + record-changing job runs, and those are refused while it runs. """ + return await self._run_held( + RESTORE_JOB_KIND, + "Database restore (admin page)", + "records", + lambda: self._restore_database(request), + ) + + async def _restore_database(self, request): max_bytes = getattr(self, "restore_max_bytes", 100 * 1024 * 1024) temp_path = None try: diff --git a/py/api/autotag_routes.py b/py/api/autotag_routes.py index 999b941..f72db79 100644 --- a/py/api/autotag_routes.py +++ b/py/api/autotag_routes.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Optional from urllib.parse import quote +from .jobs import AUTOTAG_BATCH_JOB_KIND from .prompts import thumbnail_url_for from .. import model_gate @@ -63,6 +64,12 @@ def path_is_within(candidate, root) -> bool: return child != parent and child.is_relative_to(parent) +async def _refusal(title): + """The single error event of an auto-tag batch refused by another job.""" + message = f"{title} already running" + yield {"type": "error", "status": "error", "message": message} + + class AutotagRoutesMixin: """Mixin providing auto-tagging API endpoints.""" @@ -425,6 +432,8 @@ class AutotagRoutesMixin: "general_threshold and character_threshold must be numeric" ) + errors = [] # an error event ends the run; Activity lists it failed + async def events(): try: from ..autotag import get_autotag_service @@ -432,14 +441,40 @@ class AutotagRoutesMixin: async for event in self._autotag_batch_events( get_autotag_service(), opts ): + if event.get("type") == "error": + errors.append(event) yield event except Exception: self.logger.exception("AutoTag error") + errors.append(_INTERNAL_ERROR) yield { "type": "error", "message": _INTERNAL_ERROR, } + # The batch writes prompt tags: a records hold on the registry, so it + # never runs beside a scan, restore or the workspace tagging job. + busy = self._jobs.blocker(AUTOTAG_BATCH_JOB_KIND, admission="records") + if busy is not None: + if busy.kind == AUTOTAG_BATCH_JOB_KIND: # unchanged payload + return web.json_response( + {"success": False, "error": _BUSY_ERROR}, status=409 + ) + # Both legacy pages read this route as a stream, where a 409 reads + # as a lost connection: refuse with one error frame instead (the + # admin page keys on ``type``, the gallery on ``status``). + return await self._stream_sse(request, _refusal(busy.title)) + async with self._jobs.hold( + AUTOTAG_BATCH_JOB_KIND, + "Auto-tag images (legacy page)", + admission="records", + ) as job: + response = await self._start_autotag_batch(request, opts, events) + if response.status >= 400 or errors: + job.finish("failed") + return response + + async def _start_autotag_batch(self, request, opts, events): if not _autotag_batch_lock.acquire(blocking=False): return web.json_response( {"success": False, "error": _BUSY_ERROR}, status=409 diff --git a/py/api/images.py b/py/api/images.py index 6edfdc1..69ad96f 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -19,6 +19,7 @@ from typing import Callable, NamedTuple from aiohttp import web from PIL import Image, UnidentifiedImageError +from .jobs import THUMBNAILS_JOB_KIND from .prompts import bad_request, parse_page_params, publish_image_paths from .prompts import thumbnail_url_for from .prompts import safe_error_message as _safe_error @@ -104,6 +105,10 @@ def _card_header(path: Path) -> tuple: return inspection.width, inspection.height, records +# The legacy gallery page's thumbnail generation, as Activity lists it (D80). +THUMBNAIL_GENERATION_TITLE = "Thumbnail generation (gallery page)" + + class _CardFacts(NamedTuple): """One image file's card facts, derived from its header alone. @@ -1736,11 +1741,24 @@ class ImageRoutesMixin: return cls.THUMBNAIL_SIZES.get(quality, cls.THUMBNAIL_SIZES["medium"]) async def generate_thumbnails(self, request): - """Generate every thumbnail in one executor job (POST, blocking).""" + """Generate every thumbnail in one executor job (POST, blocking). + + Holds the registry as guarded work: refused while a record-changing + job runs, and such jobs are refused while it runs. A body that cannot + be read is refused before the hold, so it is never listed as a job. + """ + data, error = await _json_body_or_error(request) + if error is not None: + return error + return await self._run_held( + THUMBNAILS_JOB_KIND, + THUMBNAIL_GENERATION_TITLE, + "guarded", + lambda: self._generate_thumbnails(data), + ) + + async def _generate_thumbnails(self, data): try: - data, error = await _json_body_or_error(request) - if error is not None: - return error thumbnail_size = self._thumbnail_size(data.get("quality", "medium")) output_dirs = self._get_all_output_dirs() @@ -1935,14 +1953,32 @@ class ImageRoutesMixin: "Connection": "keep-alive", }, ) - await response.prepare(request) + busy = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded") + if busy is not None: + # The gallery reads this stream: refuse with its error event. + await response.prepare(request) + message = f"{busy.title} already running" + await self._emit_sse( + response, "error", {"error": message, "message": message} + ) + return response + async with self._jobs.hold( + THUMBNAILS_JOB_KIND, THUMBNAIL_GENERATION_TITLE, admission="guarded" + ) as job: + await response.prepare(request) + if not await self._thumbnail_progress_stream(response, thumbnail_size): + job.finish("failed") # an error event ended the stream + return response + + async def _thumbnail_progress_stream(self, response, thumbnail_size) -> bool: + """Stream one generation run; False when it ended with an error event.""" try: output_dirs = self._get_all_output_dirs() if not output_dirs: await self._emit_sse( response, "error", {"error": "ComfyUI output directory not found"} ) - return response + return False await self._emit_sse( response, @@ -1960,6 +1996,7 @@ class ImageRoutesMixin: remaining=plan["remaining"], ) self.invalidate_gallery_cache() + return True except Exception: self.logger.exception("Thumbnail generation failed") await self._emit_sse( @@ -1970,7 +2007,7 @@ class ImageRoutesMixin: "message": "Thumbnail generation failed. Check server logs.", }, ) - return response + return False async def _stream_thumbnails( self, response, targets, thumbnail_size, skipped=0, remaining=0 @@ -2179,8 +2216,17 @@ class ImageRoutesMixin: """Clear generated thumbnails across gallery roots, preserving originals. Invalid thumbnail directories return 400; unavailable gallery roots - return 404, and unexpected filesystem failures return 500. + return 404, and unexpected filesystem failures return 500. Holds the + registry as guarded work, like generation. """ + return await self._run_held( + THUMBNAILS_JOB_KIND, + "Thumbnail clear (gallery page)", + "guarded", + self._clear_thumbnails, + ) + + async def _clear_thumbnails(self) -> web.Response: try: output_dirs = self._get_all_output_dirs() if not output_dirs: diff --git a/py/api/jobs.py b/py/api/jobs.py index 27e4ecb..568a99e 100644 --- a/py/api/jobs.py +++ b/py/api/jobs.py @@ -9,9 +9,21 @@ as a snapshot and then every event published after it, until the job ends. Jobs live only in server memory and are gone after a restart. The registry keeps the running jobs and this session's last ``MAX_FINISHED_JOBS`` finished -ones. Record-changing jobs (the metadata scan, bulk image tagging, and later -maintenance, cleanup and restore) never run together, and one job of a kind -runs at a time. +ones. One job of a kind runs at a time, and each job has an admission class +(``ADMISSIONS``) that decides what may run beside it: + +- ``records`` changes library records or files (the metadata scan, bulk image + tagging, and the legacy restore, maintenance, cleanup, duplicate delete, + LoRA import and batch auto-tag routes). It is refused while any + ``records`` or ``guarded`` job runs. +- ``guarded`` must not overlap record changes but may overlap other guarded + work (thumbnail generation and clearing, later backup). It is refused while + a ``records`` job runs. +- ``read`` only reads; it is refused only by a running job of its own kind. + +A short legacy-page operation takes a ``hold``: a running, non-cancellable +job that lasts as long as the request, so the Activity page lists it and a +workspace job is refused while it runs (and the other way round). """ import asyncio @@ -60,6 +72,41 @@ SCAN_GROUPS = { IMAGE_TAG_JOB_KIND = "image-tags" IMAGE_TAG_GROUP_NAMES = ("tagged", "unchanged", "skipped", "errors") +ADMISSIONS = ("read", "guarded", "records") +# What each admission class is refused beside (besides its own kind). +_REFUSED_BESIDE = { + "read": frozenset(), + "guarded": frozenset({"records"}), + "records": frozenset({"records", "guarded"}), +} + +# Kinds of the legacy-page operations that hold the registry (D80; titles in +# their handlers end in "(admin page)", "(gallery page)" or "(legacy page)"). +RESTORE_JOB_KIND = "restore" +MAINTENANCE_JOB_KIND = "maintenance" +DUPLICATE_CLEANUP_JOB_KIND = "duplicate_cleanup" +THUMBNAILS_JOB_KIND = "thumbnails" +LORA_IMPORT_JOB_KIND = "lora_import" +AUTOTAG_BATCH_JOB_KIND = "autotag" + + +def resolve_admission(admission=None, record_changing=None) -> str: + """The admission class for a start call; ``record_changing`` is an alias. + + ``record_changing=True`` means ``records`` and ``False`` means ``read``; + with neither given the class is ``read``. Raises ValueError for an + unknown class or one that contradicts ``record_changing``. + """ + if admission is None: + return "records" if record_changing else "read" + if admission not in ADMISSIONS: + raise ValueError(f"Unknown job admission {admission!r}") + if record_changing is not None and record_changing != (admission == "records"): + raise ValueError( + f"record_changing={record_changing!r} contradicts admission {admission!r}" + ) + return admission + def _relative_path(path) -> str: """``path`` as a root-relative POSIX path; an absolute one keeps its name.""" @@ -85,13 +132,23 @@ class Job: """One tracked operation: state, counters, outcomes, log and subscribers.""" def __init__( - self, kind, title, *, record_changing=False, scope=None, groups=(), clock=None + self, + kind, + title, + *, + admission=None, + record_changing=None, + cancellable=True, + scope=None, + groups=(), + clock=None, ): self._clock = clock or time.time self.id = uuid.uuid4().hex self.kind = kind self.title = title - self.record_changing = record_changing + self.admission = resolve_admission(admission, record_changing) + self.cancellable = bool(cancellable) self.scope = dict(scope or {}) self.state = "running" self.started_at = self._clock() @@ -110,6 +167,10 @@ class Job: self.task = None self._subscribers = set() + @property + def record_changing(self): + return self.admission == "records" + @property def running(self): return self.state == "running" @@ -240,6 +301,9 @@ class Job: # What detail() lists, so a view re-reads it only when it changed. "listed": {group: len(rows) for group, rows in self.outcomes.items()}, "log_length": len(self.log), + # A view offers Cancel only when the server can stop the job. + "cancellable": self.cancellable, + "admission": self.admission, } def detail(self): @@ -291,33 +355,88 @@ class JobRegistry: self._logger = logger_ or logger self._jobs = {} - def start(self, kind, title, *, record_changing, scope, runner, groups=()) -> Job: + def start( + self, + kind, + title, + *, + admission=None, + record_changing=None, + cancellable=True, + scope=None, + runner, + groups=(), + ) -> Job: """Create a job and run ``runner(job)`` as a task. - Raises JobBusy, naming the job in the way, when a job of ``kind`` is - running or when ``record_changing`` and another record-changing job - runs. Must be called from the event loop. + ``admission`` is one of ``ADMISSIONS``; ``record_changing=True`` is + the older spelling of ``records`` and ``False`` of ``read``. One of + the two is required (TypeError otherwise), so no new job runs as + ``read`` beside a scan by omission. Raises JobBusy, naming the job in + the way, when the admission class refuses it (see the module + docstring). Must be called from the event loop. """ - blocker = self.blocker(kind, record_changing=record_changing) - if blocker is not None: - raise JobBusy(blocker) - self._prune() - job = Job( - kind, title, record_changing=record_changing, scope=scope, groups=groups + if admission is None and record_changing is None: + raise TypeError( + f"JobRegistry.start({kind!r}) needs an admission class " + "(admission= or record_changing=)" + ) + job = self._admit( + kind, + title, + admission=admission, + record_changing=record_changing, + cancellable=cancellable, + scope=scope, + groups=groups, ) - self._jobs[job.id] = job job.task = asyncio.ensure_future(self._run(job, runner)) return job - def blocker(self, kind, *, record_changing): - """The running job a new ``kind`` job would be refused for, or None.""" + def hold(self, kind, title, *, admission, scope=None): + """A running, non-cancellable job for a legacy-page operation. + + Returns an async context manager. Entering it registers the job and + yields it, or raises JobBusy at once, without waiting, when the + admission class refuses it; nothing is registered until then, so a + hold that is never entered leaves no job behind. The job ends when + the block exits: ``finished``, or ``failed`` when it raised (unless + the body already ended it). A caller that answers a refusal in its + own way checks ``blocker(kind, admission=...)`` first, with no await + before the ``async with``. + """ + resolve_admission(admission) # an unknown class fails at the call + return _Hold( + lambda: self._admit( + kind, title, admission=admission, cancellable=False, scope=scope + ) + ) + + def _admit(self, kind, title, *, admission, record_changing=None, **job_args): + resolved = resolve_admission(admission, record_changing) + blocker = self.blocker(kind, admission=resolved) + if blocker is not None: + raise JobBusy(blocker) + self._prune() + job = Job(kind, title, admission=resolved, **job_args) + self._jobs[job.id] = job + return job + + def blocker(self, kind, *, admission=None, record_changing=None): + """The running job a new ``kind`` job would be refused for, or None. + + The same kind refuses first; then the oldest running job whose class + the new job's admission class may not run beside. + """ same_kind = self.active(kind) if same_kind is not None: return same_kind - if record_changing: - for job in self._jobs.values(): - if job.running and job.record_changing: - return job + refused = _REFUSED_BESIDE[resolve_admission(admission, record_changing)] + running = [job for job in self._jobs.values() if job.running] + running.sort(key=lambda job: job._started_seq) + for job in running: + if job.admission in refused: + return job return None async def _run(self, job, runner): @@ -362,11 +481,28 @@ class JobRegistry: def cancel(self, job_id) -> bool: """Ask a running job to stop at its next safe boundary. - Returns whether a cancellation was requested: False for a finished or - unknown job, so a cancel after the end is a no-op. + Returns whether a cancellation was requested: False for a finished, + unknown or non-cancellable job, so a cancel after the end is a no-op. """ job = self._jobs.get(job_id) - if job is None or job.finished: + if job is None or job.finished or not job.cancellable: return False job.cancelled = True return True + + +class _Hold: + """The async context manager ``JobRegistry.hold`` returns.""" + + def __init__(self, admit): + self._admit = admit + self._job = None + + async def __aenter__(self): + self._job = self._admit() + return self._job + + async def __aexit__(self, exc_type, exc, tb): + if self._job.running: + self._job.finish("finished" if exc_type is None else "failed") + return False diff --git a/py/api/lora_integration.py b/py/api/lora_integration.py index 537b35c..7e66c2d 100644 --- a/py/api/lora_integration.py +++ b/py/api/lora_integration.py @@ -5,12 +5,47 @@ from pathlib import Path from aiohttp import web +from ..config_store import ConfigUnreadable, RevisionConflict, get_config_store +from .jobs import LORA_IMPORT_JOB_KIND + +# Flags of POST /lora/enable; each must be a JSON boolean when present. +_LORA_FLAGS = ("enabled", "trigger_words_enabled", "clear_civitai_api_key") + _LORA_PATH_ERROR = ( "LoraManager path must be a directory under ComfyUI's custom_nodes " "whose name contains 'lora'" ) +def _lora_settings_mutation(values): + """The config commit for a LoRA save: only ``integrations.lora_manager``. + + The submitted keys replace their values; other LoRA keys, other + integrations and every other section stay as they are on disk. + """ + + def mutate(config): + integrations = config.get("integrations") + integrations = dict(integrations) if isinstance(integrations, dict) else {} + lora = integrations.get("lora_manager") + lora = dict(lora) if isinstance(lora, dict) else {} + lora.update(values) + integrations["lora_manager"] = lora + config["integrations"] = integrations + return config + + return mutate + + +def _lora_error_frame(message): + """An import stream's error event, as the admin page's modal reads it. + + The modal shows ``status`` and ``progress`` from every frame and closes + on ``type: error``, naming ``message``. + """ + return {"type": "error", "status": message, "progress": 0, "message": message} + + class LoraIntegrationMixin: """Mixin providing LoraManager detection, scanning, and trigger word endpoints.""" @@ -131,14 +166,25 @@ class LoraIntegrationMixin: return allowed, None async def lora_enable(self, request): - """Enable or disable LoraManager integration and save to config.json.""" + """Enable or disable LoraManager integration and save it to config.json. + + The values are committed through the config store first (#220) and + published at runtime only once config.json holds them; a save that + cannot be written reports ``persisted: false`` and changes nothing. + The CivitAI key is never echoed back. + """ try: - data = await request.json() + data, error = await self._read_json_body(request) + if error is not None: + return error + for flag in _LORA_FLAGS: + if flag in data and not isinstance(data[flag], bool): + return self._bad_lora_request(f"{flag} must be true or false") enabled = data.get("enabled", False) path = data.get("path", "") trigger_words = data.get("trigger_words_enabled", False) - from ..config import IntegrationConfig, PromptManagerConfig + from ..config import IntegrationConfig from ..lora_utils import get_trigger_cache # The key is never echoed back to the UI, so a blank or absent @@ -161,20 +207,40 @@ class LoraIntegrationMixin: if error is not None: return error - # Update in-memory config - IntegrationConfig.LORA_MANAGER_ENABLED = enabled - IntegrationConfig.LORA_MANAGER_PATH = path - IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED = trigger_words - IntegrationConfig.CIVITAI_API_KEY = civitai_key + values = { + "enabled": enabled, + "path": path, + "trigger_words_enabled": trigger_words, + "civitai_api_key": civitai_key, + } - # Persist to the configured config.json - PromptManagerConfig.save_to_file() + def publish(_config): + # Under the commit lock, so the last save written is the one + # the runtime keeps. + IntegrationConfig.LORA_MANAGER_ENABLED = enabled + IntegrationConfig.LORA_MANAGER_PATH = path + IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED = trigger_words + IntegrationConfig.CIVITAI_API_KEY = civitai_key + + try: + _saved, revision = await self._run_in_executor( + get_config_store().commit, + _lora_settings_mutation(values), + expected_revision=None, + publish=publish, + ) + except (ConfigUnreadable, RevisionConflict, OSError) as exc: + return self._config_commit_failure(exc) - # Load trigger word cache if enabling cache = get_trigger_cache() if enabled and trigger_words and path: - count = await self._run_in_executor(cache.load, path) - self.logger.info(f"Trigger word cache loaded: {count} LoRAs") + try: + count = await self._run_in_executor(cache.load, path) + self.logger.info(f"Trigger word cache loaded: {count} LoRAs") + except Exception as e: + # Saved and enabled; only the trigger words are missing. + self.logger.warning(f"Trigger word cache could not be loaded: {e}") + cache.clear() elif not enabled: cache.clear() @@ -184,6 +250,10 @@ class LoraIntegrationMixin: "enabled": enabled, "path": self._public_path(path), "trigger_words_enabled": trigger_words, + "persisted": True, + "revision": revision, + "runtime_applied": True, + "cache_loaded": cache.is_loaded, } ) except Exception as e: @@ -218,19 +288,33 @@ class LoraIntegrationMixin: reason="OK", headers={"Content-Type": "text/event-stream"}, ) - await response.prepare(request) async def send_progress(data): line = f"data: {json.dumps(data)}\n\n" await response.write(line.encode("utf-8")) - try: - await self._lora_scan_stream(allowed, send_progress) - except Exception as e: - self.logger.exception("LoRA scan failed") - await send_progress({"type": "error", "message": self._public_error(e)}) + # The import replaces prompts and images: a records hold, so it + # never runs beside a scan, restore or tagging job. + busy = self._jobs.blocker(LORA_IMPORT_JOB_KIND, admission="records") + if busy is not None: + await response.prepare(request) + message = f"{busy.title} already running" + await send_progress(_lora_error_frame(message)) + await response.write_eof() + return response - await response.write_eof() + async with self._jobs.hold( + LORA_IMPORT_JOB_KIND, "LoRA import (admin page)", admission="records" + ) as job: + await response.prepare(request) + try: + await self._lora_scan_stream(allowed, send_progress) + except Exception as e: + self.logger.exception("LoRA scan failed") + job.finish("failed") + await send_progress(_lora_error_frame(self._public_error(e))) + + await response.write_eof() return response except Exception as e: diff --git a/py/api/settings_routes.py b/py/api/settings_routes.py new file mode 100644 index 0000000..258ad2a --- /dev/null +++ b/py/api/settings_routes.py @@ -0,0 +1,339 @@ +"""Settings API routes: GET and POST /prompt_manager/settings (#220). + +A save validates the payload, commits it to config.json through the config +store and publishes it at runtime inside that commit; a save that cannot be +written changes nothing at runtime and says so. ``_config_commit_failure`` +is shared with the LoRA integration's save. +""" + +import os + +from aiohttp import web + +from ..config_store import ( + ConfigUnreadable, + RevisionConflict, + get_config_store, +) +from .admin import _current_image_monitor + +try: + from ...utils.validators import validate_result_timeout, validate_worker_threads +except ImportError: + from utils.validators import validate_result_timeout, validate_worker_threads + + +# Settings owned by POST /settings: config attribute -> (section, key). +_SETTINGS_FILE_KEYS = { + "RESULT_TIMEOUT": ("web_ui", "result_timeout"), + "WEBUI_DISPLAY_MODE": ("web_ui", "webui_display_mode"), + "INFINITE_SCROLL": ("web_ui", "infinite_scroll"), + "WORKER_THREADS": ("performance", "worker_threads"), +} + + +def _section(config, name): + """``config[name]`` as a fresh dict (a non-object section starts empty).""" + value = config.get(name) + return dict(value) if isinstance(value, dict) else {} + + +def _settings_mutation(updates, new_roots): + """The config commit for a settings save: only the submitted keys change. + + Every other section and key (integrations and the CivitAI key, database, + unknown sections) is kept as it is on disk. + """ + + def mutate(config): + for name, value in updates.items(): + section, key = _SETTINGS_FILE_KEYS[name] + merged = _section(config, section) + merged[key] = value + config[section] = merged + if new_roots is not None: + gallery = _section(config, "gallery") + monitoring = _section(gallery, "monitoring") + monitoring["directories"] = list(new_roots) + gallery["monitoring"] = monitoring + config["gallery"] = gallery + return config + + return mutate + + +async def _read_json(request): + """Parse a JSON object body through the package's capped reader.""" + from . import _read_json_body + + return await _read_json_body(request) + + +class SettingsRoutesMixin: + """Mixin providing the settings read and save endpoints.""" + + def _register_settings_routes(self, routes): + @routes.get("/prompt_manager/settings") + async def get_settings_route(request): + return await self.get_settings(request) + + @routes.post("/prompt_manager/settings") + async def save_settings_route(request): + return await self.save_settings(request) + + async def get_settings(self, request): + """Get current settings.""" + try: + from ..config import PromptManagerConfig, GalleryConfig + + root_paths = [ + self._public_path(d) for d in GalleryConfig.MONITORING_DIRECTORIES + ] + revision = await self._run_in_executor(self._config_revision) + return web.json_response( + { + "success": True, + **revision, + "settings": { + "result_timeout": PromptManagerConfig.RESULT_TIMEOUT, + "webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE, + "infinite_scroll": PromptManagerConfig.INFINITE_SCROLL, + "worker_threads": PromptManagerConfig.WORKER_THREADS, + "cpu_count": PromptManagerConfig.max_worker_threads(), + "gallery_root_paths": root_paths, + "gallery_root_path": root_paths[0] if root_paths else "", + "monitored_directories": [ + self._public_path(d) for d in self._monitored_directories() + ], + }, + } + ) + except Exception as e: + self.logger.error(f"Get settings error: {e}", exc_info=True) + return web.json_response( + {"success": False, "error": "Failed to get settings"}, + status=500, + ) + + def _monitored_directories(self): + """Directories the running image monitor watches, else the configured roots.""" + from ..config import GalleryConfig + + monitor = _current_image_monitor() + if monitor is not None: + return list(getattr(monitor, "monitored_directories", [])) + return list(GalleryConfig.MONITORING_DIRECTORIES) + + async def save_settings(self, request): + """Save settings: validate, commit to config.json, then publish (#220). + + Nothing changes at runtime until config.json holds the new values, so + a save that cannot be written reports ``persisted: false`` and leaves + the running settings as they were. ``expected_revision`` (the + ``revision`` from GET) refuses to overwrite a file changed meanwhile; + without it the payload merges into what is on disk (legacy page). + """ + try: + data, error_response = await _read_json(request) + if error_response is not None: + return error_response + + expected_revision = data.get("expected_revision") + if expected_revision is not None and not isinstance(expected_revision, str): + return web.json_response( + {"success": False, "error": "expected_revision must be a string"}, + status=400, + ) + try: + updates = self._validate_settings(data) + except ValueError as ve: + return web.json_response( + {"success": False, "error": str(ve)}, status=400 + ) + + new_roots, error = self._parse_gallery_roots(data) + if error: + return web.json_response({"success": False, "error": error}, status=400) + + published = {} + + def publish(_config): + # Under the commit lock: saves reach the runtime in the order + # they were written, so the runtime ends on the last one. + published.update(self._publish_settings(updates, new_roots)) + + try: + _saved, revision = await self._run_in_executor( + get_config_store().commit, + _settings_mutation(updates, new_roots), + expected_revision=expected_revision, + publish=publish, + ) + except (ConfigUnreadable, RevisionConflict, OSError) as exc: + return self._config_commit_failure(exc) + + return web.json_response( + { + "success": True, + "message": "Settings saved successfully", + "restart_required": bool(published["restart_required_fields"]), + "persisted": True, + "revision": revision, + **published, + } + ) + except Exception as e: + self.logger.error(f"Save settings error: {e}", exc_info=True) + return web.json_response( + {"success": False, "error": "Failed to save settings"}, + status=500, + ) + + def _publish_settings(self, updates, new_roots): + """Install committed settings at runtime (inside the config commit). + + The values are already durable. A publication that fails is reported + as ``runtime_applied: false`` with the fields that need a restart; + nothing is retried or rolled back. Never raises. + """ + from ..config import PromptManagerConfig + + for name, value in updates.items(): + setattr(PromptManagerConfig, name, value) + restart_fields = [] + runtime_applied = True + if new_roots is not None: + try: + if self._apply_gallery_roots(new_roots): + restart_fields.append("gallery_root_paths") + except Exception as e: + self.logger.error( + f"Saved gallery roots could not be applied: {e}", exc_info=True + ) + runtime_applied = False + restart_fields.append("gallery_root_paths") + return { + "runtime_applied": runtime_applied, + "restart_required_fields": restart_fields, + } + + def _config_commit_failure(self, exc): + """The response for a config commit that wrote nothing.""" + not_saved = {"success": False, "persisted": False, "runtime_applied": False} + if isinstance(exc, ConfigUnreadable): + self.logger.warning(f"Settings not saved: {exc}") + return web.json_response( + { + **not_saved, + "code": "config-unreadable", + "error": ( + "config.json cannot be read, so nothing was saved. " + "Fix or remove the file, then save again." + ), + }, + status=409, + ) + if isinstance(exc, RevisionConflict): + return web.json_response( + { + **not_saved, + "code": "settings-changed", + "error": "Settings changed since they were loaded. Reload them.", + "revision": exc.current, + }, + status=409, + ) + self.logger.error(f"Could not write config.json: {exc}") + return web.json_response( + { + **not_saved, + "code": "config-write-failed", + "error": "Settings could not be written to config.json.", + }, + status=500, + ) + + @staticmethod + def _config_revision(): + """``{"revision": ...}``, or a null revision with ``config_error``.""" + try: + _data, revision = get_config_store().read() + except ConfigUnreadable: + return {"revision": None, "config_error": "config-unreadable"} + return {"revision": revision} + + @staticmethod + def _validate_settings(data): + """Scalar settings of a payload as ``{CONFIG_ATTR: value}``. + + Raises ValueError with a user-facing message on the first bad value; + nothing is applied until every value passed. + """ + from ..config import PromptManagerConfig + + updates = {} + if "result_timeout" in data: + updates["RESULT_TIMEOUT"] = validate_result_timeout(data["result_timeout"]) + if "webui_display_mode" in data: + mode = data["webui_display_mode"] + if mode not in ("popup", "newtab"): + raise ValueError("webui_display_mode must be 'popup' or 'newtab'") + updates["WEBUI_DISPLAY_MODE"] = mode + if "infinite_scroll" in data: + if not isinstance(data["infinite_scroll"], bool): + raise ValueError("infinite_scroll must be true or false") + updates["INFINITE_SCROLL"] = data["infinite_scroll"] + if "worker_threads" in data: + updates["WORKER_THREADS"] = validate_worker_threads( + data["worker_threads"], PromptManagerConfig.max_worker_threads() + ) + return updates + + def _parse_gallery_roots(self, data): + """Extract and validate gallery roots from a settings payload. + + Accepts ``gallery_root_paths`` (list, preferred) or the legacy single + ``gallery_root_path`` string. Blank entries are dropped. + + Returns: + (roots, error): ``roots`` is the validated list, or ``None`` when + the payload carries no gallery root key; ``error`` is a message + when validation failed (and ``roots`` is then ``None``). + """ + if "gallery_root_paths" in data: + raw = data["gallery_root_paths"] + if not isinstance(raw, list): + return None, "gallery_root_paths must be a list" + elif "gallery_root_path" in data: + raw = [data["gallery_root_path"]] + else: + return None, None + + from ..config import GalleryConfig + + roots = [] + for entry in raw: + if not isinstance(entry, str): + return None, "Gallery root paths must be strings" + entry = entry.strip() + if not entry: + continue + ok, reason = GalleryConfig.validate_gallery_root(entry) + if not ok: + return ( + None, + f"Invalid gallery root '{os.path.basename(entry)}': {reason}", + ) + roots.append(GalleryConfig.resolve_gallery_root(entry)) + return roots, None + + def _apply_gallery_roots(self, roots): + """Install validated gallery roots; returns True when they changed.""" + from ..config import GalleryConfig + + if roots == list(GalleryConfig.MONITORING_DIRECTORIES): + return False + GalleryConfig.MONITORING_DIRECTORIES = roots + self._cached_output_dir = None + self.reset_gallery_roots_cache() + return True diff --git a/py/config_store.py b/py/config_store.py new file mode 100644 index 0000000..949423a --- /dev/null +++ b/py/config_store.py @@ -0,0 +1,197 @@ +"""The one serialized writer of config.json (#220). + +Every save that changes config.json goes through ``ConfigStore.commit``: the +settings page, the LoRA integration and, later, every System section. A +commit runs under one process-wide lock, re-reads the file inside it, applies +the caller's change to a copy and replaces the file atomically. So two saves +never overwrite each other's keys, sections this process does not know are +kept byte-for-byte in meaning, and a write that fails leaves the old file in +place and raises instead of being logged and reported as saved. + +A revision names the file content a client last saw: ``"sha256:"`` of +the file bytes, or ``"absent"`` when there is no file. A commit that names a +revision refuses to write over different content (``RevisionConflict``). + +Reads take the same lock, so a read handle is never open while the file is +replaced (Windows refuses that), and a caller's ``publish`` runs inside the +lock right after its write, so runtimes publish saves in the order they were +written. A symlinked config.json is written beside its real target. + +config.json can hold the CivitAI key, so the file is written owner-only. +""" + +import copy +import hashlib +import json +import os +import tempfile +import threading +import time +from typing import Callable, Optional, Tuple + +ABSENT_REVISION = "absent" + +# One lock for the process: every ConfigStore writes the same file. It is +# re-entrant so a commit's ``publish`` callback may read the config again. +_COMMIT_LOCK = threading.RLock() + +# Windows: an antivirus, indexer or backup handle on config.json makes +# os.replace fail with PermissionError for a moment, so it is retried briefly. +_REPLACE_ATTEMPTS = 5 if os.name == "nt" else 1 +_REPLACE_BACKOFF_SECONDS = 0.05 + + +class ConfigUnreadable(Exception): + """config.json exists but is not a readable JSON object; it is left alone.""" + + +class RevisionConflict(Exception): + """config.json changed since the revision the caller saw.""" + + def __init__(self, current: str): + super().__init__("config.json changed since it was read") + self.current = current + + +def revision_of(data: Optional[bytes]) -> str: + """The revision of a file's bytes; ``None`` means the file is absent.""" + if data is None: + return ABSENT_REVISION + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def _default_path() -> str: + from .config import PromptManagerConfig + + return PromptManagerConfig.get_config_path() + + +def _replace(source: str, target: str) -> None: + """``os.replace`` with a short retry on a transient Windows PermissionError.""" + for attempt in range(1, _REPLACE_ATTEMPTS + 1): + try: + os.replace(source, target) + return + except PermissionError: + if attempt == _REPLACE_ATTEMPTS: + raise + time.sleep(_REPLACE_BACKOFF_SECONDS) + + +def _write_atomically(path: str, payload: bytes) -> None: + """Replace ``path`` with ``payload``: temp file beside it, fsync, replace. + + The descriptor is closed before ``os.replace`` (Windows refuses to replace + over an open handle), and on any failure the temp file is closed and + removed, so a failed write leaves the old file and no stray temp file. + ``path`` is resolved first: a symlink keeps pointing at the updated + target. A single-file bind mount cannot be replaced (EBUSY) and fails as + an OSError, which callers report as not saved. + """ + path = os.path.realpath(path) + parent = os.path.dirname(path) or "." + os.makedirs(parent, exist_ok=True) + fd, temp_path = tempfile.mkstemp(dir=parent, prefix=".config-", suffix=".tmp") + try: + try: + handle = os.fdopen(fd, "wb") + except BaseException: + os.close(fd) + raise + with handle: # closed before the replace, also when a write fails + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + if os.name == "posix": + os.chmod(temp_path, 0o600) + _replace(temp_path, path) + except BaseException: + try: + os.unlink(temp_path) + except OSError: + pass + raise + + +class ConfigStore: + """Serialized, atomic reads and commits of config.json.""" + + def __init__(self, path_getter: Optional[Callable[[], str]] = None): + self._path_getter = path_getter or _default_path + + @property + def path(self) -> str: + """The config file this store reads and writes (resolved per call).""" + return self._path_getter() + + def read(self) -> Tuple[dict, str]: + """``(config, revision)``; a missing file is ``({}, "absent")``. + + Raises ConfigUnreadable when the file cannot be read, is not JSON, or + is not a JSON object. Waits for a running commit (see the module + docstring). + """ + with _COMMIT_LOCK: + return self._read(self.path) + + @staticmethod + def _read(path: str) -> Tuple[dict, str]: + try: + with open(path, "rb") as handle: + raw = handle.read() + except (FileNotFoundError, NotADirectoryError): + # No file there (a parent that is a file cannot hold one either); + # a commit then fails on the write, as config-write-failed. + return {}, ABSENT_REVISION + except OSError as exc: + raise ConfigUnreadable(f"config.json cannot be read: {exc}") from exc + try: + data = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, ValueError) as exc: + raise ConfigUnreadable("config.json is not valid JSON") from exc + if not isinstance(data, dict): + raise ConfigUnreadable("config.json is not a JSON object") + return data, revision_of(raw) + + def commit( + self, + mutate: Callable[[dict], dict], + *, + expected_revision: Optional[str], + publish: Optional[Callable[[dict], None]] = None, + ) -> Tuple[dict, str]: + """Apply ``mutate`` to the current config and write the result. + + ``mutate`` receives a deep copy of the file's content and returns the + whole new config (an object). ``expected_revision`` None merges into + whatever is on disk (legacy partial updates); a revision refuses to + write over different content. Returns ``(config, revision)`` as + written. Raises ConfigUnreadable, RevisionConflict, TypeError (a + non-object result) or OSError (the write failed; the old file stays). + + ``publish(config)``, when given, runs under the lock once the file + holds the new config, so concurrent saves reach the runtime in the + order they were written. It must handle its own errors: the values + are already durable when it runs. + """ + with _COMMIT_LOCK: + path = self.path + current, revision = self._read(path) + if expected_revision is not None and expected_revision != revision: + raise RevisionConflict(revision) + candidate = mutate(copy.deepcopy(current)) + if not isinstance(candidate, dict): + raise TypeError("A config commit must produce a JSON object") + payload = json.dumps(candidate, indent=2).encode("utf-8") + _write_atomically(path, payload) + if publish is not None: + publish(candidate) + return candidate, revision_of(payload) + + +_store = ConfigStore() + + +def get_config_store() -> ConfigStore: + """The process's config store (on ``PromptManagerConfig.get_config_path``).""" + return _store diff --git a/tests/browser/workflows.spec.mjs b/tests/browser/workflows.spec.mjs index 2f31eda..0ee4620 100644 --- a/tests/browser/workflows.spec.mjs +++ b/tests/browser/workflows.spec.mjs @@ -74,6 +74,24 @@ test('a page whose leave was called off reads its list again on the next input', await expect(page.locator('#resultsTitle')).toHaveText('Recent Prompts'); }); +// The gallery's startup thumbnail check runs a second after load. When that second ends while the page is leaving, +// the check must start no read: WebKit fails it with "Fetch API cannot load … due to access control checks." and +// Playwright reports a page error (seen in CI on PR #289). The page clock makes the second end after the leave began. +for (const leave of ['beforeunload', 'pagehide']) { + test(`the gallery startup check starts no read after ${leave}`, async ({ page }) => { + await page.clock.install(); + const statusReads = []; + page.on('request', req => { if (req.url().includes('/images/thumbnails/status')) statusReads.push(req.url()); }); + await page.goto('/prompt_manager/gallery'); + await expect(page.locator('#galleryGrid .image-item')).toHaveCount(2); + await page.evaluate(type => window.dispatchEvent(new Event(type)), leave); + await page.clock.runFor(1500); + // A read made after the clock ran is listed, so the listener saw every status read. + await page.evaluate(() => fetch('/prompt_manager/images/thumbnails/status?probe=1')); + expect(statusReads.map(url => new URL(url).search)).toEqual(['?probe=1']); + }); +} + test('actual gallery review generates and persists tags for the selected root', async ({ page, request }) => { await page.goto('/prompt_manager/gallery'); await expect(page.locator('#galleryGrid .image-item')).toHaveCount(2); diff --git a/tests/browser/workspace-activity.spec.mjs b/tests/browser/workspace-activity.spec.mjs index 80f9fcd..4479751 100644 --- a/tests/browser/workspace-activity.spec.mjs +++ b/tests/browser/workspace-activity.spec.mjs @@ -678,6 +678,31 @@ test('cancel scan ends cancelled with partial counts', async ({ page, request }) await expect(heading(page).getByRole('button', { name: 'Cancel scan', exact: true })).not.toHaveAttribute('aria-disabled', 'true'); }); +// Truth rule: a held legacy-page job (a restore from the admin page) cannot be cancelled, so neither the Activity +// list nor the job's view offers Cancel; the cancellable scan beside it keeps its own (review I3). +test('a held legacy job shows no Cancel in the list or its view', async ({ page, request }) => { + await page.setViewportSize({ width: 1600, height: 1100 }); + const response = await request.post('/__test__/jobs/hold', { data: { kind: 'restore', title: 'Database restore (admin page)' } }); + await expect(response).toBeOK(); + const held = (await response.json()).job; + expect(held.cancellable).toBe(false); + await page.goto(`${workspace}#/activity`); + const item = items(page).filter({ has: page.getByRole('link', { name: /Database restore \(admin page\)$/ }) }); + await expect(item.locator('.chip')).toHaveText('Running'); + await expect(item.locator('progress')).toHaveCount(1); + await expect(main(page).getByRole('button', { name: 'Cancel job', exact: true })).toHaveCount(0); + + await item.getByRole('link').click(); + await expect(heading(page).locator('h1')).toHaveText('Database restore (admin page)'); + await expect(heading(page).locator('.heading-actions > button')).toHaveText(['Download report']); + await expect(heading(page).getByRole('button', { name: /^Cancel/ })).toHaveCount(0); + + // When the hold ends the view shows the end, still without a Cancel. + await finish(request, held.id); + await expect(heading(page).locator('p.muted')).toHaveText(/^Finished/); + await expect(heading(page).locator('.heading-actions > button')).toHaveText(['Download report']); +}); + test('busy start names the running job', async ({ page, request }) => { await page.setViewportSize({ width: 1600, height: 1100 }); const running = await startJob(request, { title: 'Database maintenance', kind: 'maintenance' }); diff --git a/tests/browser/workspace_activity_fixture.py b/tests/browser/workspace_activity_fixture.py index c2a3d4d..41dba88 100644 --- a/tests/browser/workspace_activity_fixture.py +++ b/tests/browser/workspace_activity_fixture.py @@ -14,6 +14,9 @@ reading files, so a test sets its progress and counters exactly: the progress to every subscriber; - ``POST /__test__/jobs/{id}/finish`` ends it as ``finished``, ``cancelled`` or ``failed``; a cancel through the real route ends it ``cancelled``; +- ``POST /__test__/jobs/hold`` opens a legacy-page hold (``kind``, ``title``): + a running, non-cancellable ``records`` job, as a legacy restore or + maintenance request takes, until ``finish`` ends it; - ``POST /__test__/jobs/restart`` acts out a ComfyUI restart: the open job streams drop without a final event and the registry starts empty. """ @@ -86,6 +89,7 @@ def register_activity_commands(app, routes, api): streams = set() endings = {} + holders = set() @web.middleware async def track_streams(request, handler): @@ -163,6 +167,36 @@ def register_activity_commands(app, routes, api): endings[job.id] = ending return web.json_response({"success": True, "job": job.summary()}) + @routes.post("/__test__/jobs/hold") + async def hold_job(request): + data = await body(request) + loop = asyncio.get_running_loop() + ending = loop.create_future() + entered = loop.create_future() + + async def hold(): + async with api._jobs.hold( + str(data.get("kind", "restore")), + str(data.get("title", "Database restore (admin page)")), + admission="records", + ) as job: + entered.set_result(job) + state = await ending + if state != "finished": + job.finish(state) + + holder = asyncio.ensure_future(hold()) + holders.add(holder) + holder.add_done_callback(holders.discard) + await asyncio.wait({entered, holder}, return_when=asyncio.FIRST_COMPLETED) + if not entered.done(): + if isinstance(holder.exception(), JobBusy): + raise web.HTTPConflict() + raise holder.exception() + job = entered.result() + endings[job.id] = ending + return web.json_response({"success": True, "job": job.summary()}) + @routes.post("/__test__/jobs/{job_id}/step") async def step_job(request): job = job_or_404(request) @@ -210,6 +244,7 @@ def register_activity_commands(app, routes, api): for transport in list(streams): transport.close() tasks = [job.task for job in old.list() if job.task and not job.task.done()] + tasks += [holder for holder in holders if not holder.done()] for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) diff --git a/tests/js/gallery-thumbnails.test.js b/tests/js/gallery-thumbnails.test.js index 431beea..5b085f2 100644 --- a/tests/js/gallery-thumbnails.test.js +++ b/tests/js/gallery-thumbnails.test.js @@ -73,6 +73,28 @@ test('Not Now persists the collection state and suppresses the unchanged prompt' assert.equal(f.prompts.length, 1); }); +// WebKit fails a fetch a leaving page starts ("Fetch API cannot load … due to access control checks.") and logs it +// as a page error: the delayed startup check starts no read once the page has begun to leave (as admin.js, #290). +test('the startup check starts no read once the page is leaving', async () => { + const f = fixture({ success: true, total_images: 100, pending: 30, collection_id: 'a' }); + f.manager.pageLifetime = new AbortController(); + f.manager.pageLifetime.abort(); + await f.check(); + assert.equal(f.requests.length, 0); + assert.equal(f.prompts.length, 0); +}); + +test('the startup check passes the page lifetime to its read', async () => { + const f = fixture({ success: true, total_images: 100, pending: 30, collection_id: 'a' }); + const signals = []; + f.manager.pageLifetime = new AbortController(); + const get = f.manager.api.get; + f.manager.api.get = async (url, options) => { signals.push(options?.signal); return get(url); }; + await f.check(); + assert.deepEqual(signals, [f.manager.pageLifetime.signal]); + assert.equal(f.prompts.length, 1); +}); + test('disabled checks do not request collection status', async () => { const f = fixture({}, { checkThumbnailsAtStartup: false }); await f.check(); diff --git a/tests/js/legacy-refusals.test.js b/tests/js/legacy-refusals.test.js new file mode 100644 index 0000000..6daec1c --- /dev/null +++ b/tests/js/legacy-refusals.test.js @@ -0,0 +1,150 @@ +// The legacy admin and gallery pages show a server refusal's reason (#220, PR #289 review I1 and M8): a job that +// runs in the workspace refuses their record-changing requests, and a corrupt config.json refuses a settings save. +// Each page must name the reason, not report a lost connection, a bare HTTP status or a success. +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const vm = require('node:vm'); + +function elements() { + const nodes = {}; + return { + nodes, + getElementById: id => nodes[id] ||= { id, textContent: '', innerHTML: '', disabled: false, style: {}, + value: '', checked: false, addEventListener() {}, querySelectorAll: () => [] }, + }; +} + +function jsonResponse(status, body) { + return { ok: status >= 200 && status < 300, status, statusText: status === 409 ? 'Conflict' : 'OK', + json: async () => body }; +} + +function sseResponse(frames) { + const bytes = new TextEncoder().encode(frames.map(frame => `data: ${JSON.stringify(frame)}\n\n`).join('')); + let sent = false; + return { ok: true, status: 200, body: { getReader: () => ({ + read: async () => (sent ? { done: true } : (sent = true, { done: false, value: bytes })), + }) } }; +} + +function load(file, className, fetchUrl, extra = {}, dom = elements()) { + const timers = []; + const notifications = []; + const storage = new Map([['pm_cached', '1'], ['userData', 'keep']]); + const localStorage = { + getItem: key => storage.get(key) ?? null, + setItem: (key, value) => storage.set(key, String(value)), + removeItem: key => storage.delete(key), + }; + const context = vm.createContext({ + document: { readyState: 'loading', addEventListener() {}, getElementById: dom.getElementById, + querySelectorAll: () => [] }, + window: {}, console: { error() {}, warn() {}, debug() {} }, + setTimeout: callback => timers.push(callback), + confirm: () => true, + localStorage, + Object: { ...Object, keys: target => (target === localStorage ? [...storage.keys()] : Object.keys(target)) }, + TextDecoder, TextEncoder, JSON, + ApiClient: { fetchUrl }, + ...extra, + }); + const source = fs.readFileSync(path.join(__dirname, `../../web/js/${file}`), 'utf8'); + const Page = vm.runInContext(`${source}\n${className};`, context, { filename: path.join(__dirname, `../../web/js/${file}`) }); + const page = Object.create(Page.prototype); + page.showNotification = (message, type) => notifications.push([type, message]); + page.showModal = id => notifications.push(['show', id]); + page.hideModal = id => notifications.push(['hide', id]); + return { page, dom, timers, notifications, storage }; +} + +const BUSY = 'Scan for metadata already running'; + +test('gallery cache clear reports a refused server clear and never claims success', async () => { + const calls = []; + const { page, dom, notifications, storage } = load('gallery.js', 'GalleryManager', async url => { + calls.push(url); + return jsonResponse(409, { success: false, error: BUSY }); + }); + page.getSettings = () => ({ thumbnailsGenerated: true }); + dom.getElementById('thumbnailStatus').textContent = 'Generated'; + + await page.clearCache(); + + assert.deepEqual(calls, ['/prompt_manager/images/clear-thumbnails']); + assert.deepEqual(notifications, [['error', `Thumbnails were not cleared: ${BUSY}`]]); + assert.equal(dom.nodes.thumbnailStatus.textContent, 'Generated'); + assert.equal(storage.get('gallerySettings'), undefined); // thumbnailsGenerated is not reset +}); + +test('gallery cache clear still says so when the server clears', async () => { + const { page, dom, notifications, storage } = load('gallery.js', 'GalleryManager', + async () => jsonResponse(200, { success: true, cleared_count: 3 })); + page.getSettings = () => ({ thumbnailsGenerated: true }); + + await page.clearCache(); + + assert.deepEqual(notifications, [['success', 'PromptManager cache cleared successfully! Original images untouched.']]); + assert.equal(dom.nodes.thumbnailStatus.textContent, 'Not generated'); + assert.equal(JSON.parse(storage.get('gallerySettings')).thumbnailsGenerated, false); + assert.equal(storage.get('userData'), 'keep'); + assert.equal(storage.has('pm_cached'), false); +}); + +test('admin LoRA import shows a refusal frame and closes its modal', async () => { + const { page, dom, timers, notifications } = load('admin.js', 'PromptAdmin', + async () => sseResponse([{ type: 'error', status: BUSY, progress: 0, message: BUSY }])); + page.saveLoraSettings = async () => true; + page.loadStatistics = () => assert.fail('a refused import must not reload statistics'); + + await page.runLoraImport(); + + assert.equal(dom.nodes.loraImportStatus.textContent, `Error: ${BUSY}`); + assert.deepEqual(notifications.filter(([type]) => type === 'error'), [['error', `LoRA import failed: ${BUSY}`]]); + assert.equal(timers.length, 1); + timers[0](); + assert.deepEqual(notifications.at(-1), ['hide', 'loraImportModal']); +}); + +test('admin settings save shows why config.json refused it', async () => { + const message = 'config.json cannot be read, so nothing was saved. Fix or remove the file, then save again.'; + const { page, notifications } = load('admin.js', 'PromptAdmin', + async () => jsonResponse(409, { success: false, persisted: false, code: 'config-unreadable', error: message })); + page.settings = {}; + page._collectScanPaths = () => []; + page.saveLoraSettings = async () => assert.fail('nothing else is saved after a refusal'); + + await page.saveSettings(); + + assert.deepEqual(notifications, [['error', `Failed to save settings: ${message}`]]); +}); + +test('admin settings save keeps its plain message when the reply has no reason', async () => { + const { page, notifications } = load('admin.js', 'PromptAdmin', async () => ({ ok: false, status: 500, + json: async () => { throw new SyntaxError('not json'); } })); + page.settings = {}; + page._collectScanPaths = () => []; + + await page.saveSettings(); + + assert.deepEqual(notifications, [['error', 'Failed to save settings']]); +}); + +test('admin maintenance names the job that refused it', async () => { + const { page, dom, notifications } = load('admin.js', 'PromptAdmin', + async () => jsonResponse(409, { success: false, error: BUSY }), { + escapeHtml: text => String(text), + document: { readyState: 'loading', addEventListener() {}, getElementById: id => dom.getElementById(id), + querySelectorAll: selector => (selector === '.maintenance-option:checked' ? [{ id: 'vacuum' }] : []) }, + }); + + await page.runMaintenance(); + + const results = dom.nodes.maintenanceResults.innerHTML; + assert.match(results, /Maintenance Failed/); + assert.match(results, new RegExp(BUSY)); + assert.doesNotMatch(results, /HTTP 409/); + assert.deepEqual(notifications, [['error', '❌ Maintenance failed']]); + assert.equal(dom.nodes.runMaintenanceBtn.disabled, false); +}); diff --git a/tests/js/workspace-activity-store.test.js b/tests/js/workspace-activity-store.test.js index a562fec..5fce2cd 100644 --- a/tests/js/workspace-activity-store.test.js +++ b/tests/js/workspace-activity-store.test.js @@ -261,8 +261,13 @@ test('summaries are validated and the copy follows the spec', async () => { const loose = parseJob({ id: ID_A, state: 'running', progress: 140.4, processed: -1, total: 1.5, counters: { added: 'x', errors: 2 }, title: ' ', started_at: 'x', scope: [] }); assert.deepEqual(loose, { id: ID_A, kind: '', title: 'Background job', state: 'running', started_at: null, finished_at: null, - progress: 100, status_line: '', processed: 0, total: 0, counters: { added: 0, errors: 2 }, scope: {}, listed: {}, log_length: 0 }); + progress: 100, status_line: '', processed: 0, total: 0, counters: { added: 0, errors: 2 }, scope: {}, listed: {}, log_length: 0, + cancellable: true }); assert.equal(parseJob(summary({ progress: -3 })).progress, 0); + // A held legacy job cannot be cancelled; an older server sends no key, which keeps Cancel. + assert.equal(parseJob(summary({ cancellable: false })).cancellable, false); + assert.equal(parseJob(summary({ cancellable: true })).cancellable, true); + assert.equal(parseJob(summary()).cancellable, true); const scan = parseJob(summary({ progress: 42, processed: 8240, total: 19622, counters: { added: 1, errors: 1 } })); assert.deepEqual(pillParts(scan), { label: 'Scanning · 42%', files: ' · 8,240 / 19,622', text: 'Scanning · 42% · 8,240 / 19,622' }); diff --git a/tests/lock_probe.py b/tests/lock_probe.py new file mode 100644 index 0000000..88146f2 --- /dev/null +++ b/tests/lock_probe.py @@ -0,0 +1,35 @@ +"""A lock stand-in that reports when a second caller has to wait for it. + +Concurrency tests patch it over a module's lock so they can wait until the +second caller is really blocked on the lock, instead of hoping a thread +start or one event-loop tick got it there. +""" + +import threading + + +class ProbeLock: + """A re-entrant lock whose ``contended`` event is set when a caller blocks.""" + + def __init__(self): + self._lock = threading.RLock() + self.contended = threading.Event() + + def acquire(self, blocking=True, timeout=-1): + if self._lock.acquire(blocking=False): + return True + self.contended.set() + if not blocking: + return False + return self._lock.acquire(timeout=timeout) + + def release(self): + self._lock.release() + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, *exc): + self.release() + return False diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index bddcc39..61d703b 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -5,6 +5,7 @@ database and a stub ``folder_paths`` module, so the tests run without ComfyUI. """ import asyncio +import hashlib import json import os import shutil @@ -734,9 +735,9 @@ class TestSettingsReadMonitorThroughModule(AdminAPITestCase): def test_no_sys_modules_scan(self): import inspect - from py.api.admin import AdminRoutesMixin + from py.api.settings_routes import SettingsRoutesMixin - source = inspect.getsource(AdminRoutesMixin.get_settings) + source = inspect.getsource(SettingsRoutesMixin.get_settings) self.assertNotIn("sys.modules", source) async def test_reports_directories_of_running_monitor(self): @@ -1059,14 +1060,29 @@ class TestSettingsMisc(AdminAPITestCase): saved["gallery"]["monitoring"]["directories"], [self._root(self.output_dir)] ) - async def test_save_replaces_a_corrupt_config_file(self): - with open(self.config_path, "w") as f: - f.write("{not json") + async def test_save_refuses_a_corrupt_config_file(self): + # #220: a config.json that cannot be parsed is the user's file, never + # silently replaced; the save is refused and nothing is published. + Path(self.config_path).write_bytes(b"{not json") + before_timeout = PromptManagerConfig.RESULT_TIMEOUT resp = await self.client.request( "POST", "/prompt_manager/settings", json={"result_timeout": 4} ) - self.assertEqual(resp.status, 200) - self.assertEqual(self._read_config()["web_ui"]["result_timeout"], 4) + body = await resp.json() + self.assertEqual((resp.status, body["code"]), (409, "config-unreadable")) + self.assertIs(body["success"], False) + self.assertIs(body["persisted"], False) + self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, before_timeout) + self.assertEqual(Path(self.config_path).read_bytes(), b"{not json") + + async def test_get_settings_reports_an_unreadable_config(self): + Path(self.config_path).write_bytes(b"{not json") + resp = await self.client.request("GET", "/prompt_manager/settings") + body = await resp.json() + self.assertEqual(resp.status, 200) # the legacy page still loads + self.assertIsNone(body["revision"]) + self.assertEqual(body["config_error"], "config-unreadable") + self.assertIn("result_timeout", body["settings"]) async def test_saved_config_file_is_private_to_the_user(self): import stat @@ -1107,12 +1123,156 @@ class TestSettingsMisc(AdminAPITestCase): self.assertEqual(resp.status, 200) self.assertFalse(data["restart_required"]) - async def test_unwritable_config_path_still_succeeds(self): - os.environ[CONFIG_PATH_ENV] = os.path.join(self.tmpdir, "missing", "c.json") + async def test_unwritable_config_path_reports_failure_without_runtime_change(self): + # #220: the parent is a file, so config.json cannot be written. The + # save reports the failure and publishes nothing. + blocker = Path(self.tmpdir) / "not-a-dir" + blocker.write_text("x") + os.environ[CONFIG_PATH_ENV] = str(blocker / "c.json") + GalleryConfig.MONITORING_DIRECTORIES = [] + before_timeout = PromptManagerConfig.RESULT_TIMEOUT + with self.assertLogs("prompt_manager.api", level="ERROR"): + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={ + "result_timeout": before_timeout + 1, + "gallery_root_paths": [self._root(self.output_dir)], + }, + ) + body = await resp.json() + self.assertEqual( + (resp.status, body["code"], body["persisted"]), + (500, "config-write-failed", False), + ) + self.assertIs(body["success"], False) + self.assertIs(body["runtime_applied"], False) + self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, before_timeout) + self.assertEqual(GalleryConfig.MONITORING_DIRECTORIES, []) + self.assertEqual(blocker.read_text(), "x") + + async def test_save_reports_the_commit(self): resp = await self.client.request( "POST", "/prompt_manager/settings", json={"result_timeout": 3} ) + body = await resp.json() self.assertEqual(resp.status, 200) + self.assertEqual(body["message"], "Settings saved successfully") + self.assertIs(body["persisted"], True) + self.assertIs(body["runtime_applied"], True) + self.assertIs(body["restart_required"], False) + self.assertEqual(body["restart_required_fields"], []) + digest = hashlib.sha256(Path(self.config_path).read_bytes()).hexdigest() + self.assertEqual(body["revision"], "sha256:" + digest) + + async def test_changed_roots_name_the_restart_field(self): + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={"gallery_root_paths": [self._root(self.output_dir)]}, + ) + body = await resp.json() + self.assertEqual(resp.status, 200) + self.assertIs(body["restart_required"], True) + self.assertEqual(body["restart_required_fields"], ["gallery_root_paths"]) + + async def test_runtime_publication_failure_after_commit_is_reported(self): + # The values are durable; the response must not claim they are live. + self.api.reset_gallery_roots_cache = _raise + with self.assertLogs("prompt_manager.api", level="ERROR"): + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={"gallery_root_paths": [self._root(self.output_dir)]}, + ) + body = await resp.json() + self.assertEqual(resp.status, 200) + self.assertIs(body["persisted"], True) + self.assertIs(body["runtime_applied"], False) + self.assertIs(body["restart_required"], True) + self.assertEqual(body["restart_required_fields"], ["gallery_root_paths"]) + self.assertEqual( + self._read_config()["gallery"]["monitoring"]["directories"], + [self._root(self.output_dir)], + ) + + async def test_get_settings_reports_revision(self): + resp = await self.client.request("GET", "/prompt_manager/settings") + body = await resp.json() + self.assertEqual(body["revision"], "absent") + self.assertNotIn("config_error", body) + + self._write_config({"web_ui": {"result_timeout": 2}}) + body = await ( + await self.client.request("GET", "/prompt_manager/settings") + ).json() + digest = hashlib.sha256(Path(self.config_path).read_bytes()).hexdigest() + self.assertEqual(body["revision"], "sha256:" + digest) + + async def test_native_save_with_stale_revision_is_409(self): + self._write_config({"web_ui": {"result_timeout": 2}}) + body = await ( + await self.client.request("GET", "/prompt_manager/settings") + ).json() + seen = body["revision"] + self._write_config({"web_ui": {"result_timeout": 6}}) # another writer + before = Path(self.config_path).read_bytes() + before_timeout = PromptManagerConfig.RESULT_TIMEOUT + + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={"result_timeout": 9, "expected_revision": seen}, + ) + body = await resp.json() + self.assertEqual((resp.status, body["code"]), (409, "settings-changed")) + self.assertIs(body["persisted"], False) + digest = hashlib.sha256(before).hexdigest() + self.assertEqual(body["revision"], "sha256:" + digest) + self.assertEqual(Path(self.config_path).read_bytes(), before) + self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, before_timeout) + + async def test_native_save_with_current_revision_commits(self): + body = await ( + await self.client.request("GET", "/prompt_manager/settings") + ).json() + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={"result_timeout": 9, "expected_revision": body["revision"]}, + ) + self.assertEqual(resp.status, 200) + self.assertEqual(self._read_config()["web_ui"]["result_timeout"], 9) + + async def test_expected_revision_must_be_a_string(self): + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={"result_timeout": 9, "expected_revision": 5}, + ) + self.assertEqual(resp.status, 400) + self.assertFalse(os.path.exists(self.config_path)) + + async def test_legacy_partial_update_without_revision_still_merges(self): + self._write_config( + { + "integrations": {"lora_manager": {"civitai_api_key": "secret"}}, + "web_ui": {"result_timeout": 1, "webui_display_mode": "popup"}, + "custom_section": {"keep": True}, + } + ) + resp = await self.client.request( + "POST", "/prompt_manager/settings", json={"result_timeout": 8} + ) + self.assertEqual(resp.status, 200) + self.assertNotIn("secret", await resp.text()) + saved = self._read_config() + self.assertEqual( + saved["integrations"]["lora_manager"]["civitai_api_key"], "secret" + ) + self.assertEqual(saved["custom_section"], {"keep": True}) + self.assertEqual(saved["web_ui"]["result_timeout"], 8) + self.assertEqual(saved["web_ui"]["webui_display_mode"], "popup") # not sent async def test_save_unexpected_error_is_500(self): self.api._parse_gallery_roots = _raise diff --git a/tests/test_api_backup_restore.py b/tests/test_api_backup_restore.py index d91b302..aa211ff 100644 --- a/tests/test_api_backup_restore.py +++ b/tests/test_api_backup_restore.py @@ -5,12 +5,16 @@ stream the upload to disk, reject anything that is not a healthy prompts database and leave the live database untouched when it does. """ +import asyncio import glob +import hashlib import os import sqlite3 import sys import tempfile +import threading import unittest +from pathlib import Path from unittest.mock import MagicMock, patch _mock_server = MagicMock() @@ -142,6 +146,85 @@ class TestBackupRoute(BackupRestoreTestCase): self.assertIn("backup", data["error"].lower()) +class TestRestoreHoldsTheRegistry(BackupRestoreTestCase): + """A legacy restore and a workspace job refuse each other (#269 gap).""" + + def _digest(self): + return { + suffix: hashlib.sha256(Path(self.db_path + suffix).read_bytes()).hexdigest() + for suffix in ("", "-wal") + if os.path.exists(self.db_path + suffix) + } + + async def test_legacy_restore_refused_while_scan_job_runs(self): + self._save("keep me") + upload = self._make_db_file(["restored"]) + gate = asyncio.Event() + + async def runner(job): + await gate.wait() + + scan = self.api._jobs.start( + "scan", "Scan for metadata", record_changing=True, scope={}, runner=runner + ) + before = self._digest() + try: + resp = await self._restore(upload) + self.assertEqual(resp.status, 409) + self.assertEqual( + await resp.json(), + {"success": False, "error": "Scan for metadata already running"}, + ) + self.assertEqual(self._digest(), before) + self.assertEqual( + [p["text"] for p in self.api.db.search_prompts()], ["keep me"] + ) + self.assertEqual(glob.glob(self.db_path + ".backup_*"), []) + finally: + gate.set() + await asyncio.wait_for(scan.task, 5) + + async def test_scan_refused_while_legacy_restore_holds(self): + upload = self._make_db_file(["restored"]) + entered = threading.Event() + release = threading.Event() + model = self.api.db.model + real_verify = model.verify_database_file + + def gated(path, **kwargs): + entered.set() + release.wait(10) + return real_verify(path, **kwargs) + + loop = asyncio.get_running_loop() + with patch.object(model, "verify_database_file", gated): + restore = asyncio.ensure_future(self._restore(upload)) + self.assertTrue(await loop.run_in_executor(None, entered.wait, 10)) + resp = await self.client.post("/prompt_manager/jobs/scan") + body = await resp.json() + listed = (await (await self.client.get("/prompt_manager/jobs")).json())[ + "jobs" + ] + release.set() + restored = await restore + + self.assertEqual((resp.status, body["code"]), (409, "busy")) + self.assertEqual(body["running"]["title"], "Database restore (admin page)") + self.assertEqual( + [(j["kind"], j["state"], j["cancellable"]) for j in listed], + [("restore", "running", False)], + ) + self.assertEqual(restored.status, 200) + self.assertEqual(self.api._jobs.list()[0].state, "finished") + self.assertIsNone(self.api._jobs.active("restore")) + + async def test_rejected_upload_ends_its_hold_failed(self): + resp = await self._restore(self._write("notes.txt", b"not a database")) + self.assertEqual(resp.status, 400) + job = self.api._jobs.list()[0] + self.assertEqual((job.kind, job.state), ("restore", "failed")) + + class TestRestoreRoute(BackupRestoreTestCase): async def test_restore_replaces_database_and_reports_backup(self): self._save("old prompt") diff --git a/tests/test_api_images.py b/tests/test_api_images.py index 18062bb..8c0361c 100644 --- a/tests/test_api_images.py +++ b/tests/test_api_images.py @@ -5,11 +5,13 @@ Uses aiohttp's test client against a real PromptManagerAPI wired to a temporary SQLite database and a temporary output directory. """ +import asyncio import json import os import pathlib import sys import tempfile +import threading import unittest from contextlib import ExitStack from pathlib import Path @@ -638,6 +640,93 @@ class TestGenerateThumbnailsPost(ThumbnailTestCase): self.assertEqual(data["total_images"], 0) +class TestThumbnailRoutesHoldTheRegistry(ThumbnailTestCase): + """Thumbnail work is ``guarded``: never beside a record-changing job.""" + + def _run_scan(self): + gate = asyncio.Event() + + async def runner(job): + await gate.wait() + + job = self.api._jobs.start( + "scan", "Scan for metadata", admission="records", scope={}, runner=runner + ) + + async def release(): + gate.set() + await asyncio.wait_for(job.task, 5) + + self.addAsyncCleanup(release) + return job + + async def test_thumbnail_generate_refused_while_scan_runs(self): + make_png(self.output_dir / "a.png") + self._run_scan() + + resp = await self.client.request( + "POST", "/prompt_manager/images/generate-thumbnails", json={} + ) + self.assertEqual(resp.status, 409) + self.assertEqual( + await resp.json(), + {"success": False, "error": "Scan for metadata already running"}, + ) + + resp = await self.client.request("POST", self.PROGRESS_URL) + text = await resp.text() + self.assertEqual(resp.status, 200) # the gallery reads an SSE stream + self.assertEqual( + text, + "event: error\ndata: " + + json.dumps( + { + "error": "Scan for metadata already running", + "message": "Scan for metadata already running", + } + ) + + "\n\n", + ) + + thumbs = self.output_dir / "thumbnails" + make_png(thumbs / "b_thumb.png") + resp = await self.client.request( + "POST", "/prompt_manager/images/clear-thumbnails" + ) + self.assertEqual(resp.status, 409) + self.assertFalse(self._thumb_path("a").exists()) + self.assertTrue((thumbs / "b_thumb.png").exists()) + + async def test_scan_refused_while_thumbnails_generate(self): + make_png(self.output_dir / "a.png") + entered = threading.Event() + release = threading.Event() + real = self.api._generate_gallery_thumbnails_sync + + def gated(*args): + entered.set() + release.wait(10) + return real(*args) + + self.api._generate_gallery_thumbnails_sync = gated + loop = asyncio.get_running_loop() + generate = asyncio.ensure_future( + self.client.request( + "POST", "/prompt_manager/images/generate-thumbnails", json={} + ) + ) + self.assertTrue(await loop.run_in_executor(None, entered.wait, 10)) + resp = await self.client.request("POST", "/prompt_manager/jobs/scan") + body = await resp.json() + self.assertEqual((resp.status, body["code"]), (409, "busy")) + self.assertEqual( + body["running"]["title"], "Thumbnail generation (gallery page)" + ) + release.set() + self.assertEqual((await generate).status, 200) + self.assertTrue(self._thumb_path("a").exists()) + + class TestRecentImagesCaps(ImageAPITestCase): """Every path through /images/recent hands the DB a bounded window.""" diff --git a/tests/test_api_lora.py b/tests/test_api_lora.py index 36b9955..810671b 100644 --- a/tests/test_api_lora.py +++ b/tests/test_api_lora.py @@ -4,11 +4,13 @@ Filesystem detection and network downloads are patched out so the tests run anywhere without a LoraManager install or a CivitAI connection. """ +import asyncio import json import os import shutil import sys import tempfile +import threading import types import unittest from pathlib import Path @@ -34,11 +36,20 @@ from aiohttp.test_utils import AioHTTPTestCase from database.operations import PromptDatabase from py.api import PromptManagerAPI -from py import lora_utils +from py import config_store, lora_utils from py.config import IntegrationConfig, PromptManagerConfig +from py.api import settings_routes as settings_module +from py.api import lora_integration as lora_module from py.lora_utils import get_trigger_cache from utils.hashing import generate_prompt_hash +try: + from tests.lock_probe import ProbeLock + from tests.open_handles import assert_closed +except ImportError: # discovered with tests/ as the top-level directory + from lock_probe import ProbeLock + from open_handles import assert_closed + SECRET_KEY = "sk-civitai-super-secret-0123456789" @@ -56,7 +67,13 @@ class LoraAPITestCase(AioHTTPTestCase): return_value=[self.custom_nodes], ) self._custom_nodes_patch.start() - self._real_save = PromptManagerConfig.save_to_file + # Never the repository's config.json (live user data): every save in + # these tests goes to a file in this test's temp directory. + self.config_file = self.tmp_root / "config.json" + self._env_patch = patch.dict( + os.environ, {"PROMPT_MANAGER_CONFIG_PATH": str(self.config_file)} + ) + self._env_patch.start() self._config_patch = patch.multiple( IntegrationConfig, @@ -70,8 +87,6 @@ class LoraAPITestCase(AioHTTPTestCase): "py.lora_utils.detect_lora_manager", return_value=None ) self.detect_mock = self._detect_patch.start() - self._save_patch = patch.object(PromptManagerConfig, "save_to_file") - self.save_mock = self._save_patch.start() get_trigger_cache().clear() app = web.Application() @@ -83,7 +98,7 @@ class LoraAPITestCase(AioHTTPTestCase): async def tearDownAsync(self): await super().tearDownAsync() # closes the aiohttp test client - self._save_patch.stop() + self._env_patch.stop() self._detect_patch.stop() self._config_patch.stop() self._custom_nodes_patch.stop() @@ -96,8 +111,19 @@ class LoraAPITestCase(AioHTTPTestCase): ): if os.path.exists(path): os.unlink(path) + assert_closed(self, self.tmp_root) shutil.rmtree(self.tmp_root, ignore_errors=True) + def _saved(self): + return json.loads(self.config_file.read_text()) + + def _make_unwritable(self): + """Point the config path under a regular file, so no write can succeed.""" + blocker = self.tmp_root / "not-a-dir" + blocker.write_text("x") + self.config_file = blocker / "config.json" + os.environ["PROMPT_MANAGER_CONFIG_PATH"] = str(self.config_file) + async def _post_json(self, url, body): return await self.client.request( "POST", @@ -250,7 +276,7 @@ class TestLoraEnableRoute(LoraAPITestCase): ) self.assertEqual(resp.status, 400) self.assertFalse(IntegrationConfig.LORA_MANAGER_ENABLED) - self.save_mock.assert_not_called() + self.assertFalse(self.config_file.exists()) async def test_enable_persists_and_loads_trigger_cache(self): lm = _make_lora_manager(self.custom_nodes) @@ -285,7 +311,15 @@ class TestLoraEnableRoute(LoraAPITestCase): os.path.normcase(os.path.realpath(str(lm))), ) self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, SECRET_KEY) - self.save_mock.assert_called_once_with() + saved = self._saved()["integrations"]["lora_manager"] + self.assertEqual(saved["civitai_api_key"], SECRET_KEY) + self.assertEqual(saved["path"], IntegrationConfig.LORA_MANAGER_PATH) + self.assertIs(saved["enabled"], True) + self.assertIs(saved["trigger_words_enabled"], True) + self.assertEqual( + (data["persisted"], data["runtime_applied"], data["cache_loaded"]), + (True, True, True), + ) self.assertTrue(get_trigger_cache().is_loaded) self.assertEqual(get_trigger_cache().get_trigger_words("neon"), ["glow"]) @@ -319,7 +353,7 @@ class TestLoraEnableRoute(LoraAPITestCase): self.assertFalse(json.loads(body)["success"]) self.assertNotIn(str(self.tmp_root), body) self.assertFalse(IntegrationConfig.LORA_MANAGER_ENABLED) - self.save_mock.assert_not_called() + self.assertFalse(self.config_file.exists()) async def test_enable_with_custom_node_lacking_lora_in_its_name_is_400(self): lm = _make_lora_manager(self.custom_nodes).parent / "ComfyUI-Other" @@ -344,7 +378,6 @@ class TestLoraEnableRoute(LoraAPITestCase): async def test_enable_writes_the_configured_config_file(self): config_file = self.tmp_root / "cfg" / "config.json" - self.save_mock.side_effect = self._real_save with patch.dict(os.environ, {"PROMPT_MANAGER_CONFIG_PATH": str(config_file)}): resp = await self._post_json( @@ -353,7 +386,6 @@ class TestLoraEnableRoute(LoraAPITestCase): ) self.assertEqual(resp.status, 200) - self.save_mock.assert_called_once_with() self.assertTrue(config_file.is_file()) saved = json.loads(config_file.read_text())["integrations"]["lora_manager"] self.assertFalse(saved["enabled"]) @@ -414,9 +446,299 @@ class TestLoraEnableRoute(LoraAPITestCase): self.assertEqual(resp.status, 400) async def test_persist_failure_is_500(self): - self.save_mock.side_effect = OSError("read-only") - resp = await self._post_json("/prompt_manager/lora/enable", {"enabled": False}) + self._make_unwritable() + with self.assertLogs("prompt_manager.api", level="ERROR"): + resp = await self._post_json( + "/prompt_manager/lora/enable", {"enabled": False} + ) self.assertEqual(resp.status, 500) + self.assertEqual((await resp.json())["code"], "config-write-failed") + + async def test_enable_write_failure_reports_and_keeps_runtime(self): + # #220: a save that cannot be written publishes nothing and says so. + lm = _make_lora_manager(self.custom_nodes) + self.detect_mock.return_value = str(lm) + IntegrationConfig.CIVITAI_API_KEY = SECRET_KEY + self._make_unwritable() + + with self.assertLogs("prompt_manager.api", level="ERROR"): + resp = await self._post_json( + "/prompt_manager/lora/enable", + { + "enabled": True, + "path": str(lm), + "trigger_words_enabled": True, + "clear_civitai_api_key": True, + }, + ) + + text = await resp.text() + body = json.loads(text) + self.assertEqual(resp.status, 500) + self.assertEqual(body["code"], "config-write-failed") + self.assertEqual((body["persisted"], body["runtime_applied"]), (False, False)) + self.assertFalse(IntegrationConfig.LORA_MANAGER_ENABLED) + self.assertEqual(IntegrationConfig.LORA_MANAGER_PATH, "") + self.assertFalse(IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED) + self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, SECRET_KEY) + self.assertFalse(get_trigger_cache().is_loaded) + self.assertNotIn(SECRET_KEY, text) + self.assertNotIn(str(self.tmp_root), text) + + async def test_enable_preserves_unknown_sections(self): + stored = "stored-value" # a plain marker: what is on disk must stay + self.config_file.write_text( + json.dumps( + { + "custom_section": {"keep": True}, + "web_ui": {"result_timeout": 11}, + "integrations": { + "lora_manager": {"civitai_api_key": stored, "extra": 1}, + "other_tool": {"on": True}, + }, + } + ) + ) + IntegrationConfig.CIVITAI_API_KEY = stored + + resp = await self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "trigger_words_enabled": True}, + ) + + self.assertEqual(resp.status, 200) + saved = self._saved() + self.assertEqual(saved["custom_section"], {"keep": True}) + self.assertEqual(saved["web_ui"], {"result_timeout": 11}) + self.assertEqual(saved["integrations"]["other_tool"], {"on": True}) + lora = saved["integrations"]["lora_manager"] + self.assertEqual(lora["extra"], 1) + self.assertEqual(lora["civitai_api_key"], stored) + self.assertIs(lora["trigger_words_enabled"], True) + self.assertNotIn("database", saved) # no runtime defaults written + + async def test_enable_refuses_a_corrupt_config_file(self): + self.config_file.write_bytes(b"{not json") + resp = await self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "trigger_words_enabled": True}, + ) + body = await resp.json() + self.assertEqual((resp.status, body["code"]), (409, "config-unreadable")) + self.assertEqual(self.config_file.read_bytes(), b"{not json") + self.assertFalse(IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED) + + async def test_enable_requires_boolean_flags_and_body_cap(self): + for body in ( + {"enabled": "yes"}, + {"enabled": False, "trigger_words_enabled": 1}, + {"enabled": False, "clear_civitai_api_key": "true"}, + ): + with self.subTest(body=body): + resp = await self._post_json("/prompt_manager/lora/enable", body) + self.assertEqual(resp.status, 400) + resp = await self._post_json("/prompt_manager/lora/enable", [True]) + self.assertEqual(resp.status, 400) + resp = await self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "civitai_api_key": "k" * (2 * 1024 * 1024)}, + ) + self.assertEqual(resp.status, 413) + self.assertFalse(self.config_file.exists()) + self.assertFalse(IntegrationConfig.LORA_MANAGER_ENABLED) + + async def test_blank_key_unchanged_and_explicit_clear(self): + self.config_file.write_text( + json.dumps({"integrations": {"lora_manager": {"civitai_api_key": "old"}}}) + ) + IntegrationConfig.CIVITAI_API_KEY = "old" + + resp = await self._post_json( + "/prompt_manager/lora/enable", {"enabled": False, "civitai_api_key": " "} + ) + self.assertEqual(resp.status, 200) + self.assertEqual( + self._saved()["integrations"]["lora_manager"]["civitai_api_key"], "old" + ) + self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, "old") + + resp = await self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "clear_civitai_api_key": True}, + ) + self.assertEqual(resp.status, 200) + self.assertEqual( + self._saved()["integrations"]["lora_manager"]["civitai_api_key"], "" + ) + self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, "") + + async def test_cache_load_failure_reports_persisted_but_unloaded(self): + lm = _make_lora_manager(self.custom_nodes) + self.detect_mock.return_value = str(lm) + + with ( + patch.object( + type(get_trigger_cache()), "load", side_effect=OSError("unreadable") + ), + self.assertLogs("prompt_manager.api", level="WARNING"), + ): + resp = await self._post_json( + "/prompt_manager/lora/enable", + {"enabled": True, "path": str(lm), "trigger_words_enabled": True}, + ) + + text = await resp.text() + body = json.loads(text) + self.assertEqual(resp.status, 200) + self.assertIs(body["success"], True) + self.assertEqual((body["persisted"], body["cache_loaded"]), (True, False)) + self.assertTrue(IntegrationConfig.LORA_MANAGER_ENABLED) + self.assertIs(self._saved()["integrations"]["lora_manager"]["enabled"], True) + self.assertNotIn("unreadable", text) + + +class TestConfigWritersShareOneStore(LoraAPITestCase): + + async def asyncSetUp(self): + await super().asyncSetUp() + self._orig_timeout = PromptManagerConfig.RESULT_TIMEOUT + + async def asyncTearDown(self): + PromptManagerConfig.RESULT_TIMEOUT = self._orig_timeout + await super().asyncTearDown() + + async def test_settings_and_lora_saves_do_not_overwrite_each_other(self): + # #220: the LoRA save is sent while the settings save is inside its + # commit (lock held, file already read). The LoRA commit re-reads the + # file under the same lock, so both edits survive and nothing else + # is lost (before, the LoRA save rewrote the whole runtime config). + self.config_file.write_text(json.dumps({"custom_section": {"keep": True}})) + inside = threading.Event() + release = threading.Event() + real_mutation = settings_module._settings_mutation + + def gated_mutation(updates, new_roots): + mutate = real_mutation(updates, new_roots) + + def gated(config): + inside.set() + release.wait(10) + return mutate(config) + + return gated + + lora_calls = [] + real_lora_mutation = lora_module._lora_settings_mutation + + def recorded_lora_mutation(values): + mutate = real_lora_mutation(values) + + def recorded(config): + lora_calls.append(dict(config)) + return mutate(config) + + return recorded + + probe = ProbeLock() + loop = asyncio.get_running_loop() + with ( + patch.object(settings_module, "_settings_mutation", gated_mutation), + patch.object( + lora_module, "_lora_settings_mutation", recorded_lora_mutation + ), + patch.object(config_store, "_COMMIT_LOCK", probe), + ): + settings = asyncio.ensure_future( + self._post_json("/prompt_manager/settings", {"result_timeout": 13}) + ) + self.assertTrue(await loop.run_in_executor(None, inside.wait, 10)) + lora = asyncio.ensure_future( + self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "trigger_words_enabled": True}, + ) + ) + # Wait until the LoRA commit is blocked on the lock the settings + # commit holds; its mutation must not have run yet. + self.assertTrue(await loop.run_in_executor(None, probe.contended.wait, 10)) + self.assertEqual(lora_calls, []) + release.set() + settings_resp, lora_resp = await asyncio.gather(settings, lora) + self.assertEqual(len(lora_calls), 1) + self.assertEqual(lora_calls[0]["web_ui"]["result_timeout"], 13) + + self.assertEqual((settings_resp.status, lora_resp.status), (200, 200)) + saved = self._saved() + self.assertEqual(saved["custom_section"], {"keep": True}) + self.assertEqual(saved["web_ui"]["result_timeout"], 13) + self.assertIs( + saved["integrations"]["lora_manager"]["trigger_words_enabled"], True + ) + settings_body = await settings_resp.json() + lora_body = await lora_resp.json() + self.assertNotEqual(settings_body["revision"], lora_body["revision"]) + + def _delay_first_commit_result(self): + """Hold back the first config commit's result from its handler. + + The commit itself has run (the file is written); only the handler's + continuation waits until the returned event is set. Before #220's fix + round the handler published after this point, so a later save could + publish first and the earlier one overwrite it at runtime. + """ + original = self.api._run_in_executor + written = asyncio.Event() + resume = asyncio.Event() + delayed = [] + + async def run(func, *args, **kwargs): + result = await original(func, *args, **kwargs) + if getattr(func, "__func__", None) is config_store.ConfigStore.commit: + if not delayed: + delayed.append(func) + written.set() + await resume.wait() + return result + + patcher = patch.object(self.api, "_run_in_executor", run) + patcher.start() + self.addCleanup(patcher.stop) + return written, resume + + async def test_concurrent_settings_saves_end_on_the_last_written(self): + written, resume = self._delay_first_commit_result() + first = asyncio.ensure_future( + self._post_json("/prompt_manager/settings", {"result_timeout": 11}) + ) + await asyncio.wait_for(written.wait(), 10) + second = await self._post_json( + "/prompt_manager/settings", {"result_timeout": 22} + ) + resume.set() + first = await first + self.assertEqual((first.status, second.status), (200, 200)) + self.assertEqual(self._saved()["web_ui"]["result_timeout"], 22) + self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, 22) + + async def test_concurrent_lora_saves_end_on_the_last_written(self): + written, resume = self._delay_first_commit_result() + first = asyncio.ensure_future( + self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "trigger_words_enabled": True}, + ) + ) + await asyncio.wait_for(written.wait(), 10) + second = await self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "trigger_words_enabled": False}, + ) + resume.set() + first = await first + self.assertEqual((first.status, second.status), (200, 200)) + lora = self._saved()["integrations"]["lora_manager"] + self.assertIs(lora["trigger_words_enabled"], False) + self.assertIs(IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED, False) class TestLoraScanRoute(LoraAPITestCase): @@ -445,6 +767,12 @@ class TestLoraScanRoute(LoraAPITestCase): body = await resp.text() self.assertNotIn(str(self.tmp_root), body) self.assertIn("x.db", body) + # The admin page's modal shows the failure as its status and closes. + error = self._sse_events(body)[-1] + self.assertEqual( + (error["type"], error["status"], error["progress"]), + ("error", error["message"], 0), + ) @staticmethod def _sse_events(text): diff --git a/tests/test_config_store.py b/tests/test_config_store.py new file mode 100644 index 0000000..5099dbb --- /dev/null +++ b/tests/test_config_store.py @@ -0,0 +1,419 @@ +"""Tests for the serialized config store (py/config_store.py, #220). + +Every test works on a config.json inside its own temporary directory; the +repository's config.json (live user data) is never read or written. +""" + +import hashlib +import json +import os +import shutil +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +_mock_server = MagicMock() +_mock_server.PromptServer.instance.routes = MagicMock() +sys.modules.setdefault("server", _mock_server) + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from py import config_store # noqa: E402 +from py.config_store import ( # noqa: E402 + ABSENT_REVISION, + ConfigStore, + ConfigUnreadable, + RevisionConflict, + get_config_store, + revision_of, +) + +try: + from tests.lock_probe import ProbeLock # noqa: E402 + from tests.open_handles import assert_closed # noqa: E402 +except ImportError: # discovered with tests/ as the top-level directory + from lock_probe import ProbeLock # noqa: E402 + from open_handles import assert_closed # noqa: E402 + + +class ConfigStoreTests(unittest.TestCase): + def setUp(self): + self.tmpdir = os.path.realpath(tempfile.mkdtemp()) + self.path = os.path.join(self.tmpdir, "config.json") + self.store = ConfigStore(lambda: self.path) + + def tearDown(self): + try: + assert_closed(self, self.tmpdir) + finally: + shutil.rmtree(self.tmpdir, ignore_errors=True) + + def _write(self, data): + Path(self.path).write_text(json.dumps(data), encoding="utf-8") + + def _bytes(self): + return Path(self.path).read_bytes() + + def _leftovers(self): + return sorted(name for name in os.listdir(self.tmpdir) if name != "config.json") + + # ── read ────────────────────────────────────────────────────────── + + def test_read_of_a_missing_file_is_empty_and_absent(self): + self.assertEqual(self.store.read(), ({}, ABSENT_REVISION)) + + def test_read_reports_the_sha256_of_the_file_bytes(self): + self._write({"web_ui": {"result_timeout": 3}}) + data, revision = self.store.read() + self.assertEqual(data, {"web_ui": {"result_timeout": 3}}) + expected = "sha256:" + hashlib.sha256(self._bytes()).hexdigest() + self.assertEqual(revision, expected) + self.assertEqual(revision_of(self._bytes()), expected) + + def test_non_object_config_is_unreadable(self): + self._write([1, 2]) + with self.assertRaises(ConfigUnreadable): + self.store.read() + + def test_unreadable_path_is_unreadable(self): + os.mkdir(self.path) # a directory where the file should be + with self.assertRaises(ConfigUnreadable): + self.store.read() + + # ── commit ──────────────────────────────────────────────────────── + + def test_commit_preserves_unknown_sections_and_secret(self): + self._write( + { + "custom_section": {"keep": True}, + "integrations": {"lora_manager": {"civitai_api_key": "secret"}}, + "web_ui": {"result_timeout": 1, "show_test_button": True}, + } + ) + + def mutate(config): + config.setdefault("web_ui", {})["result_timeout"] = 7 + return config + + saved, revision = self.store.commit(mutate, expected_revision=None) + + on_disk = json.loads(self._bytes()) + self.assertEqual(saved, on_disk) + self.assertEqual(on_disk["custom_section"], {"keep": True}) + self.assertEqual( + on_disk["integrations"]["lora_manager"]["civitai_api_key"], "secret" + ) + self.assertEqual( + on_disk["web_ui"], {"result_timeout": 7, "show_test_button": True} + ) + self.assertEqual(revision, revision_of(self._bytes())) + self.assertEqual(self.store.read(), (on_disk, revision)) + + def test_commit_creates_a_missing_file_and_its_parent(self): + self.path = os.path.join(self.tmpdir, "nested", "config.json") + saved, revision = self.store.commit( + lambda config: {**config, "a": 1}, expected_revision=ABSENT_REVISION + ) + self.assertEqual(saved, {"a": 1}) + self.assertEqual(json.loads(Path(self.path).read_text()), {"a": 1}) + self.assertNotEqual(revision, ABSENT_REVISION) + + def test_mutate_works_on_a_copy(self): + self._write({"web_ui": {"result_timeout": 1}}) + seen = {} + + def mutate(config): + seen["config"] = config + config["web_ui"]["result_timeout"] = 2 + raise RuntimeError("validation failed late") + + before = self._bytes() + with self.assertRaises(RuntimeError): + self.store.commit(mutate, expected_revision=None) + self.assertEqual(self._bytes(), before) + self.assertEqual(self.store.read()[0], {"web_ui": {"result_timeout": 1}}) + + def test_mutate_must_return_an_object(self): + self._write({"a": 1}) + before = self._bytes() + with self.assertRaises(TypeError): + self.store.commit(lambda config: None, expected_revision=None) + self.assertEqual(self._bytes(), before) + + def test_malformed_existing_config_is_refused_and_untouched(self): + Path(self.path).write_bytes(b"{not json") + calls = [] + with self.assertRaises(ConfigUnreadable): + self.store.commit( + lambda config: calls.append(config) or config, expected_revision=None + ) + self.assertEqual(calls, []) + self.assertEqual(self._bytes(), b"{not json") + self.assertEqual(self._leftovers(), []) + + def test_write_failure_leaves_bytes_and_raises(self): + self._write({"a": 1}) + before = self._bytes() + with patch.object(config_store.os, "fsync", side_effect=OSError("disk full")): + with self.assertRaises(OSError): + self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None) + self.assertEqual(self._bytes(), before) + self.assertEqual(self._leftovers(), []) + + def test_unwritable_directory_raises_oserror(self): + blocker = os.path.join(self.tmpdir, "afile") + Path(blocker).write_text("x") + self.path = os.path.join(blocker, "config.json") # parent is a file + with self.assertRaises(OSError): + self.store.commit(lambda c: {"a": 1}, expected_revision=None) + + def test_stale_revision_conflicts_without_write(self): + self._write({"a": 1}) + _data, revision = self.store.read() + self._write({"a": 2}) # another writer got there first + before = self._bytes() + calls = [] + with self.assertRaises(RevisionConflict) as caught: + self.store.commit( + lambda c: calls.append(c) or c, expected_revision=revision + ) + self.assertEqual(caught.exception.current, revision_of(before)) + self.assertEqual(calls, []) + self.assertEqual(self._bytes(), before) + + def test_matching_revision_commits(self): + self._write({"a": 1}) + _data, revision = self.store.read() + saved, _ = self.store.commit( + lambda c: {**c, "b": 2}, expected_revision=revision + ) + self.assertEqual(saved, {"a": 1, "b": 2}) + + def test_temp_file_removed_after_replace_failure(self): + self._write({"a": 1}) + before = self._bytes() + with patch.object( + config_store.os, "replace", side_effect=PermissionError("locked") + ): + with self.assertRaises(OSError): + self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None) + self.assertEqual(self._bytes(), before) + self.assertEqual(self._leftovers(), []) + + @unittest.skipUnless(os.name == "posix", "POSIX file modes") + def test_committed_file_is_private_to_the_user(self): + import stat + + self._write({"a": 1}) + os.chmod(self.path, 0o644) + self.store.commit(lambda c: c, expected_revision=None) + self.assertEqual(stat.S_IMODE(os.stat(self.path).st_mode), 0o600) + + def _probe_lock(self): + probe = ProbeLock() + patcher = patch.object(config_store, "_COMMIT_LOCK", probe) + patcher.start() + self.addCleanup(patcher.stop) + return probe + + def _hold_lock_in_a_commit(self, value="first"): + """Start a commit that holds the lock until the returned event is set.""" + inside = threading.Event() + release = threading.Event() + errors = [] + + def slow(config): + inside.set() + release.wait(10) + config[value] = 1 + return config + + def run(): + try: + self.store.commit(slow, expected_revision=None) + except Exception as exc: # pragma: no cover - reported by the test + errors.append(exc) + + thread = threading.Thread(target=run) + thread.start() + self.assertTrue(inside.wait(10)) + return thread, release, errors + + def test_concurrent_commits_serialize(self): + # The second commit is observed blocked on the lock before the first + # one may finish, so this fails whenever commit stops taking the lock. + self._write({"keep": True}) + probe = self._probe_lock() + first, release, errors = self._hold_lock_in_a_commit() + calls = [] + + def fast(config): + calls.append(dict(config)) + config["second"] = 2 + return config + + second = threading.Thread( + target=lambda: self.store.commit(fast, expected_revision=None) + ) + second.start() + self.assertTrue(probe.contended.wait(10), "second commit never waited") + self.assertEqual(calls, []) # not called while the first holds the lock + release.set() + first.join(10) + second.join(10) + + self.assertEqual(errors, []) + self.assertEqual(calls, [{"keep": True, "first": 1}]) # re-read inside + self.assertEqual( + json.loads(self._bytes()), {"keep": True, "first": 1, "second": 2} + ) + + def test_read_waits_for_a_running_commit(self): + # Windows cannot replace a file another handle has open, so a read + # runs under the commit lock and never overlaps a replace. + self._write({"keep": True}) + probe = self._probe_lock() + first, release, errors = self._hold_lock_in_a_commit() + seen = [] + reader = threading.Thread(target=lambda: seen.append(self.store.read())) + reader.start() + self.assertTrue(probe.contended.wait(10), "read did not take the lock") + self.assertEqual(seen, []) + release.set() + first.join(10) + reader.join(10) + self.assertEqual(errors, []) + self.assertEqual(seen[0][0], {"keep": True, "first": 1}) + self.assertEqual(seen[0][1], revision_of(self._bytes())) + + def test_replace_is_retried_after_a_transient_permission_error(self): + # Windows: an antivirus or indexer handle makes os.replace fail with + # PermissionError for a moment; the commit retries before failing. + self._write({"a": 1}) + real_replace = os.replace + attempts = [] + + def flaky(src, dst): + attempts.append(dst) + if len(attempts) == 1: + raise PermissionError("in use by another process") + return real_replace(src, dst) + + with ( + patch.object(config_store, "_REPLACE_ATTEMPTS", 5), + patch.object(config_store.os, "replace", side_effect=flaky), + patch.object(config_store.time, "sleep") as sleep, + ): + saved, _ = self.store.commit( + lambda c: {**c, "b": 2}, expected_revision=None + ) + self.assertEqual(saved, {"a": 1, "b": 2}) + self.assertEqual(json.loads(self._bytes()), {"a": 1, "b": 2}) + self.assertEqual(len(attempts), 2) + sleep.assert_called_once_with(config_store._REPLACE_BACKOFF_SECONDS) + self.assertEqual(self._leftovers(), []) + + def test_replace_gives_up_after_the_last_attempt(self): + self._write({"a": 1}) + before = self._bytes() + with ( + patch.object(config_store, "_REPLACE_ATTEMPTS", 3), + patch.object( + config_store.os, "replace", side_effect=PermissionError("locked") + ) as replace, + patch.object(config_store.time, "sleep") as sleep, + ): + with self.assertRaises(PermissionError): + self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None) + self.assertEqual((replace.call_count, sleep.call_count), (3, 2)) + self.assertEqual(self._bytes(), before) + self.assertEqual(self._leftovers(), []) + + def test_replace_is_tried_once_outside_windows(self): + expected = 5 if os.name == "nt" else 1 + self.assertEqual(config_store._REPLACE_ATTEMPTS, expected) + + @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks") + def test_symlinked_config_is_written_through_the_link(self): + # A config.json that is a symlink (or a single-file bind mount) is + # written beside its real target, so the link stays a link. + real_dir = os.path.join(self.tmpdir, "real") + os.mkdir(real_dir) + target = os.path.join(real_dir, "settings.json") + Path(target).write_text(json.dumps({"a": 1}), encoding="utf-8") + try: + os.symlink(target, self.path) + except (OSError, NotImplementedError): # pragma: no cover - Windows + self.skipTest("symlinks are not permitted here") + + saved, revision = self.store.commit( + lambda c: {**c, "b": 2}, expected_revision=None + ) + + self.assertTrue(os.path.islink(self.path)) + self.assertEqual(json.loads(Path(target).read_text()), {"a": 1, "b": 2}) + self.assertEqual(self.store.read(), (saved, revision)) + self.assertEqual(os.listdir(real_dir), ["settings.json"]) + + def test_publish_runs_under_the_lock_after_the_write(self): + # Runtimes publish inside the commit, so two saves publish in the + # order they were written and the runtime ends on the last one. + self._write({"a": 1}) + seen = [] + + # The lock is re-entrant, so another thread probes whether it is held. + def probe_publish(config): + result = [] + thread = threading.Thread( + target=lambda: result.append( + config_store._COMMIT_LOCK.acquire(blocking=False) + ) + ) + thread.start() + thread.join(10) + seen.append((config, json.loads(self._bytes()), result[0])) + + saved, _ = self.store.commit( + lambda c: {**c, "b": 2}, expected_revision=None, publish=probe_publish + ) + self.assertEqual(seen, [(saved, {"a": 1, "b": 2}, False)]) + + def test_publish_is_skipped_when_the_write_fails(self): + self._write({"a": 1}) + published = [] + with patch.object(config_store.os, "fsync", side_effect=OSError("disk full")): + with self.assertRaises(OSError): + self.store.commit( + lambda c: {**c, "b": 2}, + expected_revision=None, + publish=published.append, + ) + self.assertEqual(published, []) + + def test_handles_closed_after_commit(self): + self._write({"a": 1}) + self.store.read() + self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None) + assert_closed(self, self.tmpdir) + os.unlink(self.path) # Windows refuses this while a handle is open + self.assertEqual(self._leftovers(), []) + + +class SingletonTests(unittest.TestCase): + def test_get_config_store_is_one_process_store_on_the_configured_path(self): + store = get_config_store() + self.assertIs(store, get_config_store()) + tmpdir = tempfile.mkdtemp() + try: + custom = os.path.join(tmpdir, "c.json") + with patch.dict(os.environ, {"PROMPT_MANAGER_CONFIG_PATH": custom}): + self.assertEqual(store.path, custom) + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_image_tag_jobs.py b/tests/test_image_tag_jobs.py index 3332422..8964b07 100644 --- a/tests/test_image_tag_jobs.py +++ b/tests/test_image_tag_jobs.py @@ -420,6 +420,23 @@ class TestStartAndExclusion(ImageTagJobTestCase): # A refused start never leaves the gate taken. self.assertIsNone(model_gate.holder()) + async def test_image_tag_job_refused_while_legacy_restore_holds(self): + self.png("a.png") + async with self.api._jobs.hold( + "restore", "Database restore (admin page)", admission="records" + ): + status, body = await self.start(self.add_body(["a.png"], ["x"])) + self.assertEqual((status, body["code"]), (409, "busy")) + self.assertEqual(body["running"]["title"], "Database restore (admin page)") + status, body = await self.start(self.auto_body(["a.png"])) + self.assertEqual((status, body["code"]), (409, "busy")) + self.assertIsNone(model_gate.holder()) + async with self.api._jobs.hold( + "thumbnails", "Thumbnail generation (gallery page)", admission="guarded" + ): + status, body = await self.start(self.add_body(["a.png"], ["x"])) + self.assertEqual((status, body["code"]), (409, "busy")) + async def test_record_changing_excludes_running_scan(self): self.png("a.png") gate = asyncio.Event() diff --git a/tests/test_job_admission.py b/tests/test_job_admission.py new file mode 100644 index 0000000..59e4821 --- /dev/null +++ b/tests/test_job_admission.py @@ -0,0 +1,552 @@ +"""Registry admission classes and the legacy routes that hold the registry. + +Record-changing work never runs together (Activity design): a ``records`` +job is refused while any ``records`` or ``guarded`` job runs, a ``guarded`` +job (thumbnails) while a ``records`` job runs, and ``read`` jobs only share +the one-per-kind rule. The legacy admin and gallery routes that change +records or files take a registry hold for their duration, so a workspace job +and a legacy-page operation refuse each other, naming the one that runs. + +Fake runners and gated executor calls only; ordering is driven by events. +""" + +import asyncio +import hashlib +import json +import os +import sys +import threading +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +_mock_server = MagicMock() +_mock_server.PromptServer.instance.routes = MagicMock() +sys.modules.setdefault("server", _mock_server) + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from py.api.jobs import ( # noqa: E402 + ADMISSIONS, + Job, + JobBusy, + JobRegistry, +) + +try: + from tests.test_api_admin import AdminAPITestCase # noqa: E402 +except ImportError: # discovered with tests/ as the top-level directory + from test_api_admin import AdminAPITestCase # noqa: E402 + + +def _gated(gate): + async def runner(job): + await gate.wait() + + return runner + + +async def _noop(job): + return None + + +class TestAdmissionClasses(unittest.IsolatedAsyncioTestCase): + + def _start(self, registry, kind, admission, gate=None, **kwargs): + return registry.start( + kind, + kind.title(), + admission=admission, + scope={}, + runner=_gated(gate) if gate is not None else _noop, + **kwargs, + ) + + async def test_admission_names_and_explicit_read(self): + self.assertEqual(ADMISSIONS, ("read", "guarded", "records")) + registry = JobRegistry() + job = registry.start( + "report", "Report", admission="read", scope={}, runner=_noop + ) + await asyncio.wait_for(job.task, 2) + self.assertEqual(job.admission, "read") + self.assertFalse(job.record_changing) + with self.assertRaises(ValueError): + registry.start("x", "X", admission="exclusive", scope={}, runner=_noop) + with self.assertRaises(ValueError): + registry.start( + "x", + "X", + admission="read", + record_changing=True, + scope={}, + runner=_noop, + ) + + async def test_start_requires_an_admission_class(self): + # A job that forgets its class must not silently run as "read" beside + # a scan or restore: every start call decides. + registry = JobRegistry() + with self.assertRaises(TypeError) as raised: + registry.start("review", "Review", scope={}, runner=_noop) + self.assertIn("admission", str(raised.exception)) + self.assertEqual(registry.list(), []) + job = registry.start( + "review", "Review", record_changing=False, scope={}, runner=_noop + ) + await asyncio.wait_for(job.task, 2) + self.assertEqual(job.admission, "read") + + async def test_record_changing_is_an_alias_for_records(self): + registry = JobRegistry() + gate = asyncio.Event() + job = registry.start( + "scan", "Scan", record_changing=True, scope={}, runner=_gated(gate) + ) + self.assertEqual(job.admission, "records") + self.assertTrue(job.record_changing) + self.assertEqual(job.summary()["admission"], "records") + self.assertIs( + registry.blocker("image-tags", record_changing=True), job + ) # the existing call shape keeps working + gate.set() + await asyncio.wait_for(job.task, 2) + + async def test_records_job_refused_while_guarded_runs(self): + registry = JobRegistry() + gate = asyncio.Event() + thumbs = self._start(registry, "thumbnails", "guarded", gate) + with self.assertRaises(JobBusy) as raised: + self._start(registry, "scan", "records") + self.assertIs(raised.exception.running, thumbs) + self.assertIs(registry.blocker("restore", admission="records"), thumbs) + gate.set() + await asyncio.wait_for(thumbs.task, 2) + scan = self._start(registry, "scan", "records") + await asyncio.wait_for(scan.task, 2) + self.assertEqual(scan.state, "finished") + + async def test_guarded_refused_while_records_runs(self): + registry = JobRegistry() + gate = asyncio.Event() + scan = self._start(registry, "scan", "records", gate) + with self.assertRaises(JobBusy) as raised: + self._start(registry, "thumbnails", "guarded") + self.assertIs(raised.exception.running, scan) + # A second guarded kind may run beside a guarded one, not beside records. + gate.set() + await asyncio.wait_for(scan.task, 2) + hold = asyncio.Event() + thumbs = self._start(registry, "thumbnails", "guarded", hold) + backup = self._start(registry, "backup", "guarded") + await asyncio.wait_for(backup.task, 2) + self.assertEqual(backup.state, "finished") + hold.set() + await asyncio.wait_for(thumbs.task, 2) + + async def test_read_jobs_run_in_parallel(self): + registry = JobRegistry() + gate = asyncio.Event() + scan = self._start(registry, "scan", "records", gate) + health = self._start(registry, "health", "read", gate) + duplicates = self._start(registry, "duplicates", "read", gate) + self.assertTrue(all(j.running for j in (scan, health, duplicates))) + with self.assertRaises(JobBusy) as raised: # one job of a kind + self._start(registry, "health", "read") + self.assertIs(raised.exception.running, health) + gate.set() + await asyncio.wait_for( + asyncio.gather(scan.task, health.task, duplicates.task), 2 + ) + + async def test_non_cancellable_job_ignores_cancel(self): + registry = JobRegistry() + gate = asyncio.Event() + job = self._start(registry, "restore", "records", gate, cancellable=False) + self.assertIs(job.summary()["cancellable"], False) + self.assertFalse(registry.cancel(job.id)) + self.assertFalse(job.cancelled) + gate.set() + await asyncio.wait_for(job.task, 2) + self.assertEqual(job.state, "finished") + + async def test_hold_registers_a_running_non_cancellable_job(self): + registry = JobRegistry() + held = registry.hold( + "restore", "Database restore (admin page)", admission="records" + ) + async with held as job: + self.assertIs(registry.active("restore"), job) + self.assertEqual(job.state, "running") + self.assertEqual( + (job.summary()["cancellable"], job.summary()["admission"]), + (False, "records"), + ) + self.assertFalse(registry.cancel(job.id)) + with self.assertRaises(JobBusy) as raised: + self._start(registry, "scan", "records") + self.assertIn("Database restore (admin page)", str(raised.exception)) + self.assertEqual(job.state, "finished") + self.assertIsNone(registry.active("restore")) + self.assertIs(registry.list()[0], job) + + async def test_hold_refused_without_waiting(self): + registry = JobRegistry() + gate = asyncio.Event() + scan = self._start(registry, "scan", "records", gate) + with self.assertRaises(JobBusy) as raised: + async with registry.hold("maintenance", "Maintenance", admission="records"): + self.fail("a refused hold must not run its body") + self.assertIs(raised.exception.running, scan) + self.assertEqual([j.kind for j in registry.list()], ["scan"]) + gate.set() + await asyncio.wait_for(scan.task, 2) + + async def test_hold_released_on_exception(self): + registry = JobRegistry() + with self.assertRaises(RuntimeError): + async with registry.hold( + "maintenance", "Maintenance", admission="records" + ) as job: + raise RuntimeError("boom") + self.assertEqual(job.state, "failed") + again = registry.hold("maintenance", "Maintenance", admission="records") + async with again as second: + second.finish("failed") # the body may end it itself + self.assertEqual(second.state, "failed") + + async def test_hold_admits_on_entry_not_at_the_call(self): + # A hold created but never entered (an await or an exception between + # the call and ``async with``) must not leave a running records job + # behind that blocks all record work until a restart. + registry = JobRegistry() + held = registry.hold("restore", "Restore", admission="records") + self.assertEqual(registry.list(), []) + self.assertIsNone(registry.blocker("scan", admission="records")) + del held + async with registry.hold("restore", "Restore", admission="records") as job: + self.assertIs(registry.active("restore"), job) + self.assertEqual(job.state, "finished") + + def test_job_defaults_stay_read_and_cancellable(self): + job = Job("duplicates", "Duplicate scan") + self.assertEqual((job.admission, job.cancellable), ("read", True)) + + +class _LegacyHoldCase(AdminAPITestCase): + """The admin test app plus a running workspace job on demand.""" + + def _run_job(self, kind="scan", title="Scan for metadata", admission="records"): + gate = asyncio.Event() + job = self.api._jobs.start( + kind, title, admission=admission, scope={}, runner=_gated(gate) + ) + self.addAsyncCleanup(self._release, gate, job) + return gate, job + + @staticmethod + async def _release(gate, job): + gate.set() + await asyncio.wait_for(job.task, 5) + + def _db_digest(self): + path = Path(self.api.db.model.db_path) + return { + suffix: hashlib.sha256(Path(f"{path}{suffix}").read_bytes()).hexdigest() + for suffix in ("", "-wal") + if Path(f"{path}{suffix}").exists() + } + + async def _busy(self, resp, title="Scan for metadata"): + body = await resp.json() + self.assertEqual(resp.status, 409) + self.assertEqual(body, {"success": False, "error": f"{title} already running"}) + + +class TestLegacyRoutesHoldTheRegistry(_LegacyHoldCase): + + async def test_hold_is_listed_as_running_job(self): + held = self.api._jobs.hold( + "restore", "Database restore (admin page)", admission="records" + ) + async with held: + resp = await self.client.request("GET", "/prompt_manager/jobs") + jobs = (await resp.json())["jobs"] + self.assertEqual(len(jobs), 1) + job = jobs[0] + self.assertEqual( + (job["kind"], job["title"], job["state"]), + ("restore", "Database restore (admin page)", "running"), + ) + self.assertIs(job["cancellable"], False) + self.assertEqual(job["admission"], "records") + + async def test_legacy_maintenance_and_cleanup_refused_while_scan_runs(self): + self.api.db.save_prompt(text="keep me", prompt_hash="h1") + before = self._db_digest() + self._run_job() + resp = await self.client.request( + "POST", + "/prompt_manager/maintenance", + json={"operations": ["cleanup_duplicates", "vacuum"]}, + ) + await self._busy(resp) + resp = await self.client.request("POST", "/prompt_manager/cleanup") + await self._busy(resp) + self.assertEqual(self._db_digest(), before) + + async def test_read_only_maintenance_runs_during_scan(self): + # The legacy statistics and consistency checks only read: they hold + # the registry as "read", so a long scan does not block them. + self.api.db.save_prompt(text="keep me", prompt_hash="h1") + _gate, scan = self._run_job() + listed = [] + real = self.api._run_maintenance_operations + + def observed(operations): + listed.extend( + (job.kind, job.admission, job.state) for job in self.api._jobs.list() + ) + return real(operations) + + self.api._run_maintenance_operations = observed + resp = await self.client.request( + "POST", + "/prompt_manager/maintenance", + json={ + "operations": [ + "statistics", + "check_consistency", + "check_hash_duplicates", + ] + }, + ) + body = await resp.json() + self.assertEqual(resp.status, 200) + self.assertIs(body["success"], True) + self.assertEqual( + set(body["results"]), + {"statistics", "check_consistency", "check_hash_duplicates"}, + ) + self.assertIn(("maintenance", "read", "running"), listed) + self.assertEqual(scan.state, "running") + maintenance = self.api._jobs.list()[-1] + self.assertEqual( + (maintenance.kind, maintenance.admission, maintenance.state), + ("maintenance", "read", "finished"), + ) + + async def test_mixed_maintenance_selection_is_refused_during_scan(self): + # One write operation among read-only ones makes the run "records". + self.api.db.save_prompt(text="keep me", prompt_hash="h1") + before = self._db_digest() + self._run_job() + resp = await self.client.request( + "POST", + "/prompt_manager/maintenance", + json={"operations": ["statistics", "vacuum"]}, + ) + await self._busy(resp) + self.assertEqual(self._db_digest(), before) + + async def test_default_maintenance_selection_is_refused_during_scan(self): + # No "operations" key means the default list, which vacuums. + self.api.db.save_prompt(text="keep me", prompt_hash="h1") + before = self._db_digest() + self._run_job() + resp = await self.client.request("POST", "/prompt_manager/maintenance", json={}) + await self._busy(resp) + self.assertEqual(self._db_digest(), before) + self.assertEqual([j.kind for j in self.api._jobs.list()], ["scan"]) + + async def test_legacy_maintenance_refused_while_image_tags_runs(self): + self._run_job("image-tags", "Add tags to 3 images") + resp = await self.client.request("POST", "/prompt_manager/maintenance") + await self._busy(resp, "Add tags to 3 images") + + async def test_scan_refused_while_legacy_maintenance_holds(self): + entered = threading.Event() + release = threading.Event() + real = self.api._run_maintenance_operations + + def gated(operations): + entered.set() + release.wait(10) + return real(operations) + + self.api._run_maintenance_operations = gated + loop = asyncio.get_running_loop() + maintenance = asyncio.ensure_future( + self.client.request( + "POST", "/prompt_manager/maintenance", json={"operations": ["vacuum"]} + ) + ) + self.assertTrue(await loop.run_in_executor(None, entered.wait, 10)) + resp = await self.client.request("POST", "/prompt_manager/jobs/scan") + body = await resp.json() + self.assertEqual((resp.status, body["code"]), (409, "busy")) + self.assertEqual(body["running"]["title"], "Maintenance (admin page)") + release.set() + self.assertEqual((await maintenance).status, 200) + listed = ( + await (await self.client.request("GET", "/prompt_manager/jobs")).json() + )["jobs"] + self.assertEqual( + [(j["kind"], j["state"]) for j in listed], [("maintenance", "finished")] + ) + + async def test_legacy_duplicate_delete_refused_while_scan_runs(self): + victim = self.output_dir / "dup.png" + victim.write_bytes(b"x") + self._run_job() + resp = await self.client.request( + "POST", + "/prompt_manager/delete_duplicate_images", + json={"image_paths": [str(victim)]}, + ) + await self._busy(resp) + self.assertTrue(victim.exists()) + # The duplicate family lock was not taken by the refused request. + self.assertIsNotNone(self.api._acquire_long_job(self.api.LONG_JOB_DUPLICATES)) + + async def test_lora_scan_refused_while_scan_runs(self): + from py.config import IntegrationConfig + + lm = Path(self.tmpdir) / "custom_nodes" / "ComfyUI-Lora-Manager" + (lm / "py").mkdir(parents=True) + self._run_job() + imported = [] + with ( + patch.multiple( + IntegrationConfig, LORA_MANAGER_ENABLED=True, LORA_MANAGER_PATH=str(lm) + ), + patch("py.lora_utils.resolve_lora_manager_path", return_value=str(lm)), + patch.object( + self.api, "_lora_scan_stream", side_effect=lambda *a: imported.append(a) + ), + ): + resp = await self.client.request("POST", "/prompt_manager/lora/scan") + text = await resp.text() + self.assertEqual(resp.status, 200) # the legacy page reads a stream + events = [json.loads(line[6:]) for line in text.split("\n\n") if line] + # The admin page's modal shows ``status`` and closes on ``type: error``. + message = "Scan for metadata already running" + self.assertEqual( + events, + [{"type": "error", "status": message, "progress": 0, "message": message}], + ) + self.assertEqual(imported, []) + + async def test_legacy_autotag_refused_while_scan_runs(self): + # Both pages read this route as a stream (a 409 reads as a lost + # connection there), so a refusal by another job is one error frame: + # the admin page keys on ``type``, the gallery on ``status``. + self._run_job() + resp = await self.client.request( + "POST", "/prompt_manager/autotag/start", json={} + ) + text = await resp.text() + self.assertEqual(resp.status, 200) + self.assertEqual(resp.headers["Content-Type"], "text/event-stream") + events = [json.loads(line[6:]) for line in text.split("\n\n") if line] + message = "Scan for metadata already running" + self.assertEqual( + events, [{"type": "error", "status": "error", "message": message}] + ) + self.assertEqual( + [j.kind for j in self.api._jobs.list() if j.kind == "autotag"], [] + ) + + async def test_second_legacy_autotag_batch_keeps_its_json_409(self): + from py.api.autotag_routes import _BUSY_ERROR + + self._run_job("autotag", "Auto-tag images (legacy page)") + resp = await self.client.request( + "POST", "/prompt_manager/autotag/start", json={} + ) + self.assertEqual(resp.status, 409) + self.assertEqual(await resp.json(), {"success": False, "error": _BUSY_ERROR}) + + async def test_legacy_payloads_unchanged_when_idle(self): + resp = await self.client.request( + "POST", "/prompt_manager/maintenance", json={"operations": ["vacuum"]} + ) + body = await resp.json() + self.assertEqual(resp.status, 200) + self.assertEqual( + set(body), + {"success", "operations_completed", "all_successful", "results", "message"}, + ) + resp = await self.client.request("POST", "/prompt_manager/cleanup") + body = await resp.json() + self.assertEqual(resp.status, 200) + self.assertEqual(set(body), {"success", "message", "duplicates_removed"}) + resp = await self.client.request( + "POST", "/prompt_manager/delete_duplicate_images", json={"image_paths": []} + ) + self.assertEqual( + (resp.status, await resp.json()), + (400, {"success": False, "error": "No image paths provided"}), + ) + # Each hold ended with the request; refused input never took one. + listed = await self._listed() + self.assertEqual( + [(j["kind"], j["state"]) for j in listed], + [("maintenance", "finished"), ("maintenance", "finished")], + ) + self.assertEqual( + [j["title"] for j in listed], + ["Prompt cleanup (admin page)", "Maintenance (admin page)"], + ) + + async def _listed(self): + resp = await self.client.request("GET", "/prompt_manager/jobs") + return (await resp.json())["jobs"] + + +class TestHeldJobStates(_LegacyHoldCase): + """Activity lists what a held legacy operation really did (Minor 7).""" + + async def _listed(self): + resp = await self.client.request("GET", "/prompt_manager/jobs") + return [(j["kind"], j["state"]) for j in (await resp.json())["jobs"]] + + async def test_refused_input_takes_no_hold(self): + resp = await self.client.request( + "POST", "/prompt_manager/delete_duplicate_images", json={} + ) + self.assertEqual(resp.status, 400) + resp = await self.client.request( + "POST", + "/prompt_manager/images/generate-thumbnails", + data=b"{not json", + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(resp.status, 400) + self.assertEqual(await self._listed(), []) + + async def test_thumbnail_stream_that_ends_in_error_is_failed(self): + with patch.object(self.api, "_get_all_output_dirs", return_value=[]): + resp = await self.client.request( + "POST", "/prompt_manager/images/generate-thumbnails/progress" + ) + text = await resp.text() + self.assertEqual(resp.status, 200) + self.assertIn("event: error", text) + self.assertEqual(await self._listed(), [("thumbnails", "failed")]) + + async def test_legacy_autotag_that_ends_in_error_is_failed(self): + service = MagicMock() + service.get_models_status.return_value = {} # model not downloaded + with patch("py.autotag.get_autotag_service", return_value=service): + resp = await self.client.request( + "POST", "/prompt_manager/autotag/start", json={} + ) + text = await resp.text() + self.assertEqual(resp.status, 200) + self.assertIn('"type": "error"', text) + self.assertEqual(await self._listed(), [("autotag", "failed")]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_jobs.py b/tests/test_jobs.py index f561be0..6848fa8 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -182,8 +182,13 @@ class TestJobRegistry(unittest.IsolatedAsyncioTestCase): "scope", "listed", "log_length", + "cancellable", + "admission", }, ) + self.assertEqual( + (summary["cancellable"], summary["admission"]), (True, "records") + ) self.assertEqual(summary["scope"], {"files": "new", "output_folders": 2}) # What the detail would list, so a view reads it only when that changed. self.assertEqual((summary["listed"], summary["log_length"]), ({}, 0)) diff --git a/tools/vendor/frontend-coverage-baseline.json b/tools/vendor/frontend-coverage-baseline.json index 31ab8b3..56277d9 100644 --- a/tools/vendor/frontend-coverage-baseline.json +++ b/tools/vendor/frontend-coverage-baseline.json @@ -1,9 +1,9 @@ { "admin.js": { - "lines": 28, + "lines": 31, "branches": 58, - "functions": 23, - "statements": 28 + "functions": 26, + "statements": 31 }, "api-client.js": { "lines": 98, @@ -42,10 +42,10 @@ "statements": 98 }, "gallery.js": { - "lines": 39, + "lines": 40, "branches": 63, - "functions": 43, - "statements": 39 + "functions": 44, + "statements": 40 }, "html-escape.js": { "lines": 98, @@ -157,7 +157,7 @@ }, "workspace/activity-job-view.mjs": { "lines": 97, - "branches": 83, + "branches": 86, "functions": 98, "statements": 97 }, @@ -174,10 +174,10 @@ "statements": 98 }, "workspace/activity.mjs": { - "lines": 95, + "lines": 96, "branches": 87, "functions": 98, - "statements": 95 + "statements": 96 }, "workspace/file-inspector.mjs": { "lines": 96, diff --git a/web/js/admin.js b/web/js/admin.js index da699ac..e74d347 100644 --- a/web/js/admin.js +++ b/web/js/admin.js @@ -1180,6 +1180,11 @@ `Done — ${data.imported} imported, ${data.skipped} skipped`; this.loadStatistics(); setTimeout(() => this.hideModal("loraImportModal"), 2000); + } else if (data.type === "error") { + // A refusal (another job runs) or a failed import: name it and close the modal. + document.getElementById("loraImportStatus").textContent = `Error: ${data.message}`; + this.showNotification(`LoRA import failed: ${data.message}`, "error"); + setTimeout(() => this.hideModal("loraImportModal"), 3000); } } catch (e) { /* skip malformed SSE */ } } @@ -1231,10 +1236,13 @@ } this.hideModal("settingsModal"); } else { - throw new Error("Failed to save settings"); + // A refusal names its reason (a corrupt config.json, a changed file): show it. + const data = await response.json().catch(() => ({})); + throw new Error(data.error ? `Failed to save settings: ${data.error}` : "Failed to save settings"); } } catch (error) { - this.showNotification("Failed to save settings", "error"); + const reason = error.message.startsWith("Failed to save settings:") ? error.message : "Failed to save settings"; + this.showNotification(reason, "error"); } } @@ -1848,7 +1856,9 @@ throw new Error(data.error || 'Maintenance failed'); } } else { - throw new Error(`HTTP ${response.status}: ${response.statusText}`); + // A refusal names the job that runs ("Scan for metadata already running"). + const data = await response.json().catch(() => ({})); + throw new Error(data.error || `HTTP ${response.status}: ${response.statusText}`); } } catch (error) { console.error('Maintenance error:', error); diff --git a/web/js/gallery.js b/web/js/gallery.js index 56596f7..ee9f24b 100644 --- a/web/js/gallery.js +++ b/web/js/gallery.js @@ -17,6 +17,14 @@ this.dupResults = null; this.api = ApiClient.createApiClient(); + // The page's lifetime ends when it starts to leave: WebKit fails a fetch a leaving page starts + // ("Fetch API cannot load … due to access control checks.") and logs it as a page error. + this.pageLifetime = new AbortController(); + window.addEventListener("beforeunload", () => this.pageLifetime.abort()); + window.addEventListener("pagehide", () => this.pageLifetime.abort()); + window.addEventListener("pageshow", (e) => { + if (e.persisted && this.pageLifetime.signal.aborted) this.pageLifetime = new AbortController(); + }); // Only the newest page request may update the grid this.imagesRequest = ApiClient.latestOnly(); @@ -2129,8 +2137,11 @@ Seed: ${this.currentMetadata.seed || 'Unknown'}`; headers: { 'Content-Type': 'application/json' } }); - if (response.ok) { - const data = await response.json(); + if (!response.ok) { + // Refused (a running job holds the thumbnails): say why and claim nothing. + const data = await response.json().catch(() => ({})); + this.showNotification(`Thumbnails were not cleared: ${data.error || `HTTP ${response.status}`}`, 'error'); + return; } } catch (e) { console.warn('Could not clear server thumbnails:', e); @@ -2531,8 +2542,11 @@ Seed: ${this.currentMetadata.seed || 'Unknown'}`; return; } + // A leaving page starts no read, and one it aborts is not reported. + const signal = this.pageLifetime?.signal; + if (signal?.aborted) return; try { - const data = await this.api.get('/prompt_manager/images/thumbnails/status'); + const data = await this.api.get('/prompt_manager/images/thumbnails/status', { signal }); const total = data.total_images; const pending = data.pending; if (!data.success || !Number.isSafeInteger(total) || diff --git a/web/js/workspace/activity-job-view.mjs b/web/js/workspace/activity-job-view.mjs index 9808b7a..f853cb4 100644 --- a/web/js/workspace/activity-job-view.mjs +++ b/web/js/workspace/activity-job-view.mjs @@ -274,7 +274,8 @@ export function mountActivityJobView(main, { store, scope, api, jobId }) { const lost = job.state === 'lost'; if (lost && document.activeElement === download) parts.h1.focus(); download.disabled = lost; - if (job.state === 'running') { + // Truth rule: a job that cannot be cancelled (a held legacy-page job) shows no Cancel. + if (job.state === 'running' && job.cancellable) { if (!cancel.isConnected) parts.actions.append(cancel); } else if (cancel.isConnected) { // A focused Cancel that goes away hands its focus to the job's heading, never to the page body. diff --git a/web/js/workspace/activity-store.mjs b/web/js/workspace/activity-store.mjs index 3df6405..72eb6ab 100644 --- a/web/js/workspace/activity-store.mjs +++ b/web/js/workspace/activity-store.mjs @@ -103,6 +103,8 @@ export function parseJob(raw) { scope: plainObject(raw.scope) ? { ...raw.scope } : {}, listed, log_length: count(raw.log_length), + // A held legacy-page job cannot be cancelled; an older server sends no key (cancellable then). + cancellable: raw.cancellable !== false, }; } diff --git a/web/js/workspace/activity.mjs b/web/js/workspace/activity.mjs index 48b83c4..eb7bc14 100644 --- a/web/js/workspace/activity.mjs +++ b/web/js/workspace/activity.mjs @@ -153,11 +153,15 @@ function itemFor(job, scope) { progress.textContent = `${next.progress}%`; progress.setAttribute('aria-label', `${next.title} progress`); if (!progress.isConnected) item.insertBefore(progress, actions.isConnected ? actions : null); - if (!actions.isConnected) item.append(actions); } else { + progress.remove(); + } + // Truth rule: a job that cannot be cancelled (a held legacy-page job) shows no Cancel job. + if (running && next.cancellable) { + if (!actions.isConnected) item.append(actions); + } else if (actions.isConnected) { // A focused Cancel job that goes away hands its focus to the item's title link, never to the page body. if (actions.contains(document.activeElement)) link.focus(); - progress.remove(); actions.remove(); } };