From 3b1d00501f7ec035afbf1dba8136930a76732f93 Mon Sep 17 00:00:00 2001 From: Vito Date: Thu, 22 Oct 2026 21:39:01 -0700 Subject: [PATCH] fix(config): serialized config store and registry admission (#289) * fix(config): report settings persistence failures (#220) Settings saves now go through one serialized config store (py/config_store.py). A commit re-reads config.json under a process lock, applies the change to a copy and replaces the file atomically (temp file beside it, fsync, close, os.replace), so a failed write leaves the old file in place. POST /prompt_manager/settings commits first and only then publishes the values at runtime. A write failure is 500 config-write-failed, an unparseable config.json is 409 config-unreadable (the file is kept instead of replaced) and a stale expected_revision is 409 settings-changed. A save reports persisted, revision, runtime_applied and restart_required_fields; GET adds the revision. Only submitted keys are written; other sections and the CivitAI key are kept. * fix(lora): report integration save failures (#220) POST /prompt_manager/lora/enable commits integrations.lora_manager through the config store and publishes the runtime values only after config.json holds them. Before, it set the runtime values, rewrote the whole runtime config (dropping sections it did not know) and reported success even when the write had failed. A write failure is 500 config-write-failed and changes nothing at runtime; an unparseable config.json is 409 config-unreadable. The body goes through the capped JSON reader and the flags must be booleans. A trigger word cache that fails to load is reported as persisted:true, cache_loaded:false. The CivitAI key is never echoed. The LoRA tests now write to a temporary config path. * fix(jobs): refuse legacy record-changing operations while a job runs The job registry gains admission classes. A records job is refused while any records or guarded job runs, a guarded job while a records job runs, and read jobs share only the one-job-per-kind rule. record_changing=True stays as the spelling of records. Job summaries add cancellable and admission, and a non-cancellable job ignores cancel. JobRegistry.hold() registers a running, non-cancellable job for a legacy-page operation and ends it when the request does. The legacy restore, maintenance, prompt cleanup, duplicate image delete, LoRA import and batch auto-tag routes hold as records; thumbnail generation (both routes) and thumbnail clear hold as guarded. A maintenance run of only read checks holds as read. So a restore during a scan, or a scan during a legacy restore, is refused with the running job named instead of interleaving; this closes the gap noted on #269. Refusals keep each route's shape: the JSON routes return the legacy 409 " already running", and the thumbnail and LoRA streams send their existing error event. * test(lora): keep the preserved-section fixture free of secret values CodeQL flagged the fixture that writes the secret constant into a test config file as clear-text storage. The test only needs a marker value to prove the stored key survives a save. * fix(config): read under the commit lock and publish saves in write order - ConfigStore.read takes the commit lock, so no read handle is open while config.json is replaced (Windows refuses that replace). - os.replace is retried briefly on a Windows PermissionError (antivirus or indexer handles) before the save is reported as not written. - The temp file is written beside realpath(config.json), so a symlinked config keeps its link and a bind mount fails truthfully. - commit() takes a publish callback that runs under the lock after the write; settings and LoRA saves publish there, so two concurrent saves reach the runtime in write order and the runtime ends on the last one. - The concurrency tests now wait until the second caller is blocked on the lock (tests/lock_probe.py) before releasing the first. * fix(jobs): require an admission class, admit holds on entry, true Activity states - JobRegistry.start raises TypeError when neither admission nor record_changing is given, so a new job never runs as "read" beside a scan by omission. - hold() admits when its block is entered, so a hold that is never entered leaves no running records job behind; callers that answer a refusal in their own way check blocker() first, with no await before the block. - A held SSE operation that ends in an error event (thumbnail generation, legacy batch auto-tag) is listed as failed, and refused input (duplicate delete without paths, an unreadable thumbnail body) is answered before the hold, so it is not listed as a job. - Activity keeps a job's cancellable flag (true when an older server omits it) and shows Cancel only for a running, cancellable job, in Recent work and in the job view. A held legacy job no longer offers a Cancel that does nothing. * fix(admin): legacy pages show why a request was refused A job running in the workspace refuses the legacy record-changing requests, and a corrupt config.json refuses a settings save. The legacy pages now name the reason instead of hiding it: - LoRA import: the busy and failure frames carry status and progress, and the admin modal handles type "error": it shows the message, notifies and closes, instead of staying on "Initializing..." at 0%. - Legacy batch auto-tag: a refusal by another job is a 200 stream with one error frame ({type, status, message}), which both pages already render; a second batch keeps its exact JSON 409. - Gallery cache clear: a refused server clear shows the reason and stops before the success toast and the "Not generated" reset. - Admin settings save and maintenance show the reply's error ("config.json cannot be read...", "Scan for metadata already running") instead of "Failed to save settings" or "HTTP 409: Conflict". - Tests cover the read-only /maintenance branch: a read-only selection runs during a scan as a read hold; a mixed or default selection is refused. * refactor(api): move the settings routes out of admin.py GET and POST /prompt_manager/settings, their validation, the config commit and runtime publication, and the shared commit-failure reply now live in py/api/settings_routes.py as SettingsRoutesMixin, before S3 adds host fields there. No behaviour change: the same routes, payloads and status codes; admin.py drops about 300 lines. * chore(coverage): raise the floors of the frontend files this round touched admin.js, gallery.js, activity.mjs and activity-job-view.mjs, to their measured coverage. No floor is lowered. * fix(admin): the gallery starts no thumbnail status read while it leaves The gallery's startup thumbnail check runs one second after load. When that second ended while the page was leaving (CI, WebKit: the workflow test navigates from the gallery to the admin page), WebKit failed the new read with "Fetch API cannot load ... due to access control checks." and Playwright reported a page error. Like the admin page (#290), the gallery now has a page lifetime that ends on beforeunload and pagehide (and starts again on a back/forward restore). The check starts no read once it has ended and passes it as the read's signal. A page-clock workflow test that leaves before the second ends failed on both events, in Chromium and WebKit, before this change. --- py/api/__init__.py | 21 + py/api/admin.py | 309 +++-------- py/api/autotag_routes.py | 35 ++ py/api/images.py | 62 ++- py/api/jobs.py | 186 ++++++- py/api/lora_integration.py | 124 ++++- py/api/settings_routes.py | 339 ++++++++++++ py/config_store.py | 197 +++++++ tests/browser/workflows.spec.mjs | 18 + tests/browser/workspace-activity.spec.mjs | 25 + tests/browser/workspace_activity_fixture.py | 35 ++ tests/js/gallery-thumbnails.test.js | 22 + tests/js/legacy-refusals.test.js | 150 +++++ tests/js/workspace-activity-store.test.js | 7 +- tests/lock_probe.py | 35 ++ tests/test_api_admin.py | 178 +++++- tests/test_api_backup_restore.py | 83 +++ tests/test_api_images.py | 89 +++ tests/test_api_lora.py | 352 +++++++++++- tests/test_config_store.py | 419 ++++++++++++++ tests/test_image_tag_jobs.py | 17 + tests/test_job_admission.py | 552 +++++++++++++++++++ tests/test_jobs.py | 5 + tools/vendor/frontend-coverage-baseline.json | 18 +- web/js/admin.js | 16 +- web/js/gallery.js | 20 +- web/js/workspace/activity-job-view.mjs | 3 +- web/js/workspace/activity-store.mjs | 2 + web/js/workspace/activity.mjs | 8 +- 29 files changed, 3003 insertions(+), 324 deletions(-) create mode 100644 py/api/settings_routes.py create mode 100644 py/config_store.py create mode 100644 tests/js/legacy-refusals.test.js create mode 100644 tests/lock_probe.py create mode 100644 tests/test_config_store.py create mode 100644 tests/test_job_admission.py 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:<hex>"`` 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(); } };