diff --git a/__init__.py b/__init__.py index b4971238..6b41698d 100644 --- a/__init__.py +++ b/__init__.py @@ -102,6 +102,7 @@ try: def _preview_startup(): api.start_preview_location() + api.recover_thumbnail_move() # an interrupted move, before the sweep api.sweep_thumbnail_temps() threading.Thread( @@ -118,6 +119,8 @@ try: # Build each output root's Images listing once the server's loop runs, # so the first open does not wait for the walk (#348). host_app.on_startup.append(api.warm_gallery_listings) + # #320: old thumbnails a move left to remove, once it is checked. + host_app.on_startup.append(api.finish_thumbnail_move) if host_app is not None and hasattr(host_app, "on_shutdown"): # Open Needs review change streams end before the server waits on them. host_app.on_shutdown.append(api.close_review_events) diff --git a/py/api/__init__.py b/py/api/__init__.py index 14463980..6997f686 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -62,6 +62,10 @@ from .job_routes import JobRoutesMixin from .image_tag_jobs import ImageTagJobsMixin from .thumbnail_jobs import ThumbnailJobsMixin from .thumbnail_location import ThumbnailLocationMixin +from .thumbnail_move import ThumbnailMoveMixin +from .thumbnail_move_plan import ThumbnailMovePlanMixin +from .thumbnail_move_recovery import ThumbnailMoveRecoveryMixin +from .thumbnail_prune import ThumbnailPruneMixin from .thumbnail_convert import ThumbnailConvertMixin from .thumbnail_auto import AutoThumbnailMixin from .system import SystemRoutesMixin @@ -341,6 +345,10 @@ class PromptManagerAPI( ImageTagJobsMixin, ThumbnailJobsMixin, ThumbnailLocationMixin, + ThumbnailMovePlanMixin, + ThumbnailMoveMixin, + ThumbnailMoveRecoveryMixin, + ThumbnailPruneMixin, ThumbnailConvertMixin, AutoThumbnailMixin, SystemRoutesMixin, @@ -719,6 +727,7 @@ class PromptManagerAPI( # ── Register domain-specific routes from mixins ─────────────── self._register_pair_routes(routes) + self._register_thumbnail_move_routes(routes) self._register_library_routes(routes) self._register_library_delete_routes(routes) self._register_vocabulary_routes(routes) diff --git a/py/api/preview_store.py b/py/api/preview_store.py index c0846368..50deb422 100644 --- a/py/api/preview_store.py +++ b/py/api/preview_store.py @@ -186,6 +186,13 @@ def _set_keys(parsed, marker): _key_cache.clear() +def set_cleanup(entries): + """The setting's ``cleanup`` after a removal was committed.""" + global _parsed + with _state_lock: + _parsed = replace(_parsed, cleanup=tuple(entries)) + + def mark_invalid(): """The setting cannot be used (a startup check that failed): folder mode stays in force, unavailable, so nothing is written anywhere.""" diff --git a/py/api/settings_routes.py b/py/api/settings_routes.py index b0f1c915..9ba01fe7 100644 --- a/py/api/settings_routes.py +++ b/py/api/settings_routes.py @@ -491,7 +491,8 @@ class SettingsRoutesMixin: for key in _THUMBNAIL_SETTINGS ) job = self._jobs.active(THUMBNAILS_JOB_KIND) if changed else None - if job is None or job.scope.get("mode") != "convert": + # A move copies previews of the current settings (#320). + if job is None or job.scope.get("mode") not in ("convert", "move"): return None return web.json_response( { diff --git a/py/api/thumbnail_auto.py b/py/api/thumbnail_auto.py index 5783fee3..59a80a67 100644 --- a/py/api/thumbnail_auto.py +++ b/py/api/thumbnail_auto.py @@ -775,6 +775,9 @@ class AutoThumbnailMixin: return job if job.kind == THUMBNAILS_JOB_KIND and job.title == THUMBNAIL_CLEAR_TITLE: return job + # Move thumbnails holds automatic previews for its whole run (D2A). + if job.kind == THUMBNAILS_JOB_KIND and job.scope.get("mode") == "move": + return job return None # ── the view admission ──────────────────────────────────────────── diff --git a/py/api/thumbnail_encode.py b/py/api/thumbnail_encode.py index 569f1e07..d5a847cd 100644 --- a/py/api/thumbnail_encode.py +++ b/py/api/thumbnail_encode.py @@ -114,6 +114,7 @@ def replace_unless_held(tmp, dst, guard=None): # ``temp_sibling``'s names: ``..--``. _TEMP_NAME = re.compile(r"^\..+\.(?:part|src)-(\d+)-\d+\.[A-Za-z0-9]+$") +_MOVE_TEMP_NAME = re.compile(r"\.pm-move-[0-9a-f]{8}\.tmp$") # A temp at least this old (seconds) outlived its write: no encode runs longer. STRAY_TEMP_AGE = 4 * ENCODE_TIMEOUT # When this process started (this module's import): a temp carrying its PID @@ -138,7 +139,9 @@ def sweep_stray_temps(folder, min_age=None): for root, _dirs, files in os.walk(folder): for name in files: match = _TEMP_NAME.match(name) - if match is None: + # A move's copy temp (#320): ``.pm-move-<8 hex>.tmp``. + move_temp = match is None and _MOVE_TEMP_NAME.search(name) + if match is None and not move_temp: continue path = Path(root) / name try: @@ -147,7 +150,13 @@ def sweep_stray_temps(folder, min_age=None): continue if min_age and ( now - stat.st_mtime < min_age - or (match.group(1) == own and stat.st_mtime >= PROCESS_STARTED) + or move_temp + and stat.st_mtime >= PROCESS_STARTED + or ( + match is not None + and match.group(1) == own + and stat.st_mtime >= PROCESS_STARTED + ) ): continue path.unlink() diff --git a/py/api/thumbnail_jobs.py b/py/api/thumbnail_jobs.py index 8c4d838e..7d7c8004 100644 --- a/py/api/thumbnail_jobs.py +++ b/py/api/thumbnail_jobs.py @@ -132,7 +132,10 @@ class ThumbnailJobsMixin: """The thumbnail plan, the thumbnail job and the legacy stream on it.""" async def get_thumbnail_plan(self, request): - """What a thumbnail run would do (read-only; capped like a run).""" + """What a thumbnail run would do (read-only; capped like a run). + ``mode=prune`` lists the removed sources' thumbnails (#320).""" + if request.query.get("mode") == "prune": + return await self.get_prune_plan() options, error = thumbnail_options(request.query, from_query=True) if error: return _invalid(error) @@ -180,6 +183,8 @@ class ThumbnailJobsMixin: ) except (ValueError, UnicodeDecodeError): return _invalid(_OPTIONS_ERROR) + if isinstance(body, dict) and body.get("mode") in ("move", "prune"): + return await self._start_location_job(body) options, error = thumbnail_options(body) if error: return _invalid(error) @@ -192,6 +197,25 @@ class ThumbnailJobsMixin: return busy_response(busy) return web.json_response({"success": True, "job": job.summary()}, status=202) + async def _start_location_job(self, body): + """Move thumbnails or Remove thumbnails of removed sources (#320).""" + if body["mode"] == "prune": + refused = location_refusal() + if refused is not None: + return refused + try: + if body["mode"] == "move": + started = self.start_thumbnail_move(body) + else: + started = self.start_thumbnail_prune() + except JobBusy as busy: + return busy_response(busy) + if isinstance(started, web.Response): + return started + return web.json_response( + {"success": True, "job": started.summary()}, status=202 + ) + def _start_thumbnail_job(self, options): """Register the thumbnail job synchronously; raises JobBusy. @@ -515,8 +539,12 @@ class ThumbnailJobsMixin: the way. """ running = self._jobs.active(THUMBNAILS_JOB_KIND) - if running is not None and running.scope.get("mode") == "convert": - return None, False, running # no legacy frames to follow (#317) + if running is not None and running.scope.get("mode") in ( + "convert", + "move", + "prune", + ): + return None, False, running # no legacy frames to follow (#317, #320) if running is not None and running.cancellable: return running, False, None busy = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded") diff --git a/py/api/thumbnail_location.py b/py/api/thumbnail_location.py index 437adf3c..210b39a7 100644 --- a/py/api/thumbnail_location.py +++ b/py/api/thumbnail_location.py @@ -157,4 +157,6 @@ class ThumbnailLocationMixin: if with_path: payload["path"] = location.path payload["default_path"] = default_path() + # Old copies a move still has to remove (#320 Part 3). + payload["cleanup_pending"] = len(preview_store.setting().cleanup) return payload diff --git a/py/api/thumbnail_move.py b/py/api/thumbnail_move.py new file mode 100644 index 00000000..3d4d6472 --- /dev/null +++ b/py/api/thumbnail_move.py @@ -0,0 +1,523 @@ +"""Move thumbnails (#320 Part 3): the reviewed job that changes where previews +live, the removal of old copies after it, recovery at startup and the +"Remove thumbnails of removed sources" scope (Part 5). + +Kind ``thumbnails`` with ``mode: "move"`` (D1A): admission ``guarded``, so +scans, deletes, tagging, source changes and every other thumbnail job wait +for it, while the monitor's Needs review recording runs beside it. +Automatic previews are held for the whole job (D2A, ``_auto_thumbnail_blocker``). + +Phases: review (``POST /images/thumbnails/plan``, no writes but the +validation probe), start (a plan token), hold (generation bump, journal, +target folder), copy (cancellable at every file boundary, on the job pool +in small chunks, D3A), switch (one config commit sets the location and +records the old stores under ``cleanup``; not cancellable from here), +remove old copies, end. Per-file errors switch anyway (D10A); a folder-wide +error or a cancel before the switch removes what the job wrote and changes +nothing. The move is the only writer of the setting (D4B). + +Moving away from a thumbnail folder also removes the previews of this +install's sources that are no longer listed (Q320-8: remove). +""" + +import functools +import os +import time +from pathlib import Path + +from aiohttp import web + +from . import preview_location, preview_store +from . import thumbnail_move_files as files +from .jobs import THUMBNAILS_JOB_KIND +from .thumbnail_move_plan import ( + MOVE_GROUPS, + MOVE_TITLE, + PLAN_TTL, + _number, + _plural, + _remove, + _roots_revision, + _volume, + _is_network, + _write_json, +) +from . import thumbnail_move_plan as move_plan + +try: + from ...utils import exit_signal + from ...utils.parallel import map_parallel +except ImportError: # tests import the package top-level + from utils import exit_signal + from utils.parallel import map_parallel + + +class ThumbnailMoveMixin: + """Start and run the move job; the removal of old copies.""" + + # ── start ───────────────────────────────────────────────────────── + + def _take_move_plan(self, body): + """The plan a start names, or ``(status, code)``.""" + token = body.get("token") + plan = (self._move_plans or {}).get(token) if isinstance(token, str) else None + if plan is None or time.monotonic() - plan.created > PLAN_TTL: + return 409, "plan-stale" + roots = [str(root) for root in self._get_all_output_dirs()] + if _roots_revision(roots) != plan.revision: + return 409, "plan-stale" + if preview_store.setting() != plan.old: + return 409, "plan-stale" + if plan.confirm_required and body.get("confirmed") is not True: + return 400, "confirmation-required" + return plan + + def start_thumbnail_move(self, body): + """Start the move job (sync, on the loop); raises JobBusy. Returns + the job or a refusal response.""" + if not self._move_recovered: + blocker = self._jobs.active(THUMBNAILS_JOB_KIND) + return web.json_response( + { + "success": False, + "code": "busy", + "error": "Prompt Manager is still checking an earlier move. " + "Try again in a moment.", + "running": None if blocker is None else {"id": blocker.id}, + }, + status=409, + ) + taken = self._take_move_plan(body) + if isinstance(taken, tuple): + status, code = taken + text = { + "plan-stale": "The review is out of date. Review the move again.", + "confirmation-required": "Confirm that Prompt Manager may " + "create its folders there.", + }[code] + return web.json_response( + {"success": False, "code": code, "error": text}, status=status + ) + plan = taken + self._move_plans.pop(plan.token, None) + return self._jobs.start( + THUMBNAILS_JOB_KIND, + MOVE_TITLE, + admission="guarded", + cancellable=True, + scope={ + "mode": "move", + "from": plan.old.mode, + "to": plan.target["mode"], + "to_path": plan.target["path"], + }, + groups=MOVE_GROUPS, + runner=functools.partial(self._run_thumbnail_move, plan=plan), + ) + + # ── the job ─────────────────────────────────────────────────────── + + async def _run_thumbnail_move(self, job, *, plan): + state = {"written": [], "created": [], "created_folder": False} + try: + await self._run_in_executor(self._move_hold, job, plan, state) + except (OSError, ValueError) as error: + await self._run_in_executor(self._move_rollback, plan, state) + self._end_move( + job, "failed", "The new folder could not be made; nothing changed." + ) + self.logger.warning(f"Move thumbnails could not start: {error}") + return + items = [ + (source, item) + for source in plan.sources + if source.scan is not None + for item in source.scan.copy + ] + total = len(items) + job.update(total=total, processed=0, progress=0) + chunk = self._move_chunk(plan) + workers = self._move_workers() + done = 0 + for start in range(0, total, chunk): + if job.cancelled: + return await self._move_cancelled(job, plan, state) + batch = items[start : start + chunk] + try: + results = await self._run_in_executor( + map_parallel, + functools.partial(self._move_copy_one, state=state), + batch, + workers, + ) + except exit_signal.ExitRequested: + return # exiting: the next start undoes the copies (journal) + except files.FolderGone as error: + self.logger.warning(f"Move thumbnails stopped: {error}") + await self._run_in_executor(self._move_rollback, plan, state) + self._end_move( + job, + "failed", + "The new folder could not be reached or is full; nothing changed.", + ) + return + for (source, item), (outcome, detail) in zip(batch, results): + row = { + "path": "/".join(item[0]), + "root": source.label, + "detail": detail, + } + job.record( + "moved" if outcome == "moved" else outcome, + row, + f"{outcome}: {source.label} · {'/'.join(item[0])}", + ) + done += len(batch) + label = batch[-1][0].label + job.update( + processed=done, + progress=int(done * 90 / total) if total else 90, + status_line=( + f"Copying {_number(done)} of {_number(total)} previews · {label}" + ), + ) + job.publish({"type": "progress", "summary": job.summary()}) + if job.cancelled: + return await self._move_cancelled(job, plan, state) + # The switch: not cancellable from here (an exit still stops between + # files and leaves the cleanup to the next start). + job.cancellable = False + job.update(status_line="Switching to the new location", progress=92) + job.publish({"type": "progress", "summary": job.summary()}) + try: + await self._run_in_executor(self._move_switch, plan) + except Exception: + self.logger.exception("Move thumbnails could not switch") + await self._run_in_executor(self._move_rollback, plan, state) + self._end_move( + job, "failed", "The setting could not be saved; nothing changed." + ) + return + await self._remove_old_copies(job, plan) + moved = job.counters.get("moved", 0) + job.counters.get("skipped", 0) + errors = job.counters.get("errors", 0) + where = plan.target["path"] or "inside each source" + parts = [f"{_number(moved)} moved"] + if errors: + parts.append(_plural(errors, "error")) + parts.append( + "old copies removed" + if not job.scope.get("cleanup_left") + else "some old copies are removed at the next start" + ) + self._end_move( + job, "finished", f"Thumbnails moved to {where}: " + " · ".join(parts) + "." + ) + + def _end_move(self, job, state, line): + job.update(status_line=line, progress=100) + job.log_line(line) + job.publish({"type": "complete", "summary": job.summary()}) + job.finish(state) + + def _move_workers(self): + from .images import _worker_threads + + return max(1, int(_worker_threads() or 1)) + + def _move_chunk(self, plan): + """2 files at a time across volumes or to/from a share, else 4, never + more than Worker threads (D3A: the pool keeps its size).""" + sides = [plan.target["path"] or ""] + [s.old_store or "" for s in plan.sources] + sides += [s.new_store for s in plan.sources] + network = any(_is_network(side) for side in sides if side) + volumes = {_volume(side) for side in sides if side} + size = 2 if network or len(volumes) > 1 else 4 + return max(1, min(size, self._move_workers())) + + def _move_hold(self, job, plan, state): + """Phase 3: bump the generation, write the journal, make the target.""" + preview_store.bump_generation() + if plan.to_folder: + folder = plan.target["path"] + if not os.path.isdir(folder): + if plan.target_default: + os.makedirs(folder, exist_ok=True) + else: + os.mkdir(folder) + state["created_folder"] = True + plan.target_id = preview_location.init_folder( + folder, created_by_pm=state["created_folder"] + ) + install = plan.install + + def add_keys(marker): + for source in plan.sources: + if source.key and source.key not in marker["keys"]: + marker["keys"][source.key] = { + "path": source.root, + "identity": os.path.normcase(os.path.realpath(source.root)), + "dev_ino": preview_location.file_identity(source.root), + "install": install, + } + + with preview_store.write_lock: + preview_location.write_marker(folder, add_keys) + _write_json( + move_plan.journal_path(), + { + "state": "copying", + "from": plan.old.to_config(), + "to": plan.target, + "stores": [s.new_store for s in plan.sources], + # Stores that do not exist yet are made by this job only: + # after a crash everything in them goes (B4). + "created_dirs": [ + s.new_store for s in plan.sources if not os.path.isdir(s.new_store) + ], + "created_folder": state["created_folder"], + "install": plan.install, + }, + ) + job.log_line( + f"Move thumbnails started: {plan.old.path or 'inside each source'} → " + f"{plan.target['path'] or 'inside each source'}" + ) + + def _move_copy_one(self, pair, *, state): + """Copy one preview (a pool thread); ``(group, detail)``.""" + source, (relative, size, mtime_ns) = pair + destination = Path(source.new_store).joinpath(*relative) + try: + if not os.path.isdir(destination.parent): + base = Path(source.new_store) + if not base.is_dir(): + if plan_parent_missing(base): + raise files.FolderGone("the new folder is gone") + created = [] + files._make_folders(base.parent, (base.name,), created) + state["created"].extend(created) + created = [] + files._make_folders(base, relative[:-1], created) + state["created"].extend(created) + outcome = files.copy_one( + Path(source.old_store).joinpath(*relative), + destination, + size, + mtime_ns, + state["created"], + ) + except files.CopyFailed as error: + return "errors", str(error) + if outcome == "moved": + state["written"].append(str(destination)) + return "moved", "Copied and checked" + return "skipped", "Already there" + + async def _move_cancelled(self, job, plan, state): + if exit_signal.requested(): + return # the next start removes the partial copies (journal) + await self._run_in_executor(self._move_rollback, plan, state) + self._end_move(job, "cancelled", "Move cancelled: nothing changed.") + + def _move_rollback(self, plan, state): + """Remove what this job wrote (files, folders it made, a folder and + marker it created) and the journal: nothing changed.""" + for path in state["written"]: + files.remove_file(path) + for folder in reversed(state["created"]): + files.remove_empty_folders(folder) + if plan.to_folder and state.get("created_folder"): + folder = plan.target["path"] + try: + names = os.listdir(folder) + except OSError: + names = None + if names is not None and set(names) <= {preview_location.MARKER_NAME}: + _remove(preview_location.marker_path(folder)) + files.remove_empty_folders(folder) + _remove(move_plan.journal_path()) + + def _move_switch(self, plan): + """Phase 5: one config commit sets the location and records the old + stores under ``cleanup``; the runtime follows under the write lock.""" + from ..config import PromptManagerConfig + from ..config_store import get_config_store + + entry = { + "from": plan.old.mode, + "folder": plan.old.path if plan.from_folder else None, + "stores": [s.old_store for s in plan.sources if s.old_store], + "drop_keys": self._old_keys_to_drop(plan), + "install": plan.install, + } + # An older entry never removes a store the move just filled (a move + # back while old copies still waited). + filled = {os.path.normcase(s.new_store) for s in plan.sources} + cleanup = [] + for old_entry in plan.old.cleanup: + stores = [ + store + for store in old_entry.get("stores") or [] + if isinstance(store, str) and os.path.normcase(store) not in filled + ] + cleanup.append({**old_entry, "stores": stores}) + cleanup.append(entry) + if plan.to_folder: + value = { + "mode": preview_store.FOLDER_MODE, + "path": plan.target["path"], + "id": plan.target_id, + "install": plan.install, + "cleanup": cleanup, + } + else: + value = { + "mode": preview_store.SOURCE_MODE, + "install": plan.install, + "cleanup": cleanup, + } + + def mutate(config): + section = config.get("thumbnails") + if not isinstance(section, dict): + section = {} + section["location"] = value + config["thumbnails"] = section + return config + + def publish(_config): + PromptManagerConfig.THUMBNAIL_LOCATION = value + preview_store.configure(value, check=False) + + get_config_store().commit(mutate, expected_revision=None, publish=publish) + preview_store.refresh() + _remove(move_plan.journal_path()) + + def _old_keys_to_drop(self, plan): + if not plan.from_folder: + return [] + keys = [s.old_store and os.path.basename(s.old_store) for s in plan.sources] + keys = [key for key in keys if key] + if plan.to_folder and os.path.normcase(plan.old.path or "") == os.path.normcase( + plan.target["path"] or "" + ): + return [] + return keys + [key for key, _store, _scan in plan.unlisted] + + async def _remove_old_copies(self, job, plan): + """Phase 6: run the cleanup the switch recorded (and earlier ones).""" + job.update(status_line="Removing old copies", progress=95) + job.publish({"type": "progress", "summary": job.summary()}) + left = await self._run_in_executor(self._run_cleanup, job) + if left: + job.scope["cleanup_left"] = True + + def _run_cleanup(self, job=None): + """Remove every recorded old store's previews, temps and empty + folders; keep in ``cleanup`` what could not go. Returns how many + entries are left (blocking).""" + setting = preview_store.setting() + remaining = [] + for entry in setting.cleanup: + if not self._cleanup_entry(entry, job): + remaining.append(entry) + if list(setting.cleanup) != remaining: + self._commit_cleanup(remaining) + return len(remaining) + + def _cleanup_entry(self, entry, job): + """One recorded removal; True when it is complete.""" + folder = entry.get("folder") + stores = [ + store for store in entry.get("stores") or [] if isinstance(store, str) + ] + if folder: + if not os.path.isdir(folder): + return False # the old folder is not reachable: kept for later + folder_marker = preview_location.read_marker(folder) or {} + keys = folder_marker.get("keys") or {} + stores += [ + os.path.join(folder, key) + for key in entry.get("drop_keys") or [] + if preview_location.valid_key(key) and key in keys + ] + complete = True + for store in dict.fromkeys(stores): + if not self._remove_store(store, folder, job): + complete = False + if folder and complete: + drop = set(entry.get("drop_keys") or []) + + def forget(marker): + for key in drop: + marker["keys"].pop(key, None) + + try: + with preview_store.write_lock: + marker = preview_location.write_marker(folder, forget) + except (OSError, ValueError): + return False + if not marker["keys"] and marker.get("created_by_pm"): + names = set(os.listdir(folder)) + rest = {n for n in names if not files.is_os_metadata(n)} + if rest <= {preview_location.MARKER_NAME}: + _remove(preview_location.marker_path(folder)) + files.remove_empty_folders(folder) + return complete + + def _remove_store(self, store, folder, job): + """Remove the previews and temps of one old store; True when done.""" + store = Path(store) + if folder and not preview_location.valid_key(store.name): + return True + if not store.exists(): + return True + if preview_location.is_link_like(store): + return True # never through a link; nothing of ours there + complete = True + removed = 0 + for current, dirs, names in os.walk(store): + dirs[:] = [ + d for d in dirs if not preview_location.is_link_like(Path(current) / d) + ] + for name in names: + path = Path(current) / name + if files.is_temp_name(name) or files._preview_parts(name): + if preview_location.is_link_like(path): + continue + if files.remove_file(path): + removed += 1 + else: + complete = False + if job is not None: + job.log_line( + "In use by another program; removed at the next " + f"start: {name}" + ) + files.remove_empty_folders(store) + if job is not None and removed: + job.log_line(f"Removed {_number(removed)} old files from {store.name}") + return complete + + def _commit_cleanup(self, remaining): + from ..config import PromptManagerConfig + from ..config_store import get_config_store + + def mutate(config): + section = config.get("thumbnails") + if isinstance(section, dict) and isinstance(section.get("location"), dict): + section["location"]["cleanup"] = remaining + return config + + def publish(config): + value = (config.get("thumbnails") or {}).get("location") + PromptManagerConfig.THUMBNAIL_LOCATION = value + preview_store.set_cleanup(remaining) + + get_config_store().commit(mutate, expected_revision=None, publish=publish) + + +def plan_parent_missing(store): + """A key folder's parent (the thumbnail folder) is gone mid-copy.""" + return not Path(store).parent.is_dir() diff --git a/py/api/thumbnail_move_files.py b/py/api/thumbnail_move_files.py new file mode 100644 index 00000000..ca5b147a --- /dev/null +++ b/py/api/thumbnail_move_files.py @@ -0,0 +1,316 @@ +"""The file side of Move thumbnails (#320 Part 3): what is copied, how a +copy is made and checked, and how old copies are removed. + +No job, HTTP or config code lives here; ``thumbnail_move`` drives it. + +Copy set (B5): each old store is walked with ``os.scandir``, never through a +symlink, junction or reparse point (counted as "skipped link"). A regular +file is + +- copied when it has the preview grammar (``_thumb``, any written + or legacy extension) and its original still exists in the source; +- removed with the old copies, not copied, when it is an orphan (no + original), lies under a trash folder, or is a temp sibling of a write + (``..part-…``) or of a move (``….pm-move-<8 hex>.tmp``); +- left in place otherwise ("Left in place"). OS metadata files (``Thumbs.db``, + ``desktop.ini``, ``.DS_Store``, ``._*``) are removed only when they alone + keep a preview folder from being removed. + +A copy goes to ``.pm-move-<8 hex>.tmp`` beside its destination, its +size is compared, its mtime is set to the source preview's rounded up to the +next 2-second step (FAT/exFAT keep it fresh, a stale preview stays stale), +then it is renamed into place. Every handle is closed before a rename or an +unlink (Windows). +""" + +import errno +import os +import re +import secrets +import shutil +import stat as stat_module +import time +from dataclasses import dataclass, field +from pathlib import Path + +from . import preview_location, thumbnail_settings +from .thumbnail_encode import _TEMP_NAME + +try: + from ...utils.media_folders import in_trash_folder +except ImportError: # tests import the package top-level + from utils.media_folders import in_trash_folder + +MOVE_TEMP = re.compile(r"\.pm-move-[0-9a-f]{8}\.tmp$") +MOVE_TEMP_SUFFIX = ".pm-move-{token}.tmp" +# The temp name adds this much to a destination path (".pm-move-xxxxxxxx.tmp"). +TEMP_EXTRA = 21 +WINDOWS_FILE_LIMIT = 259 +MTIME_STEP_NS = 2_000_000_000 +# A file another program holds open (Windows) is retried this often. +REMOVE_RETRIES = (0.5, 1.0, 2.0) +_WINDOWS_IN_USE = {32, 33} # ERROR_SHARING_VIOLATION, ERROR_LOCK_VIOLATION + + +def is_os_metadata(name): + folded = name.lower() + return folded in preview_location.OS_METADATA or folded.startswith("._") + + +def is_temp_name(name): + return bool(_TEMP_NAME.match(name) or MOVE_TEMP.search(name)) + + +def _preview_parts(name): + """``(base, extension)`` of a preview file name, or None.""" + stem, extension = os.path.splitext(name) + suffix = thumbnail_settings.THUMB_SUFFIX + if not stem.endswith(suffix) or len(stem) <= len(suffix): + return None + if extension.lower() not in thumbnail_settings.THUMBNAIL_EXTENSIONS: + return None + return stem[: -len(suffix)], extension + + +def _has_original(root, rel_dir, base, extension): + """Whether a media file in ``/`` owns this preview.""" + folder = Path(root).joinpath(*rel_dir) + try: + with os.scandir(folder) as entries: + for entry in entries: + stem, source_ext = os.path.splitext(entry.name) + if stem != base or not source_ext: + continue + if not thumbnail_settings.owns(extension, source_ext): + continue + if entry.is_file(follow_symlinks=False): + return True + except OSError: + return False + return False + + +@dataclass +class StoreScan: + """One old store's files, sorted into what the move does with them.""" + + store: Path + copy: list = field(default_factory=list) # (relative parts, size, mtime_ns) + orphans: list = field(default_factory=list) # relative parts + temps: list = field(default_factory=list) + left: list = field(default_factory=list) + links: int = 0 + bytes: int = 0 + + @property + def removable(self): + return [item[0] for item in self.copy] + self.orphans + self.temps + + +def scan_store(store, root, *, stop=None): + """Sort the files of *store* (the previews of the source *root*). + + A store that does not exist, or is itself a link, scans empty (a link + counts once). + """ + store = Path(store) + scan = StoreScan(store) + try: + if preview_location.is_link_like(store): + scan.links += 1 + return scan + if not store.is_dir(): + return scan + except OSError: + return scan + _walk(scan, store, (), root, stop) + return scan + + +def _walk(scan, folder, rel_dir, root, stop): + try: + with os.scandir(folder) as iterator: + entries = list(iterator) + except OSError: + return + for entry in entries: + if stop is not None and stop(): + return + relative = rel_dir + (entry.name,) + path = Path(entry.path) + try: + if preview_location.is_link_like(path): + scan.links += 1 + continue + info = entry.stat(follow_symlinks=False) + except OSError: + continue + if stat_module.S_ISDIR(info.st_mode): + _walk(scan, path, relative, root, stop) + continue + if not stat_module.S_ISREG(info.st_mode): + scan.left.append(relative) + continue + if is_temp_name(entry.name): + scan.temps.append(relative) + continue + if is_os_metadata(entry.name): + continue # removed only when it alone keeps a folder (rmdir) + parts = _preview_parts(entry.name) + if parts is None: + scan.left.append(relative) + continue + if in_trash_folder("/".join(relative)) or not _has_original( + root, rel_dir, *parts + ): + scan.orphans.append(relative) + continue + scan.copy.append((relative, info.st_size, info.st_mtime_ns)) + scan.bytes += info.st_size + + +# ── copying ────────────────────────────────────────────────────────── + + +def rounded_mtime_ns(mtime_ns): + """*mtime_ns* rounded up to the next 2-second step (FAT/exFAT).""" + return -(-mtime_ns // MTIME_STEP_NS) * MTIME_STEP_NS + + +def too_long(path, long_paths=None): + """A destination (with its temp name) over Windows' file limit.""" + if os.name != "nt" and long_paths is not False: + return False + if long_paths is None: + long_paths = preview_location.long_paths_enabled() + if long_paths: + return False + return len(str(path)) + TEMP_EXTRA > WINDOWS_FILE_LIMIT + + +class CopyFailed(Exception): + """One preview could not be copied (a per-file error, D10A).""" + + +class FolderGone(Exception): + """The destination cannot take writes any more (abort the move).""" + + +def _make_folders(base, relative_dir, created): + """Make ``base/relative_dir`` one folder at a time (*base* must exist); + every folder made is added to *created*.""" + current = Path(base) + for part in relative_dir: + current = current / part + try: + os.mkdir(current) + created.append(str(current)) + except FileExistsError: + pass + if preview_location.is_link_like(current): + raise CopyFailed("a link in the destination") + + +def copy_one(source, destination, size, mtime_ns, created, *, long_paths=None): + """Copy one preview; returns ``"moved"`` or ``"skipped"`` (already + there). Raises CopyFailed (this file) or FolderGone (the destination).""" + destination = Path(destination) + if too_long(destination, long_paths): + raise CopyFailed("Path too long for Windows; the original is shown") + wanted = rounded_mtime_ns(mtime_ns) + try: + existing = os.stat(destination, follow_symlinks=False) + except FileNotFoundError: + existing = None + except OSError as error: + raise FolderGone(str(error)) from error + if ( + existing is not None + and stat_module.S_ISREG(existing.st_mode) + and existing.st_size == size + and existing.st_mtime_ns == wanted + ): + return "skipped" + temp = destination.with_name( + destination.name + MOVE_TEMP_SUFFIX.format(token=secrets.token_hex(4)) + ) + try: + shutil.copyfile(source, temp) + if os.stat(temp).st_size != size: + raise CopyFailed("The copy did not match; made again on view") + os.utime(temp, ns=(wanted, wanted)) + os.replace(temp, destination) + except CopyFailed: + _unlink(temp) + raise + except FileNotFoundError as error: + _unlink(temp) + if not Path(source).exists(): + raise CopyFailed("Could not be read; made again on view") from error + raise FolderGone(str(error)) from error + except OSError as error: + _unlink(temp) + if error.errno in (errno.ENOSPC, errno.EDQUOT, errno.EROFS): + raise FolderGone(str(error)) from error + raise CopyFailed("Could not be copied; made again on view") from error + return "moved" + + +def _unlink(path): + try: + os.unlink(path) + except OSError: + pass + + +# ── removing ───────────────────────────────────────────────────────── + + +def _in_use(error): + return getattr(error, "winerror", None) in _WINDOWS_IN_USE + + +def remove_file(path, *, sleep=time.sleep): + """Remove one file; a file another program holds open (Windows) is + retried 3 times. Returns True when it is gone.""" + for wait in (0.0,) + REMOVE_RETRIES: + if wait: + sleep(wait) + try: + os.unlink(path) + return True + except FileNotFoundError: + return True + except OSError as error: + if not _in_use(error): + return False + return False + + +def remove_empty_folders(top, *, keep_top=False): + """``rmdir`` the empty folders below *top* (and *top*), bottom-up, never + through a link. A folder holding only OS metadata files loses them first. + Returns the folders that could not be removed.""" + top = Path(top) + left = [] + if not top.is_dir() or preview_location.is_link_like(top): + return left + for folder, dirs, _files in os.walk(top, topdown=False): + folder = Path(folder) + if folder == top and keep_top: + continue + if preview_location.is_link_like(folder): + continue + try: + names = os.listdir(folder) + except OSError: + left.append(folder) + continue + if names and all(is_os_metadata(name) for name in names): + for name in names: + remove_file(folder / name) + try: + os.rmdir(folder) + except OSError: + left.append(folder) + return left diff --git a/py/api/thumbnail_move_plan.py b/py/api/thumbnail_move_plan.py new file mode 100644 index 00000000..973a057d --- /dev/null +++ b/py/api/thumbnail_move_plan.py @@ -0,0 +1,490 @@ +"""Move thumbnails, the review (#320 Part 3 phase 1): what a move would copy, +remove and leave, checked before anything is written.""" + +import hashlib +import json +import os +import secrets +import shutil +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path + +from aiohttp import web + +from . import preview_location, preview_store +from . import thumbnail_move_files as files +from .job_routes import busy_response +from .jobs import THUMBNAILS_JOB_KIND, JobBusy +from .thumbnail_location import default_path + +try: + from ...utils import library_location +except ImportError: # tests import the package top-level + from utils import library_location + +MOVE_TITLE = "Move thumbnails" +FINISHING_TITLE = "Move thumbnails (finishing)" +PRUNE_TITLE = "Remove thumbnails of removed sources" +MOVE_GROUPS = ("moved", "skipped", "left", "errors") +PRUNE_GROUPS = ("removed", "left", "errors") +PLAN_TTL = 600.0 +JOURNAL_NAME = "thumbnail-move.json" +SPACE_MARGIN = 64 * 2**20 +SPACE_SHARE = 0.05 +LISTED = 5 +SENTENCE = ( + "Every thumbnail is copied and checked before Prompt Manager switches. " + "Scans, deletes and tagging wait until it finishes; closing this page does " + "not stop it." +) + + +def _number(value): + return f"{value:,}" + + +def _plural(count, one, many=None): + return f"{_number(count)} {one if count == 1 else (many or one + 's')}" + + +def _size(value): + size = float(value) + for unit in ("B", "KB", "MB", "GB"): + if size < 1024 or unit == "GB": + return f"{size:.1f} {unit}" if unit != "B" else f"{int(size)} B" + size /= 1024 + return f"{size:.1f} GB" + + +def _refusal(code, status=None, **extra): + known_status, text = preview_location.MESSAGES.get(code, (400, code)) + return web.json_response( + {"success": False, "code": code, "error": text, **extra}, + status=status or known_status, + ) + + +def journal_path(): + """``/data/thumbnail-move.json`` (only while a move copies).""" + return library_location.cache_dir(JOURNAL_NAME) + + +def _read_json(path): + try: + with open(path, "r", encoding="utf-8") as handle: + data = json.load(handle) + except (OSError, ValueError): + return None + return data if isinstance(data, dict) else None + + +def _write_json(path, data): + os.makedirs(os.path.dirname(path), exist_ok=True) + temp = f"{path}.{os.getpid()}.tmp" + with open(temp, "w", encoding="utf-8") as handle: + json.dump(data, handle, indent=2) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp, path) + + +def _remove(path): + try: + os.unlink(path) + except FileNotFoundError: + pass + + +def _is_network(path): + text = str(path) + if text.startswith(("\\\\", "//")): + return True + if os.name == "nt": + try: + import ctypes + + drive = os.path.splitdrive(os.path.abspath(text))[0] + "\\" + return ctypes.windll.kernel32.GetDriveTypeW(drive) == 4 # DRIVE_REMOTE + except (AttributeError, OSError): + return False + return False + + +def _volume(path): + current = Path(os.path.abspath(path)) + while not current.exists() and current.parent != current: + current = current.parent + try: + return os.stat(current).st_dev + except OSError: + return None + + +def _free(path): + current = Path(os.path.abspath(path)) + while not current.exists() and current.parent != current: + current = current.parent + try: + return shutil.disk_usage(current).free + except OSError: + return None + + +# ── the plan ───────────────────────────────────────────────────────── + + +@dataclass +class SourcePlan: + """One listed source: its old store, its new store and what moves.""" + + root: str + label: str + key: str | None + old_store: str | None + new_store: str + scan: files.StoreScan | None + available: bool = True + too_long: list = field(default_factory=list) + + +@dataclass +class MovePlan: + """A reviewed move: from, to, the sources and the removals.""" + + token: str + created: float + revision: str + old: preview_location.ParsedLocation + target: dict # {"mode", "path"} + target_id: str | None + target_exists: bool + target_default: bool + install: str + sources: list + unlisted: list # (key, store path, scan) + needs_bytes: int + confirm_required: bool + + @property + def to_folder(self): + return self.target["mode"] == preview_store.FOLDER_MODE + + @property + def from_folder(self): + return self.old.mode == preview_store.FOLDER_MODE + + +def _roots_revision(roots): + text = json.dumps(sorted(str(root) for root in roots)) + return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] + + +class ThumbnailMovePlanMixin: + """The move's review and its routes.""" + + _move_plans = None + + # ── routes ──────────────────────────────────────────────────────── + + def _register_thumbnail_move_routes(self, routes): + @routes.post("/prompt_manager/images/thumbnails/plan") + async def thumbnail_move_plan_route(request): + return await self.plan_thumbnail_move(request) + + @routes.post("/prompt_manager/images/thumbnails/location/recreate") + async def thumbnail_location_recreate_route(request): + return await self.recreate_thumbnail_folder(request) + + @routes.post("/prompt_manager/images/thumbnails/location/adopt") + async def thumbnail_location_adopt_route(request): + return await self.adopt_thumbnail_folder(request) + + # ── review ──────────────────────────────────────────────────────── + + async def plan_thumbnail_move(self, request): + try: + body = await request.json() + except (ValueError, UnicodeDecodeError): + body = None + if not isinstance(body, dict) or body.get("mode") != "move": + return web.json_response( + { + "success": False, + "code": "invalid-request", + "error": "mode must be move", + }, + status=400, + ) + target = body.get("location") + if not isinstance(target, dict) or target.get("mode") not in ( + preview_store.SOURCE_MODE, + preview_store.FOLDER_MODE, + ): + return web.json_response( + { + "success": False, + "code": "invalid-request", + "error": "location must name a mode", + }, + status=400, + ) + blocker = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded") + if blocker is not None: + return busy_response(JobBusy(blocker)) + try: + outcome = await self._run_in_executor(self._build_move_plan, target) + except Exception: + self.logger.exception("The thumbnail move could not be planned") + return web.json_response( + { + "success": False, + "code": None, + "error": "The move could not be planned", + }, + status=500, + ) + if isinstance(outcome, str): + return _refusal(outcome) + plan, payload = outcome + if payload.get("insufficient"): + return _refusal( + "insufficient-space", + status=507, + error_detail=payload["insufficient"], + plan=payload, + ) + self._store_move_plan(plan) + return web.json_response({"success": True, **payload}) + + def _store_move_plan(self, plan): + if self._move_plans is None: + self._move_plans = {} + now = time.monotonic() + self._move_plans = { + token: kept + for token, kept in self._move_plans.items() + if now - kept.created < PLAN_TTL + } + self._move_plans[plan.token] = plan + + def _build_move_plan(self, target): + """``(plan, payload)`` or a refusal code (blocking).""" + old = preview_store.setting() + location = preview_store.current() + install = old.install or str(uuid.uuid4()) + target_path = None + default = default_path() + if target["mode"] == preview_store.FOLDER_MODE: + raw = target.get("path") + if not isinstance(raw, str): + return "location-not-absolute" + context = self._thumbnail_location_context() + code = preview_location.location_problem( + raw, purpose="move", context=context + ) + if code is not None: + return code + target_path = os.path.abspath(raw) + elif old.mode == preview_store.SOURCE_MODE and not old.problem: + return "location-same" + target_default = bool(default) and os.path.normcase( + target_path or "" + ) == os.path.normcase(os.path.abspath(default)) + target_exists = bool(target_path) and os.path.isdir(target_path) + target_marker = ( + preview_location.read_marker(target_path) if target_exists else None + ) + roots = [str(root) for root in self._get_all_output_dirs()] + labels = self._source_labels(roots) + old_keys = {} + if old.mode == preview_store.FOLDER_MODE and location.available: + marker = preview_location.read_marker(old.path) or {} + old_keys = marker.get("keys") or {} + new_keys = dict((target_marker or {}).get("keys") or {}) + taken = {key: entry.get("identity") for key, entry in new_keys.items()} + sources = [] + for root in roots: + identity = os.path.normcase(os.path.realpath(root)) + dev_ino = preview_location.file_identity(root) + old_key = None + old_store = None + available = True + if old.mode == preview_store.SOURCE_MODE: + old_store = str(preview_store.source_folder(root)) + elif location.available: + old_key = preview_location.find_key(old_keys, root, identity, dev_ino) + if old_key: + old_store = os.path.join(old.path, old_key) + else: + available = False + key = None + if target_path: + key = old_key or preview_location.find_key( + new_keys, root, identity, dev_ino + ) + if key is None or taken.get(key, identity) != identity: + key = preview_location.source_key(root, identity, taken) + taken[key] = identity + new_store = os.path.join(target_path, key) + else: + new_store = str(preview_store.source_folder(root)) + scan = None + if old_store: + scan = files.scan_store(old_store, root) + plan = SourcePlan( + root, labels.get(root, root), key, old_store, new_store, scan, available + ) + if scan is not None and target_path: + plan.too_long = [ + item[0] + for item in scan.copy + if files.too_long(Path(new_store).joinpath(*item[0])) + ] + sources.append(plan) + unlisted = self._unlisted_keys(old, location, old_keys, install) + needs = sum(source.scan.bytes for source in sources if source.scan is not None) + plan = MovePlan( + token=secrets.token_urlsafe(16), + created=time.monotonic(), + revision=_roots_revision(roots), + old=old, + target={"mode": target["mode"], "path": target_path}, + target_id=(target_marker or {}).get("id"), + target_exists=target_exists, + target_default=target_default, + install=install, + sources=sources, + unlisted=unlisted, + needs_bytes=needs, + confirm_required=bool(target_path) and not target_default, + ) + return plan, self._plan_payload(plan, location) + + def _unlisted_keys(self, old, location, keys, install): + """This install's keys whose source no longer is in any root list + (``(key, store, scan)``), in the folder in use.""" + if old.mode != preview_store.FOLDER_MODE or not location.available: + return [] + listed = { + os.path.normcase(os.path.realpath(root)) + for root in self._all_source_roots() + } + found = [] + for key, entry in sorted(keys.items()): + if entry.get("install") not in (install, None): + continue + if entry.get("identity") in listed: + continue + store = os.path.join(old.path, key) + found.append( + (key, store, files.scan_store(store, entry.get("path") or store)) + ) + return found + + def _source_labels(self, roots): + from . import sources + + labels = sources._comfyui_labels() + return {root: sources.source_label(root, labels) for root in roots} + + def _plan_payload(self, plan, location): + sources = plan.sources + copies = sum(len(s.scan.copy) for s in sources if s.scan) + orphans = sum(len(s.scan.orphans) for s in sources if s.scan) + temps = sum(len(s.scan.temps) for s in sources if s.scan) + links = sum(s.scan.links for s in sources if s.scan) + left = [ + f"{s.label} · {'/'.join(item)}" + for s in sources + if s.scan + for item in s.scan.left + ] + too_long = [ + f"{s.label} · {'/'.join(item)}" for s in sources for item in s.too_long + ] + unlisted_count = sum( + len(scan.copy) + len(scan.orphans) for _k, _s, scan in plan.unlisted + ) + free = None + insufficient = None + if plan.needs_bytes: + volumes = {} + for source in sources: + if source.scan is None or not source.scan.bytes: + continue + volume = _volume(source.new_store) + entry = volumes.setdefault(volume, [source.new_store, 0]) + entry[1] += source.scan.bytes + for where, needed in volumes.values(): + available = _free(where) + if available is None: + continue + free = available if free is None else min(free, available) + margin = max(SPACE_MARGIN, int(needed * SPACE_SHARE)) + if needed + margin > available: + insufficient = { + "needed": needed + margin, + "free": available, + "text": ( + f"There is not enough space for the thumbnails: " + f"{_size(available)} free, {_size(needed + margin)} " + "needed. Free some space or choose another folder." + ), + } + payload = { + "mode": "move", + "from": { + "mode": plan.old.mode, + "path": plan.old.path, + "available": location.available, + }, + "to": { + "mode": plan.target["mode"], + "path": plan.target["path"], + "exists": plan.target_exists, + "will_create": bool(plan.target["path"]) and not plan.target_exists, + "default": plan.target_default, + }, + "sources": [ + { + "label": s.label, + "key": s.key, + "files": len(s.scan.copy) if s.scan else 0, + "bytes": s.scan.bytes if s.scan else 0, + "too_long": len(s.too_long), + "links": s.scan.links if s.scan else 0, + "orphans": len(s.scan.orphans) if s.scan else 0, + "left_in_place": len(s.scan.left) if s.scan else 0, + "available": s.available, + } + for s in sources + ], + "totals": { + "sources": len(sources), + "files": copies, + "bytes": plan.needs_bytes, + "orphans": orphans, + "temps": temps, + "links": links, + "left_in_place": len(left), + "too_long": len(too_long), + "unlisted_sources": len(plan.unlisted), + "unlisted_files": unlisted_count, + }, + "too_long_listed": too_long[:LISTED], + "left_listed": left[:LISTED], + "unlisted": [ + {"key": key, "files": len(scan.copy) + len(scan.orphans)} + for key, _store, scan in plan.unlisted + ], + "needs_bytes": plan.needs_bytes, + "free_bytes": free, + "confirm_required": plan.confirm_required, + "token": plan.token, + "sentence": SENTENCE, + } + if insufficient: + payload["insufficient"] = insufficient + return payload diff --git a/py/api/thumbnail_move_recovery.py b/py/api/thumbnail_move_recovery.py new file mode 100644 index 00000000..347557d9 --- /dev/null +++ b/py/api/thumbnail_move_recovery.py @@ -0,0 +1,246 @@ +"""Move thumbnails after a restart and the unavailable folder's actions +(#320 Parts 3 and 4): undo an interrupted copy, finish a removal that +waited, Recreate folder and Use this folder.""" + +import os +import threading +from pathlib import Path + +from aiohttp import web + +from . import preview_location, preview_store +from . import thumbnail_move_files as files +from .job_routes import busy_response +from .jobs import THUMBNAILS_JOB_KIND, JobBusy +from .thumbnail_move_plan import ( + FINISHING_TITLE, + MOVE_GROUPS, + _read_json, + _refusal, + _remove, +) +from . import thumbnail_move_plan as move_plan + + +class ThumbnailMoveRecoveryMixin: + """Startup recovery, the finishing job, Recreate and Use this folder.""" + + _move_recover_lock = threading.Lock() + # Per API: set once recover_thumbnail_move has run (a move waits for it). + _move_recovered = False + + # ── startup ─────────────────────────────────────────────────────── + + def recover_thumbnail_move(self): + """Startup (B4): undo an interrupted copy, or forget a journal the + switch already made true. Idempotent; the move start waits for it.""" + with self._move_recover_lock: + if self._move_recovered: + return + try: + self._recover_journal() + except Exception: + self.logger.warning( + "The interrupted thumbnail move check failed", exc_info=True + ) + self._move_recovered = True + + def _recover_journal(self): + path = move_plan.journal_path() + journal = _read_json(path) + if journal is None: + return + setting = preview_store.setting() + target = journal.get("to") or {} + now = setting.path if setting.mode == preview_store.FOLDER_MODE else None + if setting.mode == target.get("mode") and ( + now is None + or os.path.normcase(now) == os.path.normcase(target.get("path") or "") + ): + _remove(path) # the switch happened; only the journal was left + return + for store in journal.get("stores") or []: + if not isinstance(store, str) or not os.path.isdir(store): + continue + for current, dirs, names in os.walk(store): + dirs[:] = [ + d + for d in dirs + if not preview_location.is_link_like(Path(current) / d) + ] + for name in names: + if files.MOVE_TEMP.search(name): + files.remove_file(Path(current) / name) + for folder in reversed(journal.get("created_dirs") or []): + if not isinstance(folder, str) or not os.path.isdir(folder): + continue + for current, dirs, names in os.walk(folder): + dirs[:] = [ + d + for d in dirs + if not preview_location.is_link_like(Path(current) / d) + ] + for name in names: + if files._preview_parts(name) or files.is_temp_name(name): + files.remove_file(Path(current) / name) + files.remove_empty_folders(folder) + folder = target.get("path") + if journal.get("created_folder") and folder and os.path.isdir(folder): + names = set(os.listdir(folder)) + if names <= {preview_location.MARKER_NAME}: + _remove(preview_location.marker_path(folder)) + files.remove_empty_folders(folder) + _remove(path) + self.logger.info("An interrupted thumbnail move was undone; nothing changed") + + def _cleanup_reachable(self, entry): + folder = entry.get("folder") + if folder: + return os.path.isdir(folder) + return True + + async def finish_thumbnail_move(self, _app=None): + """Startup: recover, then start "Move thumbnails (finishing)" when old + copies wait and every one is reachable (on the server's loop).""" + await self._run_in_executor(self.recover_thumbnail_move) + cleanup = list(preview_store.setting().cleanup) + if not cleanup: + return None + reachable = await self._run_in_executor( + lambda: all(self._cleanup_reachable(entry) for entry in cleanup) + ) + if not reachable: + return None + try: + return self._jobs.start( + THUMBNAILS_JOB_KIND, + FINISHING_TITLE, + admission="guarded", + cancellable=False, + scope={"mode": "move", "phase": "remove"}, + groups=MOVE_GROUPS, + runner=self._run_finishing, + ) + except JobBusy: + return None + + async def _run_finishing(self, job): + await self._remove_old_copies(job, None) + left = len(preview_store.setting().cleanup) + self._end_move( + job, + "finished", + ( + "Old thumbnails removed." + if not left + else ( + "Some old thumbnails are still in use; they are removed at " + "the next start." + ) + ), + ) + + # ── Recreate and Use this folder (Part 4) ───────────────────────── + + async def recreate_thumbnail_folder(self, _request): + blocker = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded") + if blocker is not None: + return busy_response(JobBusy(blocker)) + code = await self._run_in_executor(self._recreate_folder) + if code is not None: + status = ( + 409 if code in ("location-has-marker", "location-not-empty") else 400 + ) + if code == "location-has-marker": + return web.json_response( + { + "success": False, + "code": code, + "error": "The folder is back with its thumbnails. Nothing " + "was recreated.", + }, + status=status, + ) + return _refusal(code, status=status) + return web.json_response( + { + "success": True, + "location": self._thumbnail_location_payload(with_path=True), + } + ) + + def _recreate_folder(self): + setting = preview_store.setting() + if setting.mode != preview_store.FOLDER_MODE or not setting.path: + return "location-same" + folder = setting.path + if os.path.isdir(folder): + if preview_location.read_marker(folder) is not None: + return "location-has-marker" + if not preview_location.holds_only_metadata(folder): + return "location-not-empty" + elif not os.path.isdir(os.path.dirname(folder)): + return "location-parent-missing" + else: + try: + os.mkdir(folder) + except OSError: + return "location-not-writable" + try: + marker_id = preview_location.init_folder(folder, created_by_pm=True) + except OSError: + return "location-not-writable" + return self._commit_location_id(marker_id) + + async def adopt_thumbnail_folder(self, _request): + blocker = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded") + if blocker is not None: + return busy_response(JobBusy(blocker)) + code = await self._run_in_executor(self._adopt_folder) + if code is not None: + return _refusal(code) + return web.json_response( + { + "success": True, + "location": self._thumbnail_location_payload(with_path=True), + } + ) + + def _adopt_folder(self): + setting = preview_store.setting() + if setting.mode != preview_store.FOLDER_MODE or not setting.path: + return "location-same" + try: + marker = preview_location.read_marker(setting.path) + except OSError: + return "location-unavailable" + if marker is None: + return "location-unavailable" + return self._commit_location_id(marker["id"]) + + def _commit_location_id(self, marker_id): + from ..config import PromptManagerConfig + from ..config_store import get_config_store + + value = dict(preview_store.setting().to_config()) + value["id"] = marker_id + + def mutate(config): + section = config.get("thumbnails") + if not isinstance(section, dict): + section = {} + section["location"] = value + config["thumbnails"] = section + return config + + def publish(_config): + PromptManagerConfig.THUMBNAIL_LOCATION = value + preview_store.configure(value, check=False) + + try: + get_config_store().commit(mutate, expected_revision=None, publish=publish) + except Exception: + self.logger.exception("The thumbnail folder could not be saved") + return "location-not-writable" + preview_store.refresh() + return None diff --git a/py/api/thumbnail_prune.py b/py/api/thumbnail_prune.py new file mode 100644 index 00000000..0ffad194 --- /dev/null +++ b/py/api/thumbnail_prune.py @@ -0,0 +1,175 @@ +"""Remove thumbnails of removed sources (#320 Part 5, D200): a Thumbnail +maintenance scope, offered only in folder mode (31A). + +*Removed* means a key of this install in the thumbnail folder's marker whose +identity matches no root in any root list. The job removes the +preview-grammar files and temps of those key folders, ``rmdir``s the empty +folders, drops the keys from the marker and lists anything else as "Left in +place". Kind ``thumbnails``, ``mode: "prune"``. +""" + +import functools +import os +from pathlib import Path + +from aiohttp import web + +from . import preview_location, preview_store +from . import thumbnail_move_files as files +from .jobs import THUMBNAILS_JOB_KIND +from .thumbnail_location import location_refusal +from .thumbnail_move_plan import _number, _plural, _size + +PRUNE_TITLE = "Remove thumbnails of removed sources" +PRUNE_GROUPS = ("removed", "left", "errors") + + +def _prune_line(count, files_count, size): + if not count: + return "No sources were removed: nothing to remove." + return ( + f"{_plural(count, 'source')} no longer listed: " + f"{_plural(files_count, 'thumbnail')} ({_size(size)}) are removed." + ) + + +class ThumbnailPruneMixin: + """The prune plan and job.""" + + def _prune_targets(self): + """``[(key, store, scan)]`` of this install's unlisted keys (blocking).""" + setting = preview_store.setting() + location = preview_store.current() + if setting.mode != preview_store.FOLDER_MODE or not location.available: + return [] + marker = preview_location.read_marker(setting.path) or {} + install = setting.install + return self._unlisted_keys(setting, location, marker.get("keys") or {}, install) + + def _prune_payload(self, targets): + files_count = sum( + len(scan.copy) + len(scan.orphans) for _k, _s, scan in targets + ) + size = sum(scan.bytes for _k, _s, scan in targets) + return { + "success": True, + "mode": "prune", + "sources": [ + { + "key": key, + "files": len(scan.copy) + len(scan.orphans), + "bytes": scan.bytes, + "left_in_place": len(scan.left), + } + for key, _store, scan in targets + ], + "targets": files_count, + "bytes": size, + "line": _prune_line(len(targets), files_count, size), + "folder": preview_store.setting().path, + } + + async def get_prune_plan(self): + refused = location_refusal() + if refused is not None: + return refused + if preview_store.current().mode != preview_store.FOLDER_MODE: + return web.json_response( + { + "success": False, + "code": "invalid-request", + "error": "Thumbnails are kept inside each source: no removed " + "source has thumbnails here.", + }, + status=400, + ) + targets = await self._run_in_executor(self._prune_targets) + return web.json_response( + self._prune_payload(targets), headers={"Cache-Control": "no-store"} + ) + + def start_thumbnail_prune(self): + """Start the prune job (sync, on the loop); raises JobBusy.""" + return self._jobs.start( + THUMBNAILS_JOB_KIND, + PRUNE_TITLE, + admission="guarded", + cancellable=True, + scope={"mode": "prune"}, + groups=PRUNE_GROUPS, + runner=self._run_thumbnail_prune, + ) + + async def _run_thumbnail_prune(self, job): + targets = await self._run_in_executor(self._prune_targets) + job.update(total=len(targets), processed=0) + setting = preview_store.setting() + for done, (key, store, scan) in enumerate(targets, 1): + if job.cancelled: + break + removed, failed = await self._run_in_executor( + functools.partial(self._prune_store, store) + ) + job.record_counted( + "removed", removed, [{"path": key, "detail": "Thumbnails removed"}] + ) + if failed: + job.record_counted("errors", failed) + if scan.left: + job.record_counted( + "left", + len(scan.left), + [{"path": f"{key}/{'/'.join(item)}"} for item in scan.left], + ) + await self._run_in_executor( + functools.partial(self._forget_key, setting.path, key, store) + ) + job.update( + processed=done, + progress=int(done * 100 / len(targets)), + status_line=f"Removing thumbnails of {_number(done)} of " + f"{_number(len(targets))} removed sources", + ) + job.publish({"type": "progress", "summary": job.summary()}) + line = f"{_number(job.counters.get('removed', 0))} thumbnails removed" + if job.counters.get("errors"): + line += f" · {_plural(job.counters['errors'], 'error')}" + job.update(status_line=line, progress=100) + job.log_line(line) + job.finish("cancelled" if job.cancelled else "finished") + + @staticmethod + def _prune_store(store): + """Remove the previews and temps of one key folder; ``(removed, failed)``.""" + removed = failed = 0 + store = Path(store) + if not store.is_dir() or preview_location.is_link_like(store): + return 0, 0 + for current, dirs, names in os.walk(store): + dirs[:] = [ + d for d in dirs if not preview_location.is_link_like(Path(current) / d) + ] + for name in names: + path = Path(current) / name + if not (files.is_temp_name(name) or files._preview_parts(name)): + continue + if preview_location.is_link_like(path): + continue + if files.remove_file(path): + removed += 1 + else: + failed += 1 + files.remove_empty_folders(store) + return removed, failed + + @staticmethod + def _forget_key(folder, key, store): + """Drop *key* from the marker once its folder is gone.""" + if os.path.exists(store): + return + + def forget(marker): + marker["keys"].pop(key, None) + + with preview_store.write_lock: + preview_location.write_marker(folder, forget) diff --git a/tests/fixtures/legacy_contract_snapshot.json b/tests/fixtures/legacy_contract_snapshot.json index 1b061454..991652d8 100644 --- a/tests/fixtures/legacy_contract_snapshot.json +++ b/tests/fixtures/legacy_contract_snapshot.json @@ -125,6 +125,9 @@ "POST /prompt_manager/images/removed/return", "POST /prompt_manager/images/rescope", "POST /prompt_manager/images/reveal", + "POST /prompt_manager/images/thumbnails/location/adopt", + "POST /prompt_manager/images/thumbnails/location/recreate", + "POST /prompt_manager/images/thumbnails/plan", "POST /prompt_manager/jobs/autotag", "POST /prompt_manager/jobs/duplicates", "POST /prompt_manager/jobs/image-tags", @@ -215,6 +218,7 @@ "thumbnail_encoders.webp", "thumbnail_location", "thumbnail_location.available", + "thumbnail_location.cleanup_pending", "thumbnail_location.default_path", "thumbnail_location.mode", "thumbnail_location.path", diff --git a/tests/test_thumbnail_move.py b/tests/test_thumbnail_move.py new file mode 100644 index 00000000..9390a11c --- /dev/null +++ b/tests/test_thumbnail_move.py @@ -0,0 +1,414 @@ +"""Move thumbnails (#320 PR 3): the reviewed job in every direction, its +failure, cancel and restart paths, Recreate and the removed-sources scope. + +Each test runs against a temp config.json (``AdminAPITestCase``) and a temp +journal; jobs are awaited through their task, never a timer. +""" + +import json +import os +import re +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +try: + from tests.test_api_admin import AdminAPITestCase + from tests.test_api_images import make_png +except ImportError: # discovered with tests/ as the top-level directory + from test_api_admin import AdminAPITestCase + from test_api_images import make_png + +from py.api import preview_location, preview_store # noqa: E402 +from py.api import thumbnail_move_files as files # noqa: E402 +from py.api import thumbnail_move_plan as move_plan # noqa: E402 +from py.config import PromptManagerConfig # noqa: E402 + +REPO = Path(__file__).resolve().parent.parent +PLAN = "/prompt_manager/images/thumbnails/plan" +START = "/prompt_manager/jobs/thumbnails" + + +class MoveCase(AdminAPITestCase): + async def asyncSetUp(self): + await super().asyncSetUp() + self.api._get_all_output_dirs = lambda: [self.output_dir] + self.api.recover_thumbnail_move() + self.journal = Path(self.tmpdir) / "data" / "thumbnail-move.json" + for target in (move_plan,): + patcher = patch.object(target, "journal_path", lambda: str(self.journal)) + patcher.start() + self.addCleanup(patcher.stop) + default = patch( + "py.api.thumbnail_move_plan.default_path", + lambda: str(Path(self.tmpdir) / "default-thumbs"), + ) + default.start() + self.addCleanup(default.stop) + self.addCleanup(preview_store.configure, None) + self.addCleanup(setattr, PromptManagerConfig, "THUMBNAIL_LOCATION", None) + self.old_store = self.output_dir / "thumbnails" + self.folder = Path(self.tmpdir) / "thumbs" + # Two media files with previews, an orphan, a temp and a note. + make_png(self.output_dir / "a.png", size=(8, 8)) + make_png(self.output_dir / "sub" / "b.png", size=(8, 8)) + make_png(self.old_store / "a_thumb.webp", size=(4, 4)) + make_png(self.old_store / "sub" / "b_thumb.webp", size=(4, 4)) + make_png(self.old_store / "gone_thumb.webp", size=(4, 4)) + (self.old_store / ".a_thumb.part-1-1.webp").write_bytes(b"t") + (self.old_store / "notes.txt").write_text("mine", encoding="utf-8") + + async def plan(self, location): + resp = await self.client.post(PLAN, json={"mode": "move", "location": location}) + return resp.status, await resp.json() + + async def move(self, location, confirmed=True): + status, plan = await self.plan(location) + self.assertEqual(status, 200, plan) + resp = await self.client.post( + START, json={"mode": "move", "token": plan["token"], "confirmed": confirmed} + ) + body = await resp.json() + self.assertEqual(resp.status, 202, body) + job = self.api._jobs.get(body["job"]["id"]) + await job.task + return plan, job + + def config(self): + with open(self.config_path, encoding="utf-8") as handle: + return json.load(handle) + + def key_files(self, folder=None): + folder = folder or self.folder + return sorted( + p.relative_to(folder).as_posix() + for p in folder.rglob("*") + if p.is_file() and p.name != preview_location.MARKER_NAME + ) + + +class TestMoveToAFolder(MoveCase): + async def test_review_counts_what_moves_is_removed_and_stays(self): + status, plan = await self.plan({"mode": "folder", "path": str(self.folder)}) + self.assertEqual(status, 200, plan) + totals = plan["totals"] + self.assertEqual( + ( + totals["files"], + totals["orphans"], + totals["temps"], + totals["left_in_place"], + ), + (2, 1, 1, 1), + ) + self.assertTrue(plan["confirm_required"]) + self.assertTrue(plan["to"]["will_create"]) + self.assertFalse(self.folder.exists()) # the review writes nothing + + async def test_a_move_copies_switches_and_removes_old_copies(self): + before = (self.output_dir / "a.png").read_bytes() + _plan, job = await self.move({"mode": "folder", "path": str(self.folder)}) + self.assertEqual(job.state, "finished", job.log) + self.assertEqual(job.counters["moved"], 2) + keys = self.key_files() + self.assertEqual(len(keys), 2) + self.assertTrue(all(re.match(r"^output-[0-9a-f]{8}/", k) for k in keys)) + location = self.config()["thumbnails"]["location"] + self.assertEqual( + (location["mode"], location["path"]), ("folder", str(self.folder)) + ) + self.assertEqual(location["cleanup"], []) + self.assertEqual(preview_store.current().mode, "folder") + # Old copies, the orphan and the temp went; the note stays, so its + # folder stays; originals are untouched. + self.assertEqual( + sorted(p.name for p in self.old_store.rglob("*")), ["notes.txt"] + ) + self.assertEqual((self.output_dir / "a.png").read_bytes(), before) + self.assertFalse(self.journal.exists()) + copied = next(self.folder.rglob("a_thumb.webp")) + self.assertEqual(copied.stat().st_mtime_ns % files.MTIME_STEP_NS, 0) + + async def test_the_move_back_removes_the_folder_it_made(self): + await self.move({"mode": "folder", "path": str(self.folder)}) + _plan, job = await self.move({"mode": "source"}) + self.assertEqual(job.state, "finished", job.log) + self.assertTrue((self.old_store / "a_thumb.webp").is_file()) + self.assertTrue((self.old_store / "sub" / "b_thumb.webp").is_file()) + self.assertFalse(self.folder.exists()) + self.assertEqual(self.config()["thumbnails"]["location"]["mode"], "source") + + async def test_folder_to_folder_keeps_the_key_names(self): + await self.move({"mode": "folder", "path": str(self.folder)}) + first = self.key_files() + other = Path(self.tmpdir) / "thumbs2" + _plan, job = await self.move({"mode": "folder", "path": str(other)}) + self.assertEqual(job.state, "finished", job.log) + self.assertEqual(self.key_files(other), first) + self.assertFalse(self.folder.exists()) + + async def test_previews_of_sources_no_longer_listed_go_when_leaving(self): + await self.move({"mode": "folder", "path": str(self.folder)}) + setting = preview_store.setting() + + def add(marker): + marker["keys"]["old-0123abcd"] = { + "path": "/gone", + "identity": "/gone", + "install": setting.install, + } + + preview_location.write_marker(str(self.folder), add) + make_png(self.folder / "old-0123abcd" / "x_thumb.webp", size=(4, 4)) + preview_store.refresh() + status, plan = await self.plan({"mode": "source"}) + self.assertEqual(plan["totals"]["unlisted_sources"], 1) + _plan, job = await self.move({"mode": "source"}) + self.assertFalse(self.folder.exists()) + + async def test_confirmation_and_a_stale_token(self): + status, plan = await self.plan({"mode": "folder", "path": str(self.folder)}) + resp = await self.client.post( + START, json={"mode": "move", "token": plan["token"]} + ) + self.assertEqual((await resp.json())["code"], "confirmation-required") + resp = await self.client.post( + START, json={"mode": "move", "token": "nope", "confirmed": True} + ) + self.assertEqual(resp.status, 409) + self.assertEqual((await resp.json())["code"], "plan-stale") + + async def test_a_folder_that_is_too_small_is_refused(self): + with patch.object(move_plan, "_free", lambda _path: 1): + status, body = await self.plan({"mode": "folder", "path": str(self.folder)}) + self.assertEqual(status, 507) + self.assertEqual(body["code"], "insufficient-space") + + async def test_validation_refusals_come_back_in_place(self): + status, body = await self.plan( + {"mode": "folder", "path": str(self.output_dir / "thumbs")} + ) + self.assertEqual((status, body["code"]), (400, "location-served")) + status, body = await self.plan({"mode": "source"}) + self.assertEqual(body["code"], "location-same") + + +class TestFailures(MoveCase): + async def test_a_per_file_error_still_switches_and_removes_its_old_copy(self): + real = files.copy_one + + def failing(source, destination, *args, **kwargs): + if Path(source).name == "a_thumb.webp": + raise files.CopyFailed("Could not be read; made again on view") + return real(source, destination, *args, **kwargs) + + with patch.object(files, "copy_one", failing): + _plan, job = await self.move({"mode": "folder", "path": str(self.folder)}) + self.assertEqual(job.state, "finished") + self.assertEqual((job.counters["moved"], job.counters["errors"]), (1, 1)) + self.assertFalse((self.old_store / "a_thumb.webp").exists()) + + async def test_a_folder_wide_error_changes_nothing(self): + def gone(*_args, **_kwargs): + raise files.FolderGone("disk full") + + with patch.object(files, "copy_one", gone): + _plan, job = await self.move({"mode": "folder", "path": str(self.folder)}) + self.assertEqual(job.state, "failed") + self.assertIn("nothing changed", job.status_line) + self.assertFalse(self.folder.exists()) + self.assertTrue((self.old_store / "a_thumb.webp").is_file()) + self.assertEqual(preview_store.current().mode, "source") + self.assertFalse(self.journal.exists()) + + async def test_cancel_before_the_switch_changes_nothing(self): + status, plan = await self.plan({"mode": "folder", "path": str(self.folder)}) + real = self.api._move_copy_one + + def cancel_first(pair, **kwargs): + job = self.api._jobs.active("thumbnails") + job.cancelled = True + return real(pair, **kwargs) + + with patch.object(self.api, "_move_copy_one", cancel_first): + resp = await self.client.post( + START, json={"mode": "move", "token": plan["token"], "confirmed": True} + ) + job = self.api._jobs.get((await resp.json())["job"]["id"]) + await job.task + self.assertEqual(job.state, "cancelled") + self.assertFalse(self.folder.exists()) + self.assertEqual(preview_store.current().mode, "source") + + async def test_an_exit_during_the_copy_leaves_the_journal_for_the_next_start(self): + status, plan = await self.plan({"mode": "folder", "path": str(self.folder)}) + real = self.api._move_copy_one + exiting = {"now": False} + + def exit_after(pair, **kwargs): + result = real(pair, **kwargs) + exiting["now"] = True # the interpreter starts exiting + return result + + from py.api import thumbnail_move + + with ( + patch.object(self.api, "_move_copy_one", exit_after), + patch.object( + thumbnail_move.exit_signal, "requested", lambda: exiting["now"] + ), + ): + resp = await self.client.post( + START, json={"mode": "move", "token": plan["token"], "confirmed": True} + ) + job = self.api._jobs.get((await resp.json())["job"]["id"]) + await job.task + self.assertTrue(self.journal.exists()) + self.assertTrue(self.key_files()) # partial copies were left + # The next start undoes it: copies and the created folder go. + self.api._move_recovered = False + self.api.recover_thumbnail_move() + self.assertFalse(self.journal.exists()) + self.assertFalse(self.folder.exists()) + self.assertTrue((self.old_store / "a_thumb.webp").is_file()) + + +class TestRestart(MoveCase): + async def test_a_journal_left_after_the_switch_is_forgotten(self): + await self.move({"mode": "folder", "path": str(self.folder)}) + self.journal.parent.mkdir(parents=True, exist_ok=True) + self.journal.write_text( + json.dumps( + {"state": "copying", "to": {"mode": "folder", "path": str(self.folder)}} + ), + encoding="utf-8", + ) + self.api._move_recovered = False + self.api.recover_thumbnail_move() + self.assertFalse(self.journal.exists()) + self.assertTrue(self.key_files()) + + async def test_old_copies_that_waited_are_removed_by_the_finishing_job(self): + with patch.object(files, "remove_file", lambda path, **_k: False): + _plan, job = await self.move({"mode": "folder", "path": str(self.folder)}) + self.assertTrue(preview_store.setting().cleanup) + self.assertTrue((self.old_store / "a_thumb.webp").exists()) + job = await self.api.finish_thumbnail_move() + await job.task + self.assertEqual(job.title, "Move thumbnails (finishing)") + self.assertFalse((self.old_store / "a_thumb.webp").exists()) + self.assertEqual(self.config()["thumbnails"]["location"]["cleanup"], []) + + +class TestAdmission(MoveCase): + async def test_automatic_previews_and_setting_changes_wait_for_the_move(self): + status, plan = await self.plan({"mode": "folder", "path": str(self.folder)}) + seen = {} + real = self.api._move_copy_one + + def observe(pair, **kwargs): + seen["blocker"] = self.api._auto_thumbnail_blocker() + return real(pair, **kwargs) + + with patch.object(self.api, "_move_copy_one", observe): + resp = await self.client.post( + START, json={"mode": "move", "token": plan["token"], "confirmed": True} + ) + job = self.api._jobs.get((await resp.json())["job"]["id"]) + refused = self.api._thumbnail_change_refused({"THUMBNAIL_QUALITY": 5}) + self.assertEqual(refused.status, 409) + await job.task + self.assertIs(seen["blocker"], job) + + def test_the_move_never_trims_memory_and_copies_on_the_job_pool(self): + text = "".join( + (REPO / "py" / "api" / name).read_text(encoding="utf-8") + for name in ("thumbnail_move.py", "thumbnail_prune.py") + ) + self.assertNotIn("_release_job_memory", text) + self.assertIn("map_parallel", (REPO / "py/api/thumbnail_move.py").read_text()) + + +class TestPrune(MoveCase): + async def test_only_this_installs_removed_sources_go(self): + await self.move({"mode": "folder", "path": str(self.folder)}) + install = preview_store.setting().install + + def add(marker): + marker["keys"]["old-0123abcd"] = {"identity": "/gone", "install": install} + marker["keys"]["theirs-0123abcd"] = {"identity": "/x", "install": "other"} + + preview_location.write_marker(str(self.folder), add) + make_png(self.folder / "old-0123abcd" / "x_thumb.webp", size=(4, 4)) + make_png(self.folder / "theirs-0123abcd" / "y_thumb.webp", size=(4, 4)) + preview_store.refresh() + resp = await self.client.get(PLAN + "?mode=prune") + body = await resp.json() + self.assertEqual([s["key"] for s in body["sources"]], ["old-0123abcd"]) + self.assertIn("1 source no longer listed", body["line"]) + resp = await self.client.post(START, json={"mode": "prune"}) + job = self.api._jobs.get((await resp.json())["job"]["id"]) + await job.task + self.assertFalse((self.folder / "old-0123abcd").exists()) + self.assertTrue((self.folder / "theirs-0123abcd" / "y_thumb.webp").exists()) + keys = preview_location.read_marker(str(self.folder))["keys"] + self.assertNotIn("old-0123abcd", keys) + + +class TestRecreate(MoveCase): + async def test_recreate_and_its_refusals(self): + await self.move({"mode": "folder", "path": str(self.folder)}) + url = "/prompt_manager/images/thumbnails/location/recreate" + resp = await self.client.post(url) + self.assertEqual((await resp.json())["code"], "location-has-marker") + for child in sorted(self.folder.rglob("*"), reverse=True): + child.rmdir() if child.is_dir() else child.unlink() + self.folder.rmdir() + self.assertFalse(preview_store.refresh()) + resp = await self.client.post(url) + self.assertEqual(resp.status, 200, await resp.text()) + self.assertTrue(preview_store.current().available) + self.assertNotEqual(self.config()["thumbnails"]["location"]["id"], None) + + async def test_use_this_folder_adopts_another_marker(self): + await self.move({"mode": "folder", "path": str(self.folder)}) + marker = preview_location.read_marker(str(self.folder)) + marker["id"] = "someone-else" + (self.folder / preview_location.MARKER_NAME).write_text( + json.dumps(marker), encoding="utf-8" + ) + self.assertFalse(preview_store.refresh()) + resp = await self.client.post( + "/prompt_manager/images/thumbnails/location/adopt" + ) + self.assertEqual(resp.status, 200, await resp.text()) + self.assertTrue(preview_store.current().available) + self.assertEqual(self.config()["thumbnails"]["location"]["id"], "someone-else") + + +class TestFiles(unittest.TestCase): + def test_mtime_rounds_up_to_two_seconds(self): + self.assertEqual(files.rounded_mtime_ns(0), 0) + self.assertEqual(files.rounded_mtime_ns(1), files.MTIME_STEP_NS) + self.assertEqual( + files.rounded_mtime_ns(files.MTIME_STEP_NS), files.MTIME_STEP_NS + ) + + def test_long_windows_destinations_are_per_file_errors(self): + self.assertTrue(files.too_long("D:\\" + "x" * 240, long_paths=False)) + self.assertFalse(files.too_long("D:\\x", long_paths=False)) + + def test_a_held_open_file_is_retried_then_left(self): + waits = [] + error = OSError("in use") + error.winerror = 32 + with patch.object(files.os, "unlink", side_effect=error): + self.assertFalse(files.remove_file("x", sleep=waits.append)) + self.assertEqual(waits, [0.5, 1.0, 2.0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/trash_route_audit.py b/tests/trash_route_audit.py index f55c027f..c8fa208e 100644 --- a/tests/trash_route_audit.py +++ b/tests/trash_route_audit.py @@ -77,6 +77,7 @@ ROUTE_AUDIT = { "POST /prompt_manager/jobs/autotag", "POST /prompt_manager/jobs/thumbnails", "GET /prompt_manager/images/thumbnails/plan", + "POST /prompt_manager/images/thumbnails/plan", "GET /prompt_manager/images/thumbnails/status", "POST /prompt_manager/images/generate-thumbnails", "POST /prompt_manager/images/generate-thumbnails/progress", @@ -171,6 +172,8 @@ ROUTE_AUDIT = { "POST /prompt_manager/cleanup", "POST /prompt_manager/diagnostics/test-link", "POST /prompt_manager/images/clear-thumbnails", + "POST /prompt_manager/images/thumbnails/location/adopt", + "POST /prompt_manager/images/thumbnails/location/recreate", "POST /prompt_manager/jobs/model-download", "POST /prompt_manager/jobs/{job_id}/cancel", "POST /prompt_manager/library",