diff --git a/TESTING.md b/TESTING.md index 9501625..3d26c39 100644 --- a/TESTING.md +++ b/TESTING.md @@ -333,6 +333,75 @@ hold a walk open with a gated scan instead of sleeping. listing gets a 500. Rescan waits for a walk that started after the click (by walk number, not by clock), Rescans clicked during one walk share the next, and a failed walk is a 500 even when a previous listing exists. +- **New outputs (added without a walk, PR 2).** The image monitor reports + every settled output it accepts (by default `.png .jpg .jpeg .webp .gif`; + videos only when their extension is in `monitoring.extensions`), linked or + not, with its `realpath`, through `set_output_listener`. On its worker + thread the API's `notify_output_settled` only does + `loop.call_soon_threadsafe(output_settled, path)`, and does nothing before + the loop is known or after `cleanup_gallery_listings`. `output_settled` + matches the path against each walk's `resolved_root` (`normcase(realpath)`, + computed in the executor; `root_contains` ends the match at a separator + and ignores a Windows `\\?\` or `\\?\UNC\` prefix on either side). +- **The add.** A root with a stored listing and no walk running gets the + file added without a walk. `relative_part` slices the path below the + root in its own case (Windows compares folded, slices the original); + `_run_adds` places the pending files of one root in one executor batch + (`_patch_listing`): `_settled_entry` applies the walk's filters + (not a symlink or folder by `lstat`, `_name_suffix` in `MEDIA_EXTENSIONS`, + no `thumbnails` folder, at most `MAX_SCAN_DEPTH` folders, every folder on + the way listable by `os.access`, which on Windows ignores ACLs: a folder + whose ACL denies listing is added there and corrected by the next walk). Where names fold case the stored + spelling must match first, as the filters compare it: Windows through + `nt._getfinalpathname` (its `realpath` keeps the given case when it cannot + open the file), macOS through `F_GETPATH` on a non-blocking open (a FIFO + never holds the executor); a case-insensitive mount on Linux is not + checked and is corrected by the next walk. `insert_entry` builds a new list with + the `(Path, mtime)` entry where the walk's newest-first sort puts it, + replacing the same path's entry. `add_folder` copies both folder sets. The + batch is stored only when no walk started and no drop happened while it + ran (a walk started after the report lists the file itself); `walked_at` + is untouched, so the 30 s refresh still walks and corrects any drift, and + `walk_root` keeps the patched list object when that walk finds the same. + Tests compare every added listing with a real walk (`_walked`). +- **When it walks instead.** The root is marked dirty (`mark_root_dirty`, + `_mark_dirty`) when the add cannot be exact: a walk is running (it may have + passed the folder: the trailing walk), the path cannot be sliced, the + `lstat` or the spelling read fails, the stored spelling differs, another + file has the same mtime (a walk orders those by its + visit order), the listing is at `MAX_GALLERY_FILES`, or the batch raised. + A file the walk's filters exclude is neither added nor walked. A root + without a listing, or a dropped one, is walked by its next read. A dirty + root gets one background walk, no sooner than `rewalk_interval` = + `min(15, max(5, 5 × last walk's duration))` seconds after the previous walk + started. `_schedule_walk` never arms a timer while a walk runs (the walk's + end re-checks `dirty`: the trailing walk), arms at most one per root and + never sooner than 0.5 s, and keeps the earlier deadline when the stale + refresh and a dirty root both want one; every walk start (read, Rescan, + warm-up or timer) cancels it and forgets the pending adds. A failed walk + clears `dirty`; a cancelled walk starts no trailing walk. When the output + roots change (`reset_gallery_roots_cache`) every record is `retired`: its + timer is cancelled and no add, mark, timer or trailing walk touches it + until a reader or the warm-up asks for that root again (the monitor keeps + watching a removed root until restart). Tests patch + `REWALK_FLOOR`/`REWALK_CAP` and run the pending timer's callback by hand + (`_fire`) instead of waiting for it, and hold an add batch in the executor + (`_hold_adds`) to race it with a walk, a drop or the cleanup. +- **Workspace caches after an add.** A patched listing shares every + unchanged entry object with the one before it, so `listing_delta` finds + the change by identity (a walk builds new tuples and is rebuilt in full, + as is any change of more than `MAX_DELTA` entries). The per-file values + build only the added files (`_workspace_listed_step`), the newest order + places them by bisection (`_newest_step`, only when the per-file cache + still holds exactly the values this call built: a page read and the warm + build run in executor threads at once, and an order labelled with older + values than it holds would add a file twice), the favorites lookup is copied + and extended (`_favorite_listing_step`, rebuilt when a key collides), and + the other orders, wanted paths and locators rebuild lazily. On a synthetic + 20,000-file listing a one-file add costs about 3 ms in the executor and + 15 ms for the warm build, against about 155 ms + 80 ms for a walk and a + full build (`test_a_warm_build_after_an_add_builds_only_the_added_values` + counts the values built and the sort keys read). - **Measuring it.** `tools/vendor/perf/images_load_probe.mjs` is a maintainer tool, never run in CI. After `npm ci --prefix tools/vendor` and `npx --prefix tools/vendor playwright install chromium`, run diff --git a/__init__.py b/__init__.py index 0e7d294..8ed1def 100644 --- a/__init__.py +++ b/__init__.py @@ -149,6 +149,9 @@ try: # Every settled output's preview is queued for the API's background # worker (#249): the workspace never asks to generate thumbnails. _global_image_monitor.set_thumbnail_queue(_review_api.queue_capture_thumbnail) + # Every settled output walks its Images listing again (#348): the + # API hops from the monitor's thread onto the server's loop. + _global_image_monitor.set_output_listener(_review_api.notify_output_settled) _review_host = getattr(config.server_instance, "app", None) if _review_host is not None and hasattr(_review_host, "on_cleanup"): @@ -157,6 +160,7 @@ try: ): monitor.set_review_recorder(None) monitor.set_thumbnail_queue(None) + monitor.set_output_listener(None) api.close_auto_thumbnails(final=True) _review_host.on_cleanup.insert(0, _stop_review_recording) diff --git a/py/api/__init__.py b/py/api/__init__.py index 35efee4..12619d5 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -362,8 +362,11 @@ class PromptManagerAPI( # background walks refreshing them (py/api/gallery_listing.py, #348). self._root_listings: dict[str, RootListing] = {} self._listing_tasks: set = set() - # The loop the walks run on, set by the first walk. + # The loop the walks run on, set by the startup warm-up or the first + # walk; the monitor's thread hops onto it (notify_output_settled). self._listing_loop = None + # Set by cleanup_gallery_listings: no background walk starts after it. + self._listings_closed = False # The workspace caches rebuilt behind a stored walk that changed the # listing, once the host's startup warm-up has run (#348). self._workspace_warm_enabled = False @@ -492,7 +495,7 @@ class PromptManagerAPI( """The shared listing record of one output root (created empty).""" return self._root_listings.setdefault(str(root), RootListing()) - def invalidate_gallery_cache(self): + def invalidate_gallery_cache(self, retire=False): """Drop every root's listing, so the next read waits for a new walk. Called after files are deleted through the app (duplicate cleanup and @@ -503,6 +506,9 @@ class PromptManagerAPI( _workspace_listed and _card_facts); its sort orders are dropped, as the walk replaces them anyway, and so is the favorites lookup, which holds the old list. + + ``retire`` (the roots changed): no background walk of a root starts + again until a reader or the warm-up asks for it. """ loop = self._listing_loop if loop is not None and loop.is_running() and _running_loop() is not loop: @@ -513,7 +519,7 @@ class PromptManagerAPI( def drop(): try: - self._drop_listings() + self._drop_listings(retire) finally: dropped.set() @@ -521,11 +527,19 @@ class PromptManagerAPI( if not dropped.wait(_OFF_LOOP_DROP_WAIT): self.logger.warning("The Images listing drop is still queued") return - self._drop_listings() + self._drop_listings(retire) - def _drop_listings(self): + def _drop_listings(self, retire=False): for record in list(self._root_listings.values()): record.drop() + if retire: + # The roots changed: a removed root must not come back through + # a pending timer, a trailing walk or a monitor mark. + record.retired = True + record.dirty = False + handle, record.rewalk_handle = record.rewalk_handle, None + if handle is not None: + handle.cancel() self._workspace_order_cache = {} self._favorite_listing_cache = {} @@ -539,15 +553,36 @@ class PromptManagerAPI( # Each stored walk that changes a listing then rebuilds the workspace's # per-file values, orders and favorites behind it (_warm_workspace). self._workspace_warm_enabled = True + self._listing_loop = asyncio.get_running_loop() + self._listings_closed = False try: for root in self._get_all_output_dirs(): + self._root_listing(root).retired = False self._schedule_walk(root) except Exception as e: self.logger.warning(f"Could not warm the Images listing: {e}") + def notify_output_settled(self, path): + """The image monitor settled an output at *path* (any thread). + + Called on the monitor's worker thread with the resolved path. It only + hands the path to the event loop (``output_settled``) and never + touches a listing from this thread; before the loop is known, or once + the listings are cleaned up, it does nothing. + """ + loop = self._listing_loop + if loop is None or self._listings_closed or not loop.is_running(): + return + try: + loop.call_soon_threadsafe(self.output_settled, path) + except RuntimeError: + pass # the loop closed meanwhile (shutdown) + async def cleanup_gallery_listings(self, app=None): """Stop the background listing walks (aiohttp on_cleanup).""" self._workspace_warm_enabled = False + # No mark, deferred walk or trailing walk starts a walk from now on. + self._listings_closed = True # Before any await, so no delayed refresh can start a walk meanwhile. for record in self._root_listings.values(): handle, record.rewalk_handle = record.rewalk_handle, None @@ -571,7 +606,7 @@ class PromptManagerAPI( def reset_gallery_roots_cache(self): """Drop every cached listing and derived workspace value (roots changed).""" - self.invalidate_gallery_cache() + self.invalidate_gallery_cache(retire=True) self._workspace_listed_cache = {} self._workspace_card_cache = {} diff --git a/py/api/gallery_listing.py b/py/api/gallery_listing.py index b873099..e6ac499 100644 --- a/py/api/gallery_listing.py +++ b/py/api/gallery_listing.py @@ -6,13 +6,20 @@ loop; the walk itself, the comparison with the previous listing and both folder sets run in the executor through ``walk_root``. Routes live in ``py/api/images.py`` (``_listing``); this module has no aiohttp import so tests can exercise it without a server. + +A settled output the image monitor reports is added to its root's stored +listing without a walk (``output_settled``, ``insert_entry``), in the place +a walk would list it. When that cannot be exact, the root is marked dirty +(``mark_root_dirty``) instead; a dirty root gets one background walk, no +sooner than ``rewalk_interval`` after its previous walk started (PR 2). """ from __future__ import annotations import asyncio import os -from dataclasses import dataclass +from bisect import bisect_left +from dataclasses import dataclass, field from pathlib import PurePath from typing import Callable, NamedTuple @@ -26,6 +33,19 @@ REFRESH_AFTER = 30 # (#348 live test: reads during the walk took 2.2 s instead of 0.07 s). REFRESH_DELAY = 1.0 +# No background walk of a root starts sooner than rewalk_interval(last walk's +# duration) after its previous walk started: FACTOR times that duration, at +# least FLOOR and at most CAP seconds. Walks up to CAP / FACTOR (3 s) then +# hold the GIL at most ~20% of the time; the cap keeps a new output listed +# within ~20 s, at the cost of a higher share (back to back from 15 s) on +# slower walks. Read through this module so tests can patch them. +REWALK_FLOOR = 5 +REWALK_FACTOR = 5 +REWALK_CAP = 15 + +# A deferred walk is never armed sooner than this: no zero-delay loop. +MIN_REWALK_DELAY = 0.5 + # A reader with no listing gives up (500) after this many walks in a row were # discarded by invalidations made while they ran. MAX_DISCARDED_WALKS = 3 @@ -80,14 +100,35 @@ class RootListing: flight: asyncio.Future | None = None # True while walks keep failing: one warning per streak. failing: bool = False - # The delayed refresh not started yet (loop.call_later), or None. + # The one deferred background walk not started yet (loop.call_later), or + # None: a stale refresh or a dirty root's rewalk, whichever is due first. rewalk_handle: asyncio.TimerHandle | None = None + # normcase(realpath(root)), computed by the walk in the executor: monitor + # paths are matched against it without file I/O on the loop. + resolved_root: str | None = None + # Monotonic start and duration of the last walk (rewalk_interval). + last_started_at: float | None = None + last_duration: float | None = None + # A settled output was reported under this root; cleared when a walk starts. + dirty: bool = False + # Dropped because the output roots changed: no background walk until a + # reader (or the warm-up) asks for this root again. The monitor keeps + # watching a removed root until ComfyUI restarts. + retired: bool = False + # Relative paths of settled outputs waiting to be added to ``files`` + # without a walk, and the one task adding them (``output_settled``). + pending_adds: list = field(default_factory=list) + add_task: asyncio.Task | None = None + # Settled outputs added to the stored listing since the last walk stored. + patched: int = 0 def drop(self) -> None: self.files = None self.folders = None self.folders_with_ancestors = None self.walked_at = None + self.pending_adds = [] + self.patched = 0 self.generation += 1 @@ -99,26 +140,174 @@ def folder_sets(root, files) -> tuple[frozenset, frozenset]: files, plus every ancestor joined by ``/``. """ # From the path strings: relative_to per file was half of a large walk. + sep, prefix = _root_prefix(root) + folders = set() + for path, _mtime in files: + rel_dir = _folder_of(root, path, sep, prefix) + if rel_dir: + folders.add(rel_dir) + with_ancestors = set(folders) + for folder in folders: + with_ancestors.update(_ancestors(folder)) + return frozenset(folders), frozenset(with_ancestors) + + +def add_folder(root, path, folders, with_ancestors) -> tuple[frozenset, frozenset]: + """Both folder sets once *path* is listed too (the same sets when it adds none).""" + sep, prefix = _root_prefix(root) + rel_dir = _folder_of(root, path, sep, prefix) + if not rel_dir or rel_dir in folders: + return folders, with_ancestors + return folders | {rel_dir}, with_ancestors | {rel_dir, *_ancestors(rel_dir)} + + +def _root_prefix(root) -> tuple[str, str]: # The separator is the path flavour's own (tests use PureWindowsPath). sep = str(type(root)("a", "b"))[1] prefix = str(root) if not prefix.endswith(sep): prefix += sep - folders = set() - for path, _mtime in files: - text = str(path) - if text.startswith(prefix): - rel_dir = text[len(prefix) :].rpartition(sep)[0] - else: - rel_dir = str(path.relative_to(root).parent) - if rel_dir and rel_dir != ".": - folders.add(rel_dir) - with_ancestors = set(folders) - for folder in folders: - parts = folder.replace("\\", "/").split("/") - for i in range(1, len(parts)): - with_ancestors.add("/".join(parts[:i])) - return frozenset(folders), frozenset(with_ancestors) + return sep, prefix + + +def _folder_of(root, path, sep, prefix) -> str: + """The folder ``/gallery/subfolders`` names for *path*; "" for the root.""" + text = str(path) + if text.startswith(prefix): + rel_dir = text[len(prefix) :].rpartition(sep)[0] + else: + rel_dir = str(path.relative_to(root).parent) + return "" if rel_dir == "." else rel_dir + + +def _ancestors(folder: str) -> list[str]: + parts = folder.replace("\\", "/").split("/") + return ["/".join(parts[:i]) for i in range(1, len(parts))] + + +def relative_part(resolved_root: str, path, pathmod=os.path) -> str | None: + """The part of the resolved *path* below *resolved_root*, as a walk spells it. + + Kept in *path*'s own case (``realpath`` gives a file's on-disk case on + Windows) with the native separator. None when *path* is not below the + root, or when the case folding changed the length so the slice would + not be exact. + """ + sep = pathmod.sep + text = str(path) + if sep == "\\": + text = _plain(text.replace("/", sep), pathmod) + folded = pathmod.normcase(text) + prefix = _plain(resolved_root, pathmod) + if not prefix.endswith(sep): + prefix += sep + if len(folded) != len(text) or not folded.startswith(prefix): + return None + relative = text[len(prefix) :] + if any(part in ("", ".", "..") for part in relative.split(sep)): + return None + return relative + + +def insert_entry(files: list, entry: tuple, max_files: int) -> list | None: + """A new listing with *entry* ``(path, mtime)`` where a walk would list it. + + *files* is newest first; an entry for the same path is replaced. Returns + *files* itself when it already holds *entry*, and None when the place is + not exact: another file has the same mtime (a walk orders those by its + own visit order), or the listing is at the walk's *max_files* cap. + Blocking (one pass over *files*): run it in the executor. + """ + path, mtime = entry + text = str(path) + kept = files + for index, (listed, listed_mtime) in enumerate(files): + if str(listed) == text: + if listed_mtime == mtime: + return files + kept = files[:index] + files[index + 1 :] + break + else: + if len(files) >= max_files: + return None + index = bisect_left(kept, -mtime, key=lambda item: -item[1]) + if index < len(kept) and kept[index][1] == mtime: + return None + return [*kept[:index], entry, *kept[index:]] + + +# A listing that differs from the cached one by more entries than this is +# rebuilt in full by the workspace caches (listing_delta). +MAX_DELTA = 256 + + +def listing_delta(old: list, new: list, limit: int = MAX_DELTA): + """``(added, removed)`` entries from *old* to *new*, or None past *limit*. + + A listing patched by ``insert_entry`` shares every unchanged entry object + with the one it came from, so entries are told apart by identity: a walk + builds new tuples, shares none and is rebuilt in full. *old* must still + be alive (the + caller's cache holds it), so no identity was reused. + """ + if abs(len(new) - len(old)) > limit: + return None + old_ids = set(map(id, old)) + added = [] + for entry in new: + if id(entry) not in old_ids: + added.append(entry) + if len(added) > limit: + return None + if new and len(added) == len(new): + return None # nothing shared: a walk + if len(old) + len(added) == len(new): + return added, [] + new_ids = set(map(id, new)) + return added, [entry for entry in old if id(entry) not in new_ids] + + +def rewalk_interval(last_duration) -> float: + """Seconds from a walk's start to the next background walk of its root.""" + scaled = REWALK_FACTOR * (last_duration or 0.0) + return min(REWALK_CAP, max(REWALK_FLOOR, scaled)) + + +def root_contains(resolved_root: str, path, pathmod=os.path) -> bool: + """True when the resolved *path* lies under *resolved_root* (no file I/O). + + *resolved_root* is already ``normcase``d; *path* is folded the same way, + so Windows matches case-insensitively and with either separator. The + match ends at a separator: ``/out2/x.png`` is not under ``/out``. + """ + target = _plain(pathmod.normcase(str(path)), pathmod) + prefix = _plain(resolved_root, pathmod) + if not prefix.endswith(pathmod.sep): + prefix += pathmod.sep + return target.startswith(prefix) + + +def _plain(path: str, pathmod) -> str: + """*path* (normcased or not) without a Windows extended-length prefix. + + ``realpath`` may keep ``\\\\?\\`` on a file past MAX_PATH while its shorter + root resolves without it. + """ + if pathmod.sep != "\\": + return path + if path[:8].lower() == "\\\\?\\unc\\": + return "\\\\" + path[8:] + if path.startswith("\\\\?\\"): + return path[4:] + return path + + +def resolve_root(root) -> str: + """``normcase(realpath(root))``, the spelling monitor paths are matched in. + + File I/O (realpath follows links): call it in the executor. + """ + return os.path.normcase(os.path.realpath(str(root))) def same_listing(old, new) -> bool: @@ -130,8 +319,10 @@ def same_listing(old, new) -> bool: def walk_root( scan_fn: Callable[[PurePath], list], root, previous_files -) -> tuple[list, frozenset, frozenset]: - """The single executor job of a walk: ``(files, folders, with_ancestors)``. +) -> tuple[list, frozenset, frozenset, str]: + """The single executor job of a walk. + + Returns ``(files, folders, with_ancestors, resolved_root)``. Opens the root first, so a missing or unreadable root raises ``OSError`` instead of listing nothing. Returns *previous_files* itself when nothing @@ -143,4 +334,4 @@ def walk_root( if same_listing(previous_files, files): files = previous_files folders, with_ancestors = folder_sets(root, files) - return files, folders, with_ancestors + return files, folders, with_ancestors, resolve_root(root) diff --git a/py/api/images.py b/py/api/images.py index 763de13..d7e5094 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -10,10 +10,12 @@ import io import json import os import re +import stat import subprocess # nosec B404 - ffmpeg frames through run_encoder, fixed argv +import sys import time as _time import urllib.parse -from bisect import bisect_left, bisect_right +from bisect import bisect_left, bisect_right, insort from collections.abc import Mapping from pathlib import Path from typing import Callable, NamedTuple @@ -589,6 +591,10 @@ def _workspace_key(entry: tuple, order: str) -> tuple: return (mtime, entry.name, entry.tie, entry.index) +def _newest_key(entry: _Listed) -> tuple: + return _workspace_key(entry, "newest") + + def _rated_key(entry: _Listed, rank: int) -> tuple: """Highest rated: rating 5..1, linked unrated (0), unlinked (-1); then newest.""" return (-rank, *_workspace_key(entry, "rating")) @@ -764,6 +770,34 @@ def _favorite_relative_key(relative: str) -> str: return _path_flavor.normcase(relative).replace(_path_flavor.sep, "/") +def _listing_posix(root, prefix, path) -> str: + """*path*'s POSIX spelling below *root* (*prefix*: ``join(root, "")``).""" + text = str(path) + if text.startswith(prefix): + relative = text[len(prefix) :] + return relative if os.sep == "/" else relative.replace(os.sep, "/") + return path.relative_to(root).as_posix() + + +def _favorite_listing_step(root, listing, added, removed) -> dict | None: + """*listing* after a patched listing's change, or None to rebuild it. + + Only when every removed file is added again under the same spelling (a + rewritten output) and no added file's key is already taken by another + spelling: the full build keeps the first spelling in listing order. + """ + prefix = os.path.join(str(root), "") + spellings = {_listing_posix(root, prefix, path) for path, _mtime in added} + if any(_listing_posix(root, prefix, path) not in spellings for path, _ in removed): + return None + listing = dict(listing) + for posix in spellings: + known = listing.setdefault(_favorite_relative_key(posix), posix) + if known != posix: + return None + return listing + + def _favorite_root_key(root) -> str: """Server-side identity of a gallery root: its real, case-normalized path. @@ -827,6 +861,91 @@ def _walk_media(output_path, max_files): stack.append((entry.path, rel_parts + (entry.name,))) +def _settled_entry(root, relative): + """``(Path, mtime)`` of a settled output as a walk would list it. + + *relative* is its part below *root* (``gallery_listing.relative_part``). + None when a walk would not list it (``_walk_media``'s filters, and every + folder on the way must be listable); raises ``OSError`` when that cannot + be told (the file is gone or unreadable, or its on-disk spelling differs + or cannot be read). Blocking: run it in the executor. + """ + path = Path(os.path.join(os.fspath(root), relative)) + info = os.lstat(path) + if stat.S_ISLNK(info.st_mode) or stat.S_ISDIR(info.st_mode): + return None + # Before the name filters: they compare the walk's (stored) spelling. + on_disk = _on_disk_path(path) + if on_disk is not None and not on_disk.endswith(os.sep + relative): + raise OSError(f"{relative} is spelled differently on disk") + *folders, name = relative.split(os.sep) + if _name_suffix(name).lower() not in MEDIA_EXTENSIONS: + return None + if "thumbnails" in folders or len(folders) > MAX_SCAN_DEPTH: + return None + folder = os.fspath(root) + for part in ("", *folders): + folder = os.path.join(folder, part) if part else folder + if not os.access(folder, os.R_OK | os.X_OK): + return None # the walk cannot list this folder + return path, info.st_mtime + + +def _on_disk_path(path, platform=None): + """The file's path as its folders store the names, where that can differ. + + Windows and macOS file systems ignore case. Windows' ``realpath`` gives + the stored case, but keeps the given one when it cannot open the file; + macOS' ``realpath`` always keeps the given one. A walk lists the names as + stored, so an add checks them: Windows through ``GetFinalPathNameByHandle`` + (``nt._getfinalpathname``), macOS through ``F_GETPATH``. Raises OSError + when it cannot be read; None elsewhere (case-sensitive file systems). + """ + platform = platform or sys.platform + if platform == "win32": + import nt + + return nt._getfinalpathname(os.fspath(path)) + if platform != "darwin": + return None + import fcntl + + # Non-blocking: a FIFO named like an image must not hold the executor. + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NONBLOCK", 0)) + try: + raw = fcntl.fcntl(descriptor, fcntl.F_GETPATH, bytes(1024)) + finally: + os.close(descriptor) + return os.fsdecode(raw.split(b"\0", 1)[0]) + + +def _patch_listing(root, files, folders, with_ancestors, relatives): + """The listing with each settled output in *relatives* added (blocking). + + Returns ``(files, folders, with_ancestors, unresolved)``; *files* is the + same list when nothing changed. ``unresolved`` is True when an output + could not be placed exactly, so the root needs a walk. + """ + unresolved = False + for relative in relatives: + try: + entry = _settled_entry(root, relative) + except OSError: + unresolved = True + continue + if entry is None: + continue # a walk would not list it either + patched = gallery_listing.insert_entry(files, entry, MAX_GALLERY_FILES) + if patched is None: + unresolved = True + elif patched is not files: + files = patched + folders, with_ancestors = gallery_listing.add_folder( + root, entry[0], folders, with_ancestors + ) + return files, folders, with_ancestors, unresolved + + def _iter_media_files(output_path, max_files): """Yield the ``Path`` of each media file under *output_path* (see _walk_media).""" for entry in _walk_media(output_path, max_files): @@ -1408,6 +1527,7 @@ class ImageRoutesMixin: even if a previous listing exists. """ record = self._root_listing(output_path) + record.retired = False # a configured root: read, so listened to needed = record.walk_seq + 1 if fresh else 0 discarded = 0 while True: @@ -1433,28 +1553,172 @@ class ImageRoutesMixin: ) def _refresh_if_stale(self, output_path, record): - """Walk a stale listing again, REFRESH_DELAY after this read (on the loop). + """Ask for a background walk when the listing is stale or dirty. - At most one pending refresh per root, none while a walk runs. + A stale listing (older than REFRESH_AFTER) is walked REFRESH_DELAY + after this read, so the walk does not share the GIL with the page's + own requests; a dirty one as soon as its interval allows. """ age = _time.monotonic() - (record.walked_at or 0.0) - if age < gallery_listing.REFRESH_AFTER: + if age >= gallery_listing.REFRESH_AFTER: + self._schedule_walk(output_path, gallery_listing.REFRESH_DELAY) + elif record.dirty: + self._schedule_walk(output_path) + + def output_settled(self, path): + """The monitor settled an output at *path*: list it (on the loop). + + *path* is resolved by the monitor's thread (``notify_output_settled``). + A root with a stored listing and no walk running gets the file added + in the executor (``_run_adds``), without a walk. Every other root + holding it is marked dirty (``mark_root_dirty``): one with a walk + running gets the trailing walk, as that walk may have passed the + file's folder already. + """ + for key, record in self._settled_records(path): + relative = None + if ( + record.files is not None + and record.flight is None + and record.resolved_root is not None + ): + relative = gallery_listing.relative_part(record.resolved_root, path) + if relative is None: + self._mark_dirty(key, record) + continue + record.pending_adds.append(relative) + if record.add_task is None: + task = asyncio.get_running_loop().create_task( + self._run_adds(Path(key), record) + ) + record.add_task = task + self._listing_tasks.add(task) + task.add_done_callback(self._listing_tasks.discard) + + def mark_root_dirty(self, path): + """A settled output at *path*: walk every root holding it again (on the loop). + + *path* is resolved by the monitor's thread (``notify_output_settled``), + so this does no file I/O: it compares ``normcase`` strings with each + walk's ``resolved_root``. Only roots with a listing, or a walk running, + are marked; a root without either is walked by its next read anyway. + """ + for key, record in self._settled_records(path): + self._mark_dirty(key, record) + + def _settled_records(self, path): + """``(key, record)`` of every listed or walking root holding *path*.""" + if self._listings_closed: + return [] + found = [] + for key, record in list(self._root_listings.items()): + if record.retired or (record.files is None and record.flight is None): + continue + resolved = record.resolved_root or os.path.normcase(key) + if gallery_listing.root_contains(resolved, path): + found.append((key, record)) + return found + + def _mark_dirty(self, key, record): + if self._listings_closed or record.retired: return - if record.flight is not None or record.rewalk_handle is not None: + if record.files is None and record.flight is None: + return # walked by its next read anyway + record.dirty = True + self._schedule_walk(Path(key)) + + async def _run_adds(self, root, record): + """Add the root's pending settled outputs to its listing (on the loop). + + Each batch is placed in the executor against the stored listing, and + stored only when no walk started and no drop happened meanwhile: a + walk started after the outputs settled lists them itself, and a + dropped listing is walked by its next read. An output that cannot be + placed exactly marks the root dirty instead. + """ + key = str(root) + try: + while record.pending_adds: + relatives, record.pending_adds = record.pending_adds, [] + if record.files is None or record.flight is not None: + continue # dropped, or a walk started after they settled + base, generation, seq = record.files, record.generation, record.walk_seq + files, folders, with_ancestors, unresolved = ( + await self._run_in_executor( + _patch_listing, + root, + base, + record.folders, + record.folders_with_ancestors, + relatives, + ) + ) + if self._listings_closed or record.retired: + return + if record.generation != generation or record.walk_seq != seq: + continue + if files is not base: + record.files = files + record.folders = folders + record.folders_with_ancestors = with_ancestors + record.patched += 1 + self._schedule_workspace_warm() + if unresolved: + self._mark_dirty(key, record) + except asyncio.CancelledError: + raise + except Exception as exc: + self.logger.warning(f"Could not add a new output to the listing: {exc}") + record.pending_adds = [] + self._mark_dirty(key, record) + finally: + record.add_task = None + + def _schedule_walk(self, output_path, delay=0.0): + """Start or defer a background walk of *output_path* (on the loop). + + - A walk is running: nothing; its end re-checks ``dirty``. + - Otherwise the walk is due *delay* from now, and never sooner than + ``rewalk_interval`` after the previous walk started. Due now: start + it. Later: one deferred walk per root, at least MIN_REWALK_DELAY + away; when one is already pending, the earlier deadline wins. + Every walk start cancels the pending one (``_start_walk``). + """ + if self._listings_closed: return - record.rewalk_handle = asyncio.get_running_loop().call_later( - gallery_listing.REFRESH_DELAY, self._delayed_walk, output_path, record + record = self._root_listing(output_path) + if record.flight is not None or record.retired: + return + now = _time.monotonic() + due = now + delay + if record.last_started_at is not None: + interval = gallery_listing.rewalk_interval(record.last_duration) + due = max(due, record.last_started_at + interval) + if due <= now: + self._start_walk(output_path, record) + return + wait = max(gallery_listing.MIN_REWALK_DELAY, due - now) + loop = asyncio.get_running_loop() + handle = record.rewalk_handle + if handle is not None: + if handle.when() - loop.time() <= wait: + return # the pending walk comes first + handle.cancel() + record.rewalk_handle = loop.call_later( + wait, self._delayed_walk, output_path, record ) def _delayed_walk(self, output_path, record): - record.rewalk_handle = None - self._schedule_walk(output_path) + """The deferred walk's timer: clear the handle first, then walk. - def _schedule_walk(self, output_path): - """Start a background walk of *output_path* unless one is running.""" - record = self._root_listing(output_path) - if record.flight is None: - self._start_walk(output_path, record) + It was armed for a stale or dirty listing at the earliest moment the + interval allowed. Only a walk start clears either reason, and every + start cancels this timer, so when it fires the walk is still due. + """ + record.rewalk_handle = None + if self._listings_closed or record.flight is not None or record.retired: + return + self._start_walk(output_path, record) def _start_walk(self, output_path, record): """Start the root's one walk; returns its flight future (on the loop).""" @@ -1462,7 +1726,11 @@ class ImageRoutesMixin: self._listing_loop = loop handle, record.rewalk_handle = record.rewalk_handle, None if handle is not None: - handle.cancel() # this walk is the refresh it was waiting to start + handle.cancel() # this walk is the one it was waiting to start + # This walk lists every output reported so far. + record.dirty = False + record.pending_adds = [] + record.last_started_at = _time.monotonic() record.walk_seq += 1 flight = loop.create_future() record.flight = flight @@ -1482,11 +1750,14 @@ class ImageRoutesMixin: error never reaches a future nobody awaits. """ outcome = None + cancelled = False + started = record.last_started_at try: - files, folders, with_ancestors = await self._run_in_executor( + files, folders, with_ancestors, resolved = await self._run_in_executor( walk_root, self._scan_gallery_files_sync, output_path, record.files ) except asyncio.CancelledError: + cancelled = True outcome = WalkOutcome.failed(RuntimeError(WALK_STOPPED)) raise except Exception as exc: @@ -1496,11 +1767,15 @@ class ImageRoutesMixin: f"Could not list output folder {output_path}: {exc}" ) record.failing = True + # Retried by the 30 s rule, not by a trailing walk. + record.dirty = False outcome = WalkOutcome.failed(exc) else: record.failing = False + record.resolved_root = resolved if record.generation == generation: changed = files is not record.files + record.patched = 0 record.files = files record.folders = folders record.folders_with_ancestors = with_ancestors @@ -1512,10 +1787,14 @@ class ImageRoutesMixin: else: outcome = WalkOutcome.discarded() finally: + record.last_duration = _time.monotonic() - started if record.flight is flight: record.flight = None if not flight.done(): flight.set_result(outcome) + if record.dirty and not cancelled: + # An output settled during this walk: the trailing walk. + self._schedule_walk(output_path) async def get_output_images(self, request): """Get all images from ComfyUI output folder(s).""" @@ -1644,6 +1923,10 @@ class ImageRoutesMixin: cached = self._workspace_listed_cache.get(key) if cached is not None and cached[0] is files: return cached[1] + if cached is not None: + delta = gallery_listing.listing_delta(cached[0], files) + if delta is not None: + return self._workspace_listed_step(key, cached, files, delta) previous = cached[2] if cached is not None else {} listed = [] for path, mtime in files: @@ -1663,7 +1946,42 @@ class ImageRoutesMixin: self._workspace_card_cache[key] = { file: facts for file, facts in cards.items() if file in lookup } - self._workspace_listed_cache[key] = (files, listed, lookup) + self._workspace_listed_cache[key] = (files, listed, lookup, None) + return listed + + def _workspace_listed_step(self, key, cached, files, delta) -> list: + """The _Listed values after a few files were added or replaced. + + A listing patched by a settled output (``output_settled``) shares + every other entry with the cached one, so only the added files are + built; the order of the values does not matter (every sort is a total + order). The step ``(previous values, added, removed)`` lets the sort + orders follow it the same way (``_workspace_orders``). + """ + index, root = key[0], Path(key[1]) + _files, old_listed, old_lookup, _step = cached + added, removed = delta + if not added and not removed: + self._workspace_listed_cache[key] = (files, old_listed, old_lookup, None) + return old_listed + lookup = dict(old_lookup) + gone = [lookup.pop(item) for item in removed if item in lookup] + fresh = [_listed(path, mtime, root, index) for path, mtime in added] + for entry in fresh: + lookup[(entry.path, entry.mtime)] = entry + if gone: + gone_ids = {id(entry) for entry in gone} + listed = [entry for entry in old_listed if id(entry) not in gone_ids] + cards = self._workspace_card_cache.get(key) + if cards: + self._workspace_card_cache[key] = { + file: facts for file, facts in cards.items() if file in lookup + } + else: + listed = list(old_listed) + listed.extend(fresh) + step = (old_listed, fresh, gone) + self._workspace_listed_cache[key] = (files, listed, lookup, step) return listed def _workspace_orders(self, listings: list[tuple]) -> dict: @@ -1678,14 +1996,54 @@ class ImageRoutesMixin: old is new for old, new in zip(cached[0], parts, strict=True) ): return cached[1] - newest = sorted( - (entry for part in parts for entry in part), - key=lambda entry: _workspace_key(entry, "newest"), - ) + newest = None + if cached is not None: + newest = self._newest_step(listings, parts, cached) + if newest is None: + newest = sorted( + (entry for part in parts for entry in part), + key=_newest_key, + ) + # The other orders, the wanted paths and the locators follow lazily. orders = {"newest": newest} self._workspace_order_cache[key] = (parts, orders) return orders + def _newest_step(self, listings, parts, cached) -> list | None: + """The cached newest order with each part's last step applied, or None. + + Only when every changed part is exactly the value list this call + built (*parts*) and one step (``_workspace_listed_step``) past the + part the cached order was built from: the added values are placed by + bisection instead of sorting the whole listing again. Page reads and + the warm build run in executor threads at once, so the per-file cache + may already hold a newer step than *parts*; then this sorts in full. + """ + old_parts, orders = cached + added, gone = [], [] + for (index, root, _files), part, old in zip( + listings, parts, old_parts, strict=True + ): + if part is old: + continue + listed = self._workspace_listed_cache.get((index, str(root))) + if listed is None or listed[1] is not part: + return None + step = listed[3] + if step is None or step[0] is not old: + return None + added.extend(step[1]) + gone.extend(step[2]) + newest = orders["newest"] + if gone: + gone_ids = {id(entry) for entry in gone} + newest = [entry for entry in newest if id(entry) not in gone_ids] + else: + newest = list(newest) + for entry in added: + insort(newest, entry, key=_newest_key) + return newest + def _workspace_wanted(self, listings: list[tuple]) -> frozenset: """The selected listing's normalized paths (cached with its orders).""" orders = self._workspace_orders(listings) @@ -2967,18 +3325,19 @@ class ImageRoutesMixin: cached = self._favorite_listing_cache.get(cache_key) if cached is not None and cached[0] is files: return cached[1] - # From the path strings, as _listed does: relative_to per file was the - # largest cost of a cold first open (#348). - prefix = os.path.join(str(root), "") - listing = {} - for path, _mtime in files: - text = str(path) - if text.startswith(prefix): - relative = text[len(prefix) :] - posix = relative if os.sep == "/" else relative.replace(os.sep, "/") - else: - posix = path.relative_to(root).as_posix() - listing.setdefault(_favorite_relative_key(posix), posix) + listing = None + if cached is not None: + delta = gallery_listing.listing_delta(cached[0], files) + if delta is not None: + listing = _favorite_listing_step(root, cached[1], *delta) + if listing is None: + # From the path strings, as _listed does: relative_to per file was + # the largest cost of a cold first open (#348). + prefix = os.path.join(str(root), "") + listing = {} + for path, _mtime in files: + posix = _listing_posix(root, prefix, path) + listing.setdefault(_favorite_relative_key(posix), posix) self._favorite_listing_cache[cache_key] = (files, listing) return listing diff --git a/tests/test_api_images.py b/tests/test_api_images.py index 4f8cfc6..b93ad42 100644 --- a/tests/test_api_images.py +++ b/tests/test_api_images.py @@ -2540,7 +2540,10 @@ class TestSharedListing(ImageAPITestCase): from py.api import gallery_listing record = self._record() - record.walked_at -= (seconds or gallery_listing.REFRESH_AFTER) + 1 + shift = (seconds or gallery_listing.REFRESH_AFTER) + 1 + record.walked_at -= shift + # An old listing's walk also started long ago (rewalk_interval). + record.last_started_at -= shift async def _settle(self): """Wait for every background walk this API started.""" @@ -3073,6 +3076,1117 @@ class TestSharedListing(ImageAPITestCase): self.assertTrue(task.done()) self.assertIsNone(self.api._workspace_warm_task) + # -- monitor-triggered walks (#348, PR 2) --------------------------- + + def _interval(self, seconds): + """Patch rewalk_interval() to *seconds*, whatever the walk took.""" + for name in ("REWALK_FLOOR", "REWALK_CAP"): + patcher = patch.object(gallery_listing, name, seconds) + patcher.start() + self.addCleanup(patcher.stop) + + def _output(self, name): + """A new output on disk, spelled as the monitor reports it (resolved).""" + return os.path.realpath(make_png(self.output_dir / name)) + + async def _listed(self): + """A stored listing of the root, built by a read.""" + await self._json("GET", self.URL) + await self._settle() + return self._record() + + def _fire(self, record): + """Run the pending deferred walk now, as its timer would.""" + handle = record.rewalk_handle + self.assertIsNotNone(handle, "no deferred walk is pending") + handle.cancel() + self.api._delayed_walk(self.output_dir, record) + + def _remaining(self, handle): + return handle.when() - asyncio.get_running_loop().time() + + async def test_a_settled_output_is_listed_without_a_reload(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(0) + self.api.mark_root_dirty(self._output("b.png")) + self.assertIsNotNone(record.flight) # the interval allows: walks now + self.assertFalse(record.dirty) + await self._settle() + self.assertEqual(self.scan.calls, 2) + self.assertEqual(len(record.files), 2) + _, data = await self._json("GET", self.URL) + self.assertEqual(data["total"], 2) + self.assertEqual(self.scan.calls, 2) + + async def test_the_monitor_thread_hops_onto_the_loop(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(0) + path = self._output("b.png") + calls = [] + real = self.api.output_settled + + def settled(seen): + calls.append(threading.get_ident()) + real(seen) + + self.api.output_settled = settled + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, self.api.notify_output_settled, path) + # The hop was queued before the executor's result: one loop turn runs it. + await asyncio.sleep(0) + self.assertEqual(calls, [threading.get_ident()]) + await self._settle() + self.assertEqual(len(record.files), 2) + self.assertEqual(self.scan.calls, 1) # added, not walked + + async def test_notify_before_the_loop_is_known_does_nothing(self): + self.assertIsNone(self.api._listing_loop) + with patch.object(self.api, "output_settled") as mark: + self.api.notify_output_settled(self._output("b.png")) + await asyncio.sleep(0) + mark.assert_not_called() + + async def test_notify_after_cleanup_does_nothing(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + await self.api.cleanup_gallery_listings(web.Application()) + self._interval(0) + with patch.object(self.api, "output_settled") as mark: + self.api.notify_output_settled(self._output("b.png")) + await asyncio.sleep(0) + mark.assert_not_called() + # A hop queued before the cleanup is dropped when it runs. + self.api.output_settled(self._output("c.png")) + self.api.mark_root_dirty(self._output("d.png")) + self.assertEqual(self.api._listing_tasks, set()) + self.assertEqual(record.pending_adds, []) + self.assertFalse(record.dirty) + self.assertIsNone(record.flight) + self.assertIsNone(record.rewalk_handle) + + async def test_notify_on_a_closed_loop_does_nothing(self): + loop = asyncio.new_event_loop() + loop.close() + self.api._listing_loop = loop + self.api.notify_output_settled(self._output("b.png")) # no exception + + async def test_notify_racing_the_loop_closing_does_nothing(self): + class ClosingLoop: + def is_running(self): + return True + + def call_soon_threadsafe(self, *_args): + raise RuntimeError("Event loop is closed") + + self.api._listing_loop = ClosingLoop() + self.api.notify_output_settled(self._output("b.png")) # no exception + + async def test_an_output_outside_every_root_is_ignored(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(0) + elsewhere = make_png(self.comfy_dir / "output2" / "x.png") + self.api.mark_root_dirty(os.path.realpath(elsewhere)) + self.assertFalse(record.dirty) + self.assertIsNone(record.flight) + self.assertIsNone(record.rewalk_handle) + + async def test_a_root_never_listed_is_not_walked_by_a_mark(self): + # No listing, no walk running: the first read walks anyway. + self.api.mark_root_dirty(self._output("b.png")) + self.assertEqual(self.api._root_listings, {}) + self.assertEqual(self.scan.calls, 0) + + async def test_a_dropped_listing_is_left_to_the_next_read(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self.api.invalidate_gallery_cache() + self._interval(0) + self.api.mark_root_dirty(self._output("b.png")) + self.assertIsNone(record.flight) + self.assertEqual(self.scan.calls, 1) + + async def test_the_walk_records_the_resolved_root_and_its_timing(self): + make_png(self.output_dir / "a.png") + before = images_module._time.monotonic() + record = await self._listed() + self.assertEqual( + record.resolved_root, os.path.normcase(os.path.realpath(self.output_dir)) + ) + self.assertGreaterEqual(record.last_started_at, before) + self.assertLessEqual(record.last_started_at, record.walked_at) + # Measured when the walk ended, after its listing was stored. + self.assertGreaterEqual( + record.last_duration, record.walked_at - record.last_started_at + ) + + async def test_a_failed_walk_records_its_timing_and_clears_dirty(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + self.api.mark_root_dirty(self._output("b.png")) + self.assertTrue(record.dirty) + self.api._scan_gallery_files_sync = _raise + # Moved back, so the failed walk's start is later even on a coarse + # monotonic clock (Windows ticks every ~15.6 ms). + record.last_started_at -= 1000 + record.last_duration = None + started = record.last_started_at + with self.assertLogs(self.api.logger, "WARNING"): + self._fire(record) + await self._settle() + self.assertGreater(record.last_started_at, started) + self.assertIsNotNone(record.last_duration) + self.assertFalse(record.dirty) + self.assertIsNone(record.rewalk_handle) # no trailing walk after a failure + self.assertEqual(len(record.files), 1) # the previous listing is kept + + async def test_a_burst_of_marks_coalesces_into_one_walk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + for i in range(5): + self.api.mark_root_dirty(self._output(f"b{i}.png")) + handle = record.rewalk_handle + self.assertIsNotNone(handle) + self.assertIsNone(record.flight) + self.assertEqual(self.scan.calls, 1) + # Due one interval after the last walk started, not 30 s from now. + due = record.last_started_at + 30 - images_module._time.monotonic() + self.assertAlmostEqual(self._remaining(handle), due, delta=1) + self._fire(record) + await self._settle() + self.assertEqual(self.scan.calls, 2) + self.assertEqual(len(record.files), 6) + self.assertFalse(record.dirty) + self.assertIsNone(record.rewalk_handle) + + async def test_a_mark_during_a_walk_gives_exactly_one_trailing_walk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + gate = self.scan.hold_next() + rescan = asyncio.ensure_future( + self._json("POST", "/prompt_manager/gallery/rescan") + ) + await self.scan.wait_entered() # listed before b.png exists + for name in ("b.png", "c.png"): + self.api.mark_root_dirty(self._output(name)) + # Never armed while a walk runs: the walk's end re-checks dirty. + self.assertIsNone(record.rewalk_handle) + self.assertTrue(record.dirty) + gate.set() + status, _ = await rescan + self.assertEqual(status, 200) + self.assertEqual(len(record.files), 1) + handle = record.rewalk_handle + self.assertIsNotNone(handle) # the trailing walk, deferred by the interval + self.assertGreaterEqual(self._remaining(handle), 0.5) + self._fire(record) + await self._settle() + self.assertEqual(self.scan.calls, 3) + self.assertEqual(len(record.files), 3) + self.assertIsNone(record.rewalk_handle) + self.assertFalse(record.dirty) + + async def test_a_walk_slower_than_the_interval_trails_at_once_and_stops(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(0) # the walk outlasts its interval + gate = self.scan.hold_next() + self.api.mark_root_dirty(self._output("b.png")) + await self.scan.wait_entered() + self.api.mark_root_dirty(self._output("c.png")) + self.assertIsNone(record.rewalk_handle) + gate.set() + await self._settle() + # One walk for the first mark, one trailing walk for the second, then + # nothing: no mark arrived during the trailing walk. + self.assertEqual(self.scan.calls, 3) + self.assertEqual(len(record.files), 3) + self.assertIsNone(record.rewalk_handle) + self.assertIsNone(record.flight) + + async def test_a_read_triggered_walk_cancels_a_pending_rewalk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + self.api.mark_root_dirty(self._output("b.png")) + handle = record.rewalk_handle + status, data = await self._json("POST", "/prompt_manager/gallery/rescan") + self.assertEqual((status, data["total"]), (200, 2)) + self.assertTrue(handle.cancelled()) + self.assertIsNone(record.rewalk_handle) + self.assertFalse(record.dirty) + await self._settle() + self.assertEqual(self.scan.calls, 2) + + async def test_after_the_trailing_walk_a_new_mark_arms_again(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + self.api.mark_root_dirty(self._output("b.png")) + self._fire(record) + await self._settle() + self.assertIsNone(record.rewalk_handle) + self.api.mark_root_dirty(self._output("c.png")) + self.assertIsNotNone(record.rewalk_handle) + self._fire(record) + await self._settle() + self.assertEqual(self.scan.calls, 3) + self.assertEqual(len(record.files), 3) + + async def test_a_deferred_walk_waits_at_least_half_a_second(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + # The interval ends in a few milliseconds: still never a zero delay. + record.last_started_at = images_module._time.monotonic() - 29.99 + self.api.mark_root_dirty(self._output("b.png")) + handle = record.rewalk_handle + self.assertIsNotNone(handle) + self.assertGreaterEqual(self._remaining(handle), 0.45) + handle.cancel() + + async def test_a_mark_moves_a_later_stale_refresh_earlier(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._age() + self._interval(5) + with patch.object(gallery_listing, "REFRESH_DELAY", 30): + await self._json("GET", self.URL) + stale = record.rewalk_handle + self.assertGreater(self._remaining(stale), 25) + # The interval has long passed: the mark's walk starts now, and that + # walk is the refresh the stale read was waiting for. + self.api.mark_root_dirty(self._output("b.png")) + self.assertTrue(stale.cancelled()) + self.assertIsNotNone(record.flight) + self.assertIsNone(record.rewalk_handle) + await self._settle() + self.assertEqual(self.scan.calls, 2) + self.assertEqual(len(record.files), 2) + + async def test_a_stale_read_keeps_an_earlier_pending_rewalk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(20) + self.api.mark_root_dirty(self._output("b.png")) + pending = record.rewalk_handle + when = pending.when() + # Stale by its age, while the interval since the last walk still runs. + record.walked_at -= gallery_listing.REFRESH_AFTER + 1 + with patch.object(gallery_listing, "REFRESH_DELAY", 30): + await self._json("GET", self.URL) + self.assertIs(record.rewalk_handle, pending) + self.assertEqual(pending.when(), when) + self.assertFalse(pending.cancelled()) + + async def test_a_stale_read_moves_a_later_pending_rewalk_earlier(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(15) + self.api.mark_root_dirty(self._output("b.png")) + pending = record.rewalk_handle + self.assertGreater(self._remaining(pending), 10) + # A listing older than REFRESH_AFTER whose interval is over. + self._age() + with patch.object(gallery_listing, "REFRESH_DELAY", 2): + await self._json("GET", self.URL) + self.assertTrue(pending.cancelled()) + replaced = record.rewalk_handle + self.assertIsNotNone(replaced) + self.assertLess(self._remaining(replaced), 3) + self.assertEqual(self.scan.calls, 1) + replaced.cancel() + + async def test_a_dirty_read_rearms_a_lost_rewalk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + record.dirty = True # e.g. its timer was cancelled by hand + await self._json("GET", self.URL) + self.assertIsNotNone(record.rewalk_handle) + self.assertEqual(self.scan.calls, 1) + + async def test_cleanup_cancels_a_pending_rewalk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + self.api.mark_root_dirty(self._output("b.png")) + handle = record.rewalk_handle + await self.api.cleanup_gallery_listings(web.Application()) + self.assertTrue(handle.cancelled()) + self.assertIsNone(record.rewalk_handle) + self.assertEqual(self.scan.calls, 1) + + async def test_a_walk_cancelled_by_cleanup_starts_no_trailing_walk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(0) + gate = self.scan.hold_next() + self.api.mark_root_dirty(self._output("b.png")) + await self.scan.wait_entered() + self.api.mark_root_dirty(self._output("c.png")) + self.assertTrue(record.dirty) + await self.api.cleanup_gallery_listings(web.Application()) + gate.set() + self.assertIsNone(record.flight) + self.assertIsNone(record.rewalk_handle) + self.assertEqual(self.api._listing_tasks, set()) + self.assertEqual(self.scan.calls, 2) + + async def test_a_warm_up_after_cleanup_listens_again(self): + make_png(self.output_dir / "a.png") + self._interval(0) + await self._warmed() + await self.api.cleanup_gallery_listings(web.Application()) + await self._warmed() + record = self._record() + self.api.mark_root_dirty(self._output("b.png")) + await self._settle() + self.assertEqual(len(record.files), 2) + + async def test_a_root_removed_from_the_roots_is_never_walked_again(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(30) + self.api.mark_root_dirty(self._output("b.png")) + handle = record.rewalk_handle + self.assertIsNotNone(handle) + self.api.reset_gallery_roots_cache() # e.g. the gallery root changed + self.assertTrue(handle.cancelled()) + self.assertIsNone(record.rewalk_handle) + self.assertFalse(record.dirty) + self._interval(0) + # The monitor still watches the old root until ComfyUI restarts. + self.api.mark_root_dirty(self._output("c.png")) + self.api._delayed_walk(self.output_dir, record) # a timer that raced + self.assertIsNone(record.flight) + self.assertIsNone(record.files) + self.assertEqual(self.scan.calls, 1) + + async def test_a_walk_running_when_the_roots_change_gets_no_trailing_walk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(0) + gate = self.scan.hold_next() + self.api.mark_root_dirty(self._output("b.png")) + await self.scan.wait_entered() + self.api.reset_gallery_roots_cache() + self.api.mark_root_dirty(self._output("c.png")) + gate.set() + await self._settle() + self.assertEqual(self.scan.calls, 2) + self.assertIsNone(record.files) # the running walk was discarded + self.assertFalse(record.dirty) + + async def test_a_root_read_again_after_the_roots_change_listens_again(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self.api.reset_gallery_roots_cache() + record = await self._listed() # still configured: read and walked + self._interval(0) + self.api.mark_root_dirty(self._output("b.png")) + await self._settle() + self.assertEqual(len(record.files), 2) + + async def test_a_cancelled_walk_starts_no_trailing_walk(self): + make_png(self.output_dir / "a.png") + record = await self._listed() + self._interval(0) + gate = self.scan.hold_next() + self.api.mark_root_dirty(self._output("b.png")) + await self.scan.wait_entered() + self.api.mark_root_dirty(self._output("c.png")) + (task,) = self.api._listing_tasks + seq = record.walk_seq + task.cancel() # e.g. the loop shutting down without on_cleanup + await asyncio.gather(task, return_exceptions=True) + gate.set() + await self._settle() + self.assertTrue(task.cancelled()) + self.assertEqual(record.walk_seq, seq) # no walk started after it + self.assertEqual(self.scan.calls, 2) + self.assertIsNone(record.flight) + self.assertIsNone(record.rewalk_handle) + self.assertEqual(self.api._listing_tasks, set()) + + # -- settled outputs added without a walk (live test round) ----------- + + def _walked(self): + """The listing a full walk of the root gives now (not counted).""" + return self.scan.real(self.output_dir) + + def _at(self, name, mtime): + """A new output on disk with *mtime*, spelled as the monitor reports it.""" + path = make_png(self.output_dir / name) + os.utime(path, (mtime, mtime)) + return os.path.realpath(path) + + def _hold_adds(self): + """Hold each add batch in the executor until the returned gate is set.""" + entered, gate, batches = threading.Event(), threading.Event(), [] + real = images_module._patch_listing + + def held(*args): + batches.append(list(args[4])) + entered.set() + gate.wait(_HANG_LIMIT) + return real(*args) + + patcher = patch.object(images_module, "_patch_listing", held) + patcher.start() + self.addCleanup(patcher.stop) + self.addCleanup(gate.set) + return entered, gate, batches + + async def _entered(self, event): + loop = asyncio.get_running_loop() + if not await loop.run_in_executor(None, event.wait, _HANG_LIMIT): + raise AssertionError("the add never reached the executor") + + def _no_walk_pending(self, record): + self.assertFalse(record.dirty) + self.assertIsNone(record.flight) + self.assertIsNone(record.rewalk_handle) + + async def test_a_settled_output_is_added_without_a_walk(self): + self._at("a.png", 1000) + record = await self._listed() + self._interval(0) + self.api.output_settled(self._at("b.png", 2000)) + self.assertIsNone(record.flight) + await self._settle() + self.assertEqual(self.scan.calls, 1) + self.assertEqual(record.patched, 1) + self._no_walk_pending(record) + self.assertEqual(record.files, self._walked()) + self.assertEqual(str(record.files[0][0]), str(self.output_dir / "b.png")) + _, data = await self._json("GET", self.URL) + self.assertEqual(data["total"], 2) + self.assertEqual(self.scan.calls, 1) + + async def test_adds_land_where_a_walk_lists_them_and_name_new_folders(self): + for index, name in enumerate(("x/a.png", "x/y/b.png", "c.png", "z/d.png")): + self._at(name, 1000 + 100 * index) + record = await self._listed() + old_folders = record.folders + for name, mtime in ( + ("x/y/n1.png", 1150), + ("new/deep/n2.png", 1250), + ("n3.png", 50), + ("x/n4.webp", 9000), + ): + make_png(self.output_dir / name.replace(".webp", ".png")) + if name.endswith(".webp"): + os.replace(self.output_dir / "x" / "n4.png", self.output_dir / name) + os.utime(self.output_dir / name, (mtime, mtime)) + self.api.output_settled(os.path.realpath(self.output_dir / name)) + await self._settle() + self.assertEqual(self.scan.calls, 1) + walked = self._walked() + self.assertEqual(record.files, walked) + self.assertEqual( + [str(path) for path, _ in record.files], [str(path) for path, _ in walked] + ) + expected = gallery_listing.folder_sets(self.output_dir, walked) + self.assertEqual((record.folders, record.folders_with_ancestors), expected) + self.assertIsNot(record.folders, old_folders) + _, data = await self._json("GET", "/prompt_manager/gallery/subfolders") + self.assertIn(os.path.join("new", "deep"), data["subfolders"]) + + async def test_a_rewritten_output_moves_to_its_new_place(self): + self._at("a.png", 1000) + self._at("b.png", 2000) + record = await self._listed() + self.api.output_settled(self._at("a.png", 3000)) + await self._settle() + self.assertEqual(self.scan.calls, 1) + self.assertEqual(record.files, self._walked()) + self.assertEqual([path.name for path, _ in record.files], ["a.png", "b.png"]) + + async def test_a_repeated_report_changes_nothing(self): + self._at("a.png", 1000) + path = self._at("b.png", 2000) + record = await self._listed() + self.api.output_settled(path) + await self._settle() + files, patched = record.files, record.patched + with patch.object(self.api, "_schedule_workspace_warm") as warm: + self.api.output_settled(path) + await self._settle() + warm.assert_not_called() + self.assertIs(record.files, files) + self.assertEqual((record.patched, self.scan.calls), (patched, 1)) + + async def test_the_next_full_walk_keeps_the_patched_listing(self): + self._at("a.png", 1000) + record = await self._listed() + self.api.output_settled(self._at("sub/b.png", 2000)) + await self._settle() + patched = record.files + self._age() + with patch.object(self.api, "_schedule_workspace_warm") as warm: + await self._json("GET", self.URL) # the 30 s rule walks behind + await self._settle() + self.assertEqual(self.scan.calls, 2) + self.assertIs(record.files, patched) # equal: the walk kept it + self.assertEqual(record.patched, 0) + warm.assert_not_called() + + async def test_the_walks_filters_apply_to_an_add(self): + self._at("a.png", 1000) + record = await self._listed() + files = record.files + deep = os.path.join(*[f"d{i}" for i in range(images_module.MAX_SCAN_DEPTH + 1)]) + skipped = [ + self._at(".png", 2000), # no suffix, as Path.suffix on 3.10-3.13 + self._at("notes.txt", 2000), + self._at(os.path.join("thumbnails", "t.png"), 2000), + self._at(os.path.join("x", "thumbnails", "t.png"), 2000), + self._at(os.path.join(deep, "too-deep.png"), 2000), + ] + for path in skipped: + self.api.output_settled(path) + await self._settle() + self.assertIs(record.files, files) + self.assertEqual(record.files, self._walked()) + self._no_walk_pending(record) + self.assertEqual(self.scan.calls, 1) + # The deepest listed folder and a "..png" name are listed by a walk. + deepest = os.path.join(*[f"d{i}" for i in range(images_module.MAX_SCAN_DEPTH)]) + for path in ( + self._at("..png", 3000), + self._at(os.path.join(deepest, "deepest.png"), 4000), + ): + self.api.output_settled(path) + await self._settle() + self.assertEqual(record.files, self._walked()) + self.assertEqual(len(record.files), 3) + self.assertEqual(self.scan.calls, 1) + + async def test_a_symlinked_output_is_not_added(self): + self._at("a.png", 1000) + record = await self._listed() + target = make_png(self.comfy_dir / "elsewhere.png") + link = self.output_dir / "link.png" + try: + os.symlink(target, link) + except (OSError, NotImplementedError): + self.skipTest("symlinks are not available") + files = record.files + self.api.output_settled(str(link)) # as an unresolved report would name it + await self._settle() + self.assertIs(record.files, files) + self.assertEqual(record.files, self._walked()) + self._no_walk_pending(record) + + async def _assert_walks_instead(self, record, report): + self._interval(0) + self.api.output_settled(report) + await self._settle() + self.assertEqual(self.scan.calls, 2) + self.assertEqual(record.patched, 0) + self.assertEqual(record.files, self._walked()) + + async def test_a_tie_in_mtime_walks_instead(self): + self._at("a.png", 1000) + record = await self._listed() + await self._assert_walks_instead(record, self._at("b.png", 1000)) + + async def test_a_report_that_cannot_be_read_walks_instead(self): + self._at("a.png", 1000) + record = await self._listed() + gone = self._at("gone.png", 2000) + os.unlink(gone) + await self._assert_walks_instead(record, gone) + + async def test_a_listing_at_the_cap_walks_instead(self): + self._at("a.png", 1000) + record = await self._listed() + with patch.object(images_module, "MAX_GALLERY_FILES", 1): + self._interval(0) + self.api.output_settled(self._at("b.png", 2000)) + await self._settle() + self.assertEqual((self.scan.calls, record.patched), (2, 0)) + self.assertEqual(len(record.files), 1) # the walk is capped too + + async def test_a_path_that_cannot_be_sliced_walks_instead(self): + self._at("a.png", 1000) + record = await self._listed() + self._at("b.png", 2000) + report = os.path.join(os.path.realpath(self.output_dir), ".", "b.png") + await self._assert_walks_instead(record, report) + + async def test_an_on_disk_spelling_mismatch_walks_instead(self): + self._at("a.png", 1000) + record = await self._listed() + report = self._at("b.png", 2000) + with patch.object(images_module, "_on_disk_path", return_value="/x/B.png"): + await self._assert_walks_instead(record, report) + + async def test_an_unexpected_add_failure_walks_instead(self): + self._at("a.png", 1000) + record = await self._listed() + report = self._at("b.png", 2000) + with patch.object(images_module, "_patch_listing", side_effect=ValueError): + with self.assertLogs(self.api.logger, "WARNING"): + await self._assert_walks_instead(record, report) + self.assertIsNone(record.add_task) + + async def test_a_report_during_a_walk_gets_the_trailing_walk(self): + self._at("a.png", 1000) + record = await self._listed() + self._interval(0) + self._age() + gate = self.scan.hold_next() + await self._json("GET", self.URL) + await self.scan.wait_entered() # listed already: misses b.png + self.api.output_settled(self._at("b.png", 2000)) + self.assertTrue(record.dirty) + self.assertEqual(record.pending_adds, []) + gate.set() + await self._settle() + self.assertEqual(self.scan.calls, 3) + self.assertEqual(record.files, self._walked()) + + async def test_a_walk_started_during_an_add_lists_it_and_the_add_is_dropped(self): + self._at("a.png", 1000) + record = await self._listed() + files = record.files + entered, gate, _ = self._hold_adds() + self.api.output_settled(self._at("b.png", 2000)) + await self._entered(entered) + walk = self.scan.hold_next() + rescan = asyncio.ensure_future( + self._json("POST", "/prompt_manager/gallery/rescan") + ) + await self.scan.wait_entered() + add = record.add_task + gate.set() + await asyncio.wait_for(asyncio.shield(add), _HANG_LIMIT) + # The add ended while the Rescan walk runs: not stored over it. + self.assertIs(record.files, files) + self.assertEqual(record.patched, 0) + walk.set() + status, data = await rescan + await self._settle() + self.assertEqual((status, data["total"]), (200, 2)) + self.assertEqual(self.scan.calls, 2) + self.assertEqual(record.files, self._walked()) + self._no_walk_pending(record) + + async def test_a_roots_change_during_an_add_drops_it(self): + self._at("a.png", 1000) + record = await self._listed() + entered, gate, _ = self._hold_adds() + self.api.output_settled(self._at("b.png", 2000)) + await self._entered(entered) + self.api.reset_gallery_roots_cache() + gate.set() + await self._settle() + self.assertIsNone(record.files) + self.assertEqual((self.scan.calls, record.patched), (1, 0)) + self._no_walk_pending(record) + + async def test_an_invalidation_during_an_add_drops_it(self): + self._at("a.png", 1000) + record = await self._listed() + entered, gate, batches = self._hold_adds() + self.api.output_settled(self._at("b.png", 2000)) + await self._entered(entered) + self.api.output_settled(self._at("c.png", 3000)) # waits for the batch + self.api.invalidate_gallery_cache() + gate.set() + await self._settle() + self.assertEqual(len(batches), 1) # the dropped listing gets no second + self.assertIsNone(record.files) # the next read walks + self.assertEqual((self.scan.calls, record.patched), (1, 0)) + self._no_walk_pending(record) + _, data = await self._json("GET", self.URL) + self.assertEqual((data["total"], self.scan.calls), (3, 2)) + + async def test_an_add_failing_after_a_drop_walks_nothing(self): + self._at("a.png", 1000) + record = await self._listed() + + def drop_then_fail(*_args): + self.api.invalidate_gallery_cache() # a settings save, off the loop + raise ValueError("boom") + + with patch.object(images_module, "_patch_listing", drop_then_fail): + with self.assertLogs(self.api.logger, "WARNING"): + self.api.output_settled(self._at("b.png", 2000)) + await self._settle() + self.assertIsNone(record.files) + self.assertEqual(self.scan.calls, 1) + self._no_walk_pending(record) + + async def test_cleanup_during_an_add_cancels_it(self): + self._at("a.png", 1000) + record = await self._listed() + files = record.files + entered, gate, _ = self._hold_adds() + self.api.output_settled(self._at("b.png", 2000)) + await self._entered(entered) + self.api.output_settled(self._at("c.png", 3000)) + cleanup = asyncio.ensure_future( + self.api.cleanup_gallery_listings(web.Application()) + ) + await asyncio.sleep(0) + gate.set() + await asyncio.wait_for(cleanup, _HANG_LIMIT) + self.assertIs(record.files, files) + self.assertIsNone(record.add_task) + self.assertEqual(self.api._listing_tasks, set()) + self._no_walk_pending(record) + + async def test_a_removed_root_gets_no_add(self): + self._at("a.png", 1000) + record = await self._listed() + self.api.reset_gallery_roots_cache() + self.api.output_settled(self._at("b.png", 2000)) + await self._settle() + self.assertIsNone(record.files) + self.assertEqual(self.api._listing_tasks, set()) + self.assertEqual(self.scan.calls, 1) + + async def test_a_burst_is_added_in_batches_with_one_warm_build_each(self): + self._at("a.png", 1000) + await self.api.warm_gallery_listings(web.Application()) + await self._settle() + record = self._record() + entered, gate, batches = self._hold_adds() + built = [] + real = self.api._build_workspace_caches + + def build(listings): + built.append(len(listings[0][2])) + real(listings) + + with patch.object(self.api, "_build_workspace_caches", side_effect=build): + self.api.output_settled(self._at("b.png", 2000)) + await self._entered(entered) + for index in range(4): # land while the first batch is placed + self.api.output_settled(self._at(f"c{index}.png", 3000 + index)) + gate.set() + await self._settle() + self.assertEqual([len(batch) for batch in batches], [1, 4]) + self.assertEqual(self.scan.calls, 1) + self.assertEqual(record.patched, 2) + self.assertEqual(record.files, self._walked()) + self.assertEqual(built[-1], 6) + self.assertLessEqual(len(built), 2) + self.assertTrue(self._workspace_ready()) + + async def _workspace_pages(self, **params): + """Every workspace order's full first page, as file names.""" + pages = {} + for order in images_module.WORKSPACE_ORDERS: + query = "&".join(f"{k}={v}" for k, v in {"order": order, **params}.items()) + _, data = await self._json( + "GET", f"{self.URL}?mode=workspace&limit=100&{query}" + ) + pages[order] = [image["filename"] for image in data["images"]] + return pages + + def _drop_workspace_caches(self): + self.api._workspace_listed_cache = {} + self.api._workspace_order_cache = {} + self.api._workspace_card_cache = {} + self.api._favorite_listing_cache = {} + + async def test_workspace_pages_after_adds_match_a_cold_rebuild(self): + for index, name in enumerate(("a.png", "x/b.png", "x/c.mp4", "d.png")): + path = self.output_dir / name + path.parent.mkdir(exist_ok=True) + path.write_bytes(b"x") + os.utime(path, (1000 + index, 1000 + index)) + await self._warmed() + for name in ("x/b.png", "d.png"): + status, _ = await self._json( + "PUT", + "/prompt_manager/images/favorite", + json={"root_index": 0, "path": name}, + ) + self.assertEqual(status, 200) + await self._workspace_pages() + await self._workspace_pages(favorites=1) + self.api.output_settled(self._at("y/new.png", 5000)) # a new folder + self.api.output_settled(self._at("x/b.png", 6000)) # rewritten favorite + await self._settle() + self.assertEqual(self.scan.calls, 1) + patched = ( + await self._workspace_pages(), + await self._workspace_pages(favorites=1), + ) + self._drop_workspace_caches() + cold = (await self._workspace_pages(), await self._workspace_pages(favorites=1)) + self.assertEqual(patched, cold) + self.assertEqual(patched[0]["newest"][:2], ["b.png", "new.png"]) + self.assertEqual(sorted(patched[1]["newest"]), ["b.png", "d.png"]) + + async def test_a_warm_build_after_an_add_builds_only_the_added_values(self): + for index in range(300): + path = self.output_dir / f"f{index:03d}.png" + path.write_bytes(b"x") + os.utime(path, (1000 + index, 1000 + index)) + await self._warmed() + record = self._record() + listed, keys = [], [] + real_listed, real_key = images_module._listed, images_module._newest_key + + def counted_listed(*args): + listed.append(args[0].name) + return real_listed(*args) + + def counted_key(entry): + keys.append(entry) + return real_key(entry) + + with ( + patch.object(images_module, "_listed", counted_listed), + patch.object(images_module, "_newest_key", counted_key), + patch.object( + images_module, + "_favorite_relative_key", + wraps=images_module._favorite_relative_key, + ) as favorite_keys, + ): + self.api.output_settled(self._at("new.png", 9000)) + await self._settle() + self.assertEqual(self.scan.calls, 1) + self.assertTrue(self._workspace_ready()) + self.assertEqual(listed, ["new.png"]) + self.assertLess(len(keys), 20) # bisection, not a sort of 301 files + self.assertLess(favorite_keys.call_count, 10) # not one per file + self.assertEqual( + self.api._workspace_order([(0, self.output_dir, record.files)], "newest")[ + 0 + ].path.name, + "new.png", + ) + + async def test_a_rewritten_output_drops_its_old_card_facts(self): + self._at("a.png", 1000) + self._at("b.png", 2000) + await self._warmed() + await self._json("GET", f"{self.URL}?mode=workspace&limit=10") + key = (0, str(self.output_dir)) + before = set(self.api._workspace_card_cache.get(key, {})) + self.assertIn(Path(self._at("a.png", 1000)).name, {p.name for p, _ in before}) + self.api.output_settled(self._at("a.png", 3000)) + await self._settle() + after = self.api._workspace_card_cache.get(key, {}) + self.assertNotIn((self.output_dir / "a.png", 1000.0), after) + self.assertIn((self.output_dir / "b.png", 2000.0), after) + + def _synthetic(self, root, count, start=1000.0): + return [(root / f"f{i:04d}.png", start + i) for i in range(count)][::-1] + + def test_the_newest_order_sorts_again_when_it_missed_a_step(self): + root = self.output_dir + files = self._synthetic(root, 50) + listings = [(0, root, files)] + self.api._workspace_orders(listings) + one = gallery_listing.insert_entry(files, (root / "n1.png", 5000.0), 100) + self.api._workspace_listed(0, root, one) # values follow, the order not + two = gallery_listing.insert_entry(one, (root / "n2.png", 6000.0), 100) + newest = self.api._workspace_orders([(0, root, two)])["newest"] + self.assertEqual([entry.path for entry in newest], [p for p, _m in two]) + + def test_the_newest_order_follows_one_changed_root_of_two(self): + first, second = self.output_dir, self.comfy_dir / "second" + files = self._synthetic(first, 30) + others = self._synthetic(second, 30, start=1000.5) + self.api._workspace_orders([(0, first, files), (1, second, others)]) + patched = gallery_listing.insert_entry(others, (second / "n.png", 1010.25), 100) + with patch.object( + images_module, "_newest_key", wraps=images_module._newest_key + ) as keys: + newest = self.api._workspace_orders( + [(0, first, files), (1, second, patched)] + )["newest"] + self.assertLess(keys.call_count, 20) + expected = sorted( + [(p, m, 0) for p, m in files] + [(p, m, 1) for p, m in patched], + key=lambda item: -item[1], + ) + self.assertEqual( + [entry.path for entry in newest], [path for path, _m, _i in expected] + ) + + def test_the_on_disk_spelling_is_read_where_names_fold_case(self): + read = images_module._on_disk_path + self.assertIsNone(read(__file__, platform="linux")) + fake_fcntl = MagicMock(F_GETPATH=50) + fake_fcntl.fcntl.return_value = b"/Out/Sub/A.png" + bytes(10) + fake_nt = MagicMock() + fake_nt._getfinalpathname.return_value = "\\\\?\\C:\\Out\\A.png" + with patch.dict(sys.modules, {"fcntl": fake_fcntl, "nt": fake_nt}): + self.assertEqual(read(__file__, platform="darwin"), "/Out/Sub/A.png") + self.assertEqual(read(__file__, platform="win32"), "\\\\?\\C:\\Out\\A.png") + self.assertEqual(fake_fcntl.fcntl.call_args.args[1], 50) + fake_nt._getfinalpathname.assert_called_once_with(__file__) + + def test_a_fifo_named_like_an_image_never_blocks_the_spelling_read(self): + if not hasattr(os, "mkfifo"): + self.skipTest("no FIFOs here") + fifo = self.output_dir / "pipe.png" + os.mkfifo(fifo) + fake = MagicMock(F_GETPATH=50) + fake.fcntl.return_value = os.fsencode(str(fifo)) + done = [] + + def read(): + with patch.dict(sys.modules, {"fcntl": fake}): + done.append(images_module._on_disk_path(fifo, platform="darwin")) + + reader = threading.Thread(target=read, daemon=True) + reader.start() + reader.join(_HANG_LIMIT) + if reader.is_alive(): + # Unblock the open so the thread ends, then fail. + with open(fifo, "wb"): + pass + self.fail("opening the FIFO blocked") + self.assertEqual(done, [str(fifo)]) + + async def test_a_folder_the_walk_cannot_list_gets_no_add(self): + if os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() == 0: + self.skipTest("needs POSIX permissions and a non-root user") + self._at("a.png", 1000) + record = await self._listed() + report = self._at(os.path.join("locked", "b.png"), 2000) + locked = self.output_dir / "locked" + os.chmod(locked, 0o300) # enter but not list + try: + self.assertEqual(self._walked(), record.files) # the walk skips it + files = record.files + self.api.output_settled(report) + await self._settle() + finally: + os.chmod(locked, 0o700) + self.assertIs(record.files, files) + self._no_walk_pending(record) + + async def test_a_folder_named_like_an_image_is_not_added(self): + self._at("a.png", 1000) + record = await self._listed() + folder = self.output_dir / "folder.png" + folder.mkdir() + files = record.files + self.api.output_settled(os.path.realpath(folder)) + await self._settle() + self.assertIs(record.files, files) + self._no_walk_pending(record) + + async def test_the_filters_read_the_stored_spelling(self): + # Reported as thumbnails/, stored as Thumbnails/: a walk lists it, so + # the add must not skip it as a preview folder; the spelling differs, + # so the root is walked. + self._at("a.png", 1000) + record = await self._listed() + report = self._at(os.path.join("thumbnails", "t.png"), 2000) + stored = os.path.join(str(self.output_dir), "Thumbnails", "t.png") + with patch.object(images_module, "_on_disk_path", return_value=stored): + await self._assert_walks_instead(record, report) + + def test_a_stale_read_racing_a_newer_build_sorts_in_full(self): + root = self.output_dir + f0 = self._synthetic(root, 50) + f1 = gallery_listing.insert_entry(f0, (root / "n1.png", 5000.0), 100) + f2 = gallery_listing.insert_entry(f1, (root / "n2.png", 6000.0), 100) + self.api._workspace_orders([(0, root, f0)]) + key = (0, str(root)) + built_from_f0 = self.api._workspace_listed_cache[key] + real = self.api._workspace_listed + armed = [True] + + def listed(index, part_root, files): + values = real(index, part_root, files) + if armed[0] and files is f1: + armed[0] = False + # Another executor thread read the f0 values before this one + # stored f1's, and stores its f2 step (from f0) only now. + self.api._workspace_listed_cache[key] = built_from_f0 + real(index, part_root, f2) + return values + + with patch.object(self.api, "_workspace_listed", side_effect=listed): + stale = self.api._workspace_orders([(0, root, f1)])["newest"] + self.assertEqual([entry.path for entry in stale], [p for p, _m in f1]) + for files in (f1, f2, f1, f2): + newest = self.api._workspace_orders([(0, root, files)])["newest"] + self.assertEqual([entry.path for entry in newest], [p for p, _m in files]) + + def test_a_roots_change_during_an_order_build_sorts_in_full(self): + root = self.output_dir + f0 = self._synthetic(root, 20) + f1 = gallery_listing.insert_entry(f0, (root / "n1.png", 5000.0), 100) + self.api._workspace_orders([(0, root, f0)]) + real = self.api._workspace_listed + + def listed(index, part_root, files): + values = real(index, part_root, files) + # A roots reset in another thread, between this call's two cache + # reads: the per-file values are gone, the order not yet. + self.api._workspace_listed_cache = {} + return values + + with patch.object(self.api, "_workspace_listed", side_effect=listed): + newest = self.api._workspace_orders([(0, root, f1)])["newest"] + self.assertEqual([entry.path for entry in newest], [p for p, _m in f1]) + + def test_the_favorite_listing_step_rebuilds_when_it_cannot_follow(self): + root = self.output_dir + listing = {"a.png": "a.png"} + step = images_module._favorite_listing_step + self.assertEqual( + step(root, listing, [(root / "b.png", 2.0)], []), + {"a.png": "a.png", "b.png": "b.png"}, + ) + self.assertEqual(listing, {"a.png": "a.png"}) # never changed in place + # A removed file not added again, or a key another spelling holds. + self.assertIsNone(step(root, listing, [], [(root / "a.png", 1.0)])) + with patch.object( + images_module, "_favorite_relative_key", side_effect=str.lower + ): + self.assertIsNone(step(root, listing, [(root / "A.png", 2.0)], [])) + + async def test_monitor_walks_during_a_workspace_build_add_one_more_build(self): + make_png(self.output_dir / "a.png") + release, entered = threading.Event(), threading.Event() + built = [] + real = self.api._build_workspace_caches + + def build(listings): + built.append(len(listings[0][2])) + if len(built) == 1: + entered.set() + release.wait(_HANG_LIMIT) + real(listings) + + self.addCleanup(release.set) + self._interval(0) + with patch.object(self.api, "_build_workspace_caches", side_effect=build): + await self.api.warm_gallery_listings(web.Application()) + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, entered.wait, _HANG_LIMIT) + for name in ("b.png", "c.png", "d.png"): # three monitor walks + self.api.mark_root_dirty(self._output(name)) + while self.api._listing_tasks: + await asyncio.gather(*list(self.api._listing_tasks)) + await asyncio.sleep(0) + release.set() + await self._settle() + self.assertEqual(built, [1, 4]) + self.assertEqual(self.scan.calls, 4) + self.assertTrue(self._workspace_ready()) + class TestGalleryListingHostHooks(unittest.IsolatedAsyncioTestCase): async def test_root_registers_the_warm_up_and_its_cleanup(self): @@ -3092,3 +4206,26 @@ class TestGalleryListingHostHooks(unittest.IsolatedAsyncioTestCase): ) as (package, host): self.assertIn(package.api.warm_gallery_listings, host.app.on_startup) self.assertIn(package.api.cleanup_gallery_listings, host.app.on_cleanup) + + async def test_root_connects_the_monitor_to_the_listing(self): + try: + from test_init_package import fake_server_module, load_package + except ImportError: + from tests.test_init_package import fake_server_module, load_package + + with tempfile.TemporaryDirectory() as directory: + server = fake_server_module() + server.PromptServer.instance.app = web.Application() + with load_package( + "gallery_listing_monitor_pkg", + str(Path(directory) / "db.sqlite"), + directory, + server, + ) as (package, host): + monitor = package._global_image_monitor + self.assertEqual( + monitor.output_listener, package.api.notify_output_settled + ) + # The first cleanup hook stops the monitor's hand-offs. + await host.app.on_cleanup[0](host.app) + self.assertIsNone(monitor.output_listener) diff --git a/tests/test_api_workspace_images.py b/tests/test_api_workspace_images.py index d5f1aaa..60110db 100644 --- a/tests/test_api_workspace_images.py +++ b/tests/test_api_workspace_images.py @@ -188,6 +188,7 @@ class TestWorkspaceImages(ImageAPITestCase): newest = self.api._workspace_order_cache[key][1]["newest"] favorites = self.api._favorite_listing_cache[str(self.output_dir)] record.walked_at -= gallery_listing.REFRESH_AFTER + 1 + record.last_started_at -= gallery_listing.REFRESH_AFTER + 1 with patch.object( self.api, "_scan_gallery_files_sync", diff --git a/tests/test_gallery_listing.py b/tests/test_gallery_listing.py index 3a57922..ca796bf 100644 --- a/tests/test_gallery_listing.py +++ b/tests/test_gallery_listing.py @@ -1,10 +1,13 @@ """Pure helpers of the shared Images listing (py/api/gallery_listing.py, #348).""" +import ntpath import os +import posixpath import sys import tempfile import unittest -from pathlib import Path, PurePath, PureWindowsPath +from pathlib import Path, PurePath, PurePosixPath, PureWindowsPath +from types import SimpleNamespace from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -16,7 +19,13 @@ from py.api.gallery_listing import ( # noqa: E402 STORED, RootListing, WalkOutcome, + add_folder, folder_sets, + insert_entry, + listing_delta, + relative_part, + rewalk_interval, + root_contains, same_listing, walk_root, ) @@ -123,24 +132,36 @@ class WalkRootTests(unittest.TestCase): def test_returns_the_listing_and_both_folder_sets(self): files = [(self.root / "sub" / "a.png", 2.0), (self.root / "b.png", 1.0)] scan, calls = self._scan(files) - listed, folders, with_ancestors = walk_root(scan, self.root, None) + listed, folders, with_ancestors, resolved = walk_root(scan, self.root, None) self.assertEqual(calls, [self.root]) self.assertEqual(listed, files) self.assertEqual(folders, frozenset({"sub"})) self.assertEqual(with_ancestors, frozenset({"sub"})) + self.assertEqual(resolved, os.path.normcase(os.path.realpath(self.root))) + + def test_the_resolved_root_follows_a_symlinked_root(self): + link = Path(self._tmp.name) / "linked" + try: + os.symlink(self.root, link, target_is_directory=True) + except (OSError, NotImplementedError) as error: + self.skipTest(f"symlinks unavailable: {error}") + scan, _calls = self._scan([]) + resolved = walk_root(scan, link, None)[3] + self.assertEqual(resolved, os.path.normcase(os.path.realpath(self.root))) + self.assertNotEqual(resolved, os.path.normcase(str(link))) def test_an_unchanged_walk_returns_the_previous_list_object(self): files = [(self.root / "sub" / "a.png", 2.0)] previous = list(files) scan, _calls = self._scan(files) - listed, _folders, _ancestors = walk_root(scan, self.root, previous) + listed, _folders, _ancestors, _resolved = walk_root(scan, self.root, previous) self.assertIs(listed, previous) def test_a_changed_walk_returns_the_new_list(self): previous = [(self.root / "sub" / "a.png", 2.0)] files = [(self.root / "sub" / "a.png", 3.0)] scan, _calls = self._scan(files) - listed, _folders, _ancestors = walk_root(scan, self.root, previous) + listed, _folders, _ancestors, _resolved = walk_root(scan, self.root, previous) self.assertIsNot(listed, previous) self.assertEqual(listed, files) @@ -180,6 +201,278 @@ class RecordTests(unittest.TestCase): self.assertEqual(gallery_listing.REFRESH_DELAY, 1.0) self.assertIsNone(RootListing().rewalk_handle) + def test_a_new_record_is_clean_and_never_walked(self): + record = RootListing() + self.assertFalse(record.dirty) + self.assertIsNone(record.resolved_root) + self.assertIsNone(record.last_started_at) + self.assertIsNone(record.last_duration) + + def test_drop_forgets_pending_adds(self): + record = RootListing(files=[], pending_adds=["a.png"], patched=2) + record.drop() + self.assertEqual((record.pending_adds, record.patched), ([], 0)) + + def test_drop_keeps_the_resolved_root_and_the_walk_timing(self): + # A drop clears the listing, not what is known about the root itself. + record = RootListing(resolved_root="/o", last_started_at=1.0) + record.last_duration = 0.5 + record.dirty = True + record.drop() + self.assertEqual(record.resolved_root, "/o") + self.assertEqual((record.last_started_at, record.last_duration), (1.0, 0.5)) + self.assertTrue(record.dirty) + + +class RewalkIntervalTests(unittest.TestCase): + def test_constants(self): + self.assertEqual( + ( + gallery_listing.REWALK_FLOOR, + gallery_listing.REWALK_FACTOR, + gallery_listing.REWALK_CAP, + gallery_listing.MIN_REWALK_DELAY, + ), + (5, 5, 15, 0.5), + ) + + def test_short_walks_wait_the_floor(self): + self.assertEqual(rewalk_interval(0.2), 5) + self.assertEqual(rewalk_interval(0), 5) + self.assertEqual(rewalk_interval(None), 5) # never walked + + def test_the_interval_scales_with_the_last_walk(self): + self.assertEqual(rewalk_interval(1.0), 5) + self.assertEqual(rewalk_interval(2.0), 10) + self.assertAlmostEqual(rewalk_interval(2.5), 12.5) + + def test_slow_walks_are_capped(self): + self.assertEqual(rewalk_interval(3.0), 15) + self.assertEqual(rewalk_interval(60.0), 15) + + def test_the_constants_are_read_when_called(self): + with ( + patch.object(gallery_listing, "REWALK_FLOOR", 0), + patch.object(gallery_listing, "REWALK_CAP", 0), + ): + self.assertEqual(rewalk_interval(2.0), 0) + + +class RootContainsTests(unittest.TestCase): + def test_posix_paths_under_the_root_match(self): + root = posixpath.normcase("/srv/out") + self.assertTrue(root_contains(root, "/srv/out/a.png", posixpath)) + self.assertTrue(root_contains(root, "/srv/out/sub/deep/a.png", posixpath)) + + def test_a_sibling_sharing_the_prefix_does_not_match(self): + self.assertFalse(root_contains("/srv/out", "/srv/out2/a.png", posixpath)) + self.assertFalse(root_contains("/srv/out", "/srv/outside.png", posixpath)) + self.assertFalse(root_contains("/srv/out", "/srv/other/a.png", posixpath)) + + def test_posix_keeps_case(self): + self.assertFalse(root_contains("/srv/out", "/srv/OUT/a.png", posixpath)) + + def test_the_root_itself_is_not_an_output(self): + self.assertFalse(root_contains("/srv/out", "/srv/out", posixpath)) + + def test_windows_folds_case_and_separators(self): + root = ntpath.normcase("C:\\ComfyUI\\Output") + self.assertTrue(root_contains(root, "c:\\comfyui\\output\\a.png", ntpath)) + self.assertTrue(root_contains(root, "C:/ComfyUI/OUTPUT/sub/a.PNG", ntpath)) + self.assertFalse(root_contains(root, "C:\\ComfyUI\\Output2\\a.png", ntpath)) + self.assertFalse(root_contains(root, "D:\\ComfyUI\\Output\\a.png", ntpath)) + + def test_a_windows_drive_root_and_a_unc_share(self): + drive = ntpath.normcase("E:\\") + self.assertTrue(root_contains(drive, "e:\\x.png", ntpath)) + self.assertFalse(root_contains(drive, "F:\\x.png", ntpath)) + share = ntpath.normcase("\\\\Server\\Share\\Out") + self.assertTrue(root_contains(share, "\\\\server\\share\\out\\a.png", ntpath)) + self.assertFalse(root_contains(share, "\\\\server\\share\\a.png", ntpath)) + + def test_a_windows_extended_length_prefix_is_ignored(self): + root = ntpath.normcase("C:\\ComfyUI\\Output") + long_file = "\\\\?\\C:\\ComfyUI\\Output\\deep\\a.png" + self.assertTrue(root_contains(root, long_file, ntpath)) + prefixed_root = ntpath.normcase("\\\\?\\C:\\ComfyUI\\Output") + self.assertTrue( + root_contains(prefixed_root, "C:\\ComfyUI\\Output\\a.png", ntpath) + ) + share = ntpath.normcase("\\\\Server\\Share\\Out") + unc_file = "\\\\?\\UNC\\server\\share\\out\\a.png" + self.assertTrue(root_contains(share, unc_file, ntpath)) + self.assertFalse( + root_contains(root, "\\\\?\\C:\\ComfyUI\\Output2\\a.png", ntpath) + ) + + def test_posix_keeps_a_literal_question_mark_folder(self): + self.assertFalse(root_contains("/srv/out", "//?/srv/out/a.png", posixpath)) + + def test_the_default_is_this_platforms_rule(self): + root = os.path.normcase(os.path.realpath(tempfile.gettempdir())) + inside = os.path.join(root, "x.png") + self.assertTrue(root_contains(root, inside)) + self.assertTrue(root_contains(root, inside.upper()) == (os.name == "nt")) + if __name__ == "__main__": unittest.main() + + +class RelativePartTests(unittest.TestCase): + def test_posix_keeps_the_case_and_the_nested_folders(self): + self.assertEqual( + relative_part("/srv/out", "/srv/out/Day 1/A.png", posixpath), + "Day 1/A.png", + ) + + def test_posix_outside_the_root_a_sibling_or_the_root_itself_is_none(self): + for path in ("/srv/other/a.png", "/srv/out2/a.png", "/srv/out", "/srv/out/"): + self.assertIsNone(relative_part("/srv/out", path, posixpath), path) + + def test_a_root_spelled_with_a_trailing_separator(self): + self.assertEqual(relative_part("/", "/a.png", posixpath), "a.png") + + def test_windows_matches_folded_and_slices_the_original_case(self): + root = ntpath.normcase("C:\\ComfyUI\\Output") + self.assertEqual( + relative_part(root, "c:/COMFYUI/output/Sub/ComfyUI_001.PNG", ntpath), + "Sub\\ComfyUI_001.PNG", + ) + + def test_windows_extended_length_and_unc_forms(self): + root = ntpath.normcase("C:\\Out") + self.assertEqual( + relative_part(root, "\\\\?\\C:\\Out\\Deep\\a.png", ntpath), + "Deep\\a.png", + ) + share = ntpath.normcase("\\\\Server\\Share\\Out") + self.assertEqual( + relative_part(share, "\\\\?\\UNC\\Server\\Share\\Out\\B.png", ntpath), + "B.png", + ) + drive = ntpath.normcase("D:\\") + self.assertEqual(relative_part(drive, "D:\\X\\a.png", ntpath), "X\\a.png") + + def test_a_case_fold_that_changes_the_length_is_not_sliced(self): + # str.lower folds the Turkish dotted I to two characters (Windows' + # own normcase keeps the length; the guard covers any fold that not). + folding = SimpleNamespace( + sep="\\", normcase=lambda text: text.replace("/", "\\").lower() + ) + path = "C:\\Out\\\u0130mage.png" + self.assertNotEqual(len(folding.normcase(path)), len(path)) + self.assertIsNone(relative_part("c:\\out", path, folding)) + self.assertEqual(relative_part("c:\\out", "C:\\Out\\Ab.png", folding), "Ab.png") + + def test_dot_parts_are_never_sliced(self): + for path in ("/srv/out/../x.png", "/srv/out/./x.png", "/srv/out//x.png"): + self.assertIsNone(relative_part("/srv/out", path, posixpath), path) + + +class InsertEntryTests(unittest.TestCase): + def setUp(self): + self.root = PurePosixPath("/out") + self.files = [ + (self.root / "c.png", 30.0), + (self.root / "b.png", 20.0), + (self.root / "a.png", 10.0), + ] + + def test_newest_first_position_at_the_front_middle_and_end(self): + for mtime, index in ((40.0, 0), (25.0, 1), (15.0, 2), (5.0, 3)): + entry = (self.root / "n.png", mtime) + patched = insert_entry(self.files, entry, 100) + self.assertIs(patched[index], entry) + self.assertEqual(len(patched), 4) + self.assertEqual(len(self.files), 3) # never changed in place + + def test_unchanged_entries_are_the_same_objects(self): + patched = insert_entry(self.files, (self.root / "n.png", 40.0), 100) + self.assertTrue(all(new is old for new, old in zip(patched[1:], self.files))) + + def test_the_same_path_with_a_new_mtime_moves(self): + patched = insert_entry(self.files, (PurePosixPath("/out/a.png"), 35.0), 100) + self.assertEqual( + [(str(path), mtime) for path, mtime in patched], + [("/out/a.png", 35.0), ("/out/c.png", 30.0), ("/out/b.png", 20.0)], + ) + + def test_an_entry_already_listed_returns_the_listing_itself(self): + self.assertIs( + insert_entry(self.files, (PurePosixPath("/out/b.png"), 20.0), 3), self.files + ) + + def test_a_tie_in_mtime_is_not_placed(self): + self.assertIsNone(insert_entry(self.files, (self.root / "n.png", 20.0), 100)) + + def test_a_tie_with_the_replaced_entry_itself_is_placed(self): + files = [(self.root / "b.png", 20.0), (self.root / "a.png", 10.0)] + patched = insert_entry(files, (self.root / "a.png", 20.5), 100) + self.assertEqual([mtime for _path, mtime in patched], [20.5, 20.0]) + + def test_the_cap_refuses_a_new_path_but_replaces_a_listed_one(self): + self.assertIsNone(insert_entry(self.files, (self.root / "n.png", 40.0), 3)) + patched = insert_entry(self.files, (self.root / "a.png", 40.0), 3) + self.assertEqual(str(patched[0][0]), "/out/a.png") + + def test_an_empty_listing(self): + entry = (self.root / "n.png", 1.0) + self.assertEqual(insert_entry([], entry, 100), [entry]) + + +class AddFolderTests(unittest.TestCase): + def test_a_new_nested_folder_adds_it_and_its_ancestors(self): + root = PurePath("/out") + folders, with_ancestors = folder_sets(root, [(root / "x" / "a.png", 1.0)]) + new = add_folder(root, root / "y" / "z" / "b.png", folders, with_ancestors) + self.assertEqual( + new, + folder_sets( + root, [(root / "x" / "a.png", 1.0), (root / "y" / "z" / "b.png", 2.0)] + ), + ) + + def test_a_known_folder_or_the_root_keeps_the_same_sets(self): + root = PurePath("/out") + sets = folder_sets(root, [(root / "x" / "a.png", 1.0)]) + for path in (root / "x" / "b.png", root / "c.png"): + new = add_folder(root, path, *sets) + self.assertIs(new[0], sets[0]) + self.assertIs(new[1], sets[1]) + + def test_windows_spelling_matches_folder_sets(self): + root = PureWindowsPath("C:/Out") + path = root / "A" / "B" / "c.png" + self.assertEqual( + add_folder(root, path, frozenset(), frozenset()), + folder_sets(root, [(path, 1.0)]), + ) + + +class ListingDeltaTests(unittest.TestCase): + def setUp(self): + self.old = [(PurePath(f"/out/{i}.png"), float(i)) for i in range(10)] + + def test_an_added_entry(self): + entry = (PurePath("/out/n.png"), 99.0) + self.assertEqual(listing_delta(self.old, [entry, *self.old]), ([entry], [])) + + def test_a_replaced_entry(self): + entry = (PurePath("/out/3.png"), 99.0) + new = insert_entry(self.old, entry, 100) + self.assertEqual(listing_delta(self.old, new), ([entry], [self.old[3]])) + + def test_the_same_listing_has_no_change(self): + self.assertEqual(listing_delta(self.old, list(self.old)), ([], [])) + + def test_a_walk_builds_new_entries_and_is_past_the_limit(self): + walked = [(PurePath(str(path)), mtime) for path, mtime in self.old] + self.assertEqual(walked, self.old) + self.assertIsNone(listing_delta(self.old, walked, limit=5)) + self.assertIsNone(listing_delta(self.old, walked)) # shares nothing + self.assertIsNone(listing_delta(self.old, self.old[:2], limit=5)) + + def test_an_emptied_listing_and_a_first_file(self): + self.assertEqual(listing_delta(self.old[:1], []), ([], self.old[:1])) + self.assertIsNone(listing_delta([], self.old[:1])) diff --git a/tests/test_image_monitor_singleton.py b/tests/test_image_monitor_singleton.py index b065910..c77811c 100644 --- a/tests/test_image_monitor_singleton.py +++ b/tests/test_image_monitor_singleton.py @@ -449,6 +449,109 @@ class TestHandlerScheduling(unittest.TestCase): self.assertEqual(self.processed, [first, second]) +class TestOutputListener(unittest.TestCase): + """Every settled output is reported to the Images listing (#348, PR 2).""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.handler = im_mod.ImageGenerationHandler(MagicMock(), MagicMock()) + self.addCleanup(self.handler.shutdown) + self.handler.processing_delay = 0 + self.handler.settle_interval = 0.01 + self.handler.process_new_image = lambda path, prompt_snapshot=None: None + self.reported = [] + self.handler.output_listener = self.reported.append + + def _image(self, name, content=b"x" * 16): + path = os.path.join(self.tmp.name, name) + with open(path, "wb") as f: + f.write(content) + return path + + def test_a_settled_output_is_reported_once_with_its_resolved_path(self): + path = self._image("a.png") + self.handler._process_pending(path) + self.assertEqual(self.reported, [os.path.realpath(path)]) + + def test_a_symlinked_spelling_is_reported_resolved(self): + real_dir = os.path.join(self.tmp.name, "real") + os.makedirs(real_dir) + link_dir = os.path.join(self.tmp.name, "link") + try: + os.symlink(real_dir, link_dir, target_is_directory=True) + except (OSError, NotImplementedError) as error: + self.skipTest(f"symlinks unavailable: {error}") + with open(os.path.join(real_dir, "b.png"), "wb") as f: + f.write(b"x" * 16) + self.handler._process_pending(os.path.join(link_dir, "b.png")) + self.assertEqual( + self.reported, [os.path.realpath(os.path.join(real_dir, "b.png"))] + ) + + def test_it_is_reported_even_when_linking_raises(self): + def explode(path, prompt_snapshot=None): + raise RuntimeError("boom") + + self.handler.process_new_image = explode + path = self._image("c.png") + with self.assertRaises(RuntimeError): + self.handler._process_pending(path) + self.assertEqual(self.reported, [os.path.realpath(path)]) + self.assertEqual(self.handler.pending_count(), 0) + + def test_an_output_that_never_settles_is_not_reported(self): + self.handler.settle_attempts = 3 + self.handler._process_pending(os.path.join(self.tmp.name, "gone.png")) + empty = self._image("empty.png", content=b"") + self.handler._process_pending(empty) + self.assertEqual(self.reported, []) + + def test_a_video_is_reported_when_its_extension_is_monitored(self): + self.handler.supported_extensions = (".png", ".mp4") + path = self._image("clip.mp4") + self.handler.on_created(FakeEvent(path)) + self.assertTrue(wait_for(lambda: self.reported != [])) + self.assertEqual(self.reported, [os.path.realpath(path)]) + + def test_a_failing_listener_is_logged_and_the_worker_carries_on(self): + def broken(_path): + raise RuntimeError("listing gone") + + self.handler.output_listener = broken + first = self._image("d.png") + with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs: + self.handler._process_pending(first) # no exception + self.assertIn("Images listing", "\n".join(logs.output)) + self.handler.output_listener = self.reported.append + second = self._image("e.png") + self.handler.schedule(second) + self.assertTrue(wait_for(lambda: self.reported != [])) + self.assertEqual(self.reported, [os.path.realpath(second)]) + + def test_no_listener_reports_nothing(self): + self.handler.output_listener = None + self.handler._process_pending(self._image("f.png")) # no exception + self.assertEqual(self.reported, []) + + def test_the_monitor_wires_running_and_later_handlers(self): + monitor = im_mod.ImageMonitor(MagicMock(), MagicMock()) + listener = self.reported.append + monitor.set_output_listener(listener) + with ( + patch.object(im_mod, "Observer", FakeObserver), + patch.object(im_mod.ImageMonitor, "_gallery_config", return_value=None), + ): + monitor.start_monitoring([self.tmp.name]) + try: + self.assertIs(monitor.handler.output_listener, listener) + monitor.set_output_listener(None) + self.assertIsNone(monitor.output_listener) + self.assertIsNone(monitor.handler.output_listener) + finally: + monitor.stop_monitoring() + + class TestModuleIsolation(unittest.TestCase): """This module must not leave mocks behind for later test modules.""" diff --git a/utils/image_monitor.py b/utils/image_monitor.py index 3edbb44..773af07 100644 --- a/utils/image_monitor.py +++ b/utils/image_monitor.py @@ -106,6 +106,10 @@ class ImageGenerationHandler(FileSystemEventHandler): # ``queue_capture_thumbnail``, wired at start-up; #249). It only # queues: the preview is written on its own worker. None: no queue. self.thumbnail_queue: Optional[Callable[[str], str]] = None + # Told about every settled output, with its resolved path (the API's + # ``notify_output_settled``, wired at start-up; #348): the Images + # listing walks that root again. None: nobody is told. + self.output_listener: Optional[Callable[[str], None]] = None # Read from GalleryConfig if available, otherwise use defaults try: @@ -259,8 +263,12 @@ class ImageGenerationHandler(FileSystemEventHandler): f"Image never settled or vanished, skipping: {image_path}" ) return - self.process_new_image(image_path, prompt_snapshot=prompt_snapshot) - self._queue_thumbnail(image_path) + try: + self.process_new_image(image_path, prompt_snapshot=prompt_snapshot) + self._queue_thumbnail(image_path) + finally: + # Linked or not, even if linking raised: the file exists. + self._report_output(image_path) finally: with self._pending_lock: self._pending.pop(image_path, None) @@ -285,6 +293,25 @@ class ImageGenerationHandler(FileSystemEventHandler): return self.logger.debug(f"Preview {outcome}: {os.path.basename(image_path)}") + def _report_output(self, image_path: str) -> None: + """Tell the output listener a settled output exists, best effort. + + The path is resolved here, on the worker thread, so the listener's + event loop does no file I/O. A failure is logged, never raised: the + output is then listed by the listing's 30 s refresh. + """ + listener = getattr(self, "output_listener", None) + if listener is None: + return + try: + listener(os.path.realpath(image_path)) + except Exception: + self.logger.warning( + f"{os.path.basename(image_path)} could not be reported to the " + "Images listing", + exc_info=True, + ) + def wait_until_settled(self, image_path: str, stop=None) -> bool: """True once the file's size is non-zero and unchanged across two reads. @@ -848,6 +875,14 @@ class ImageMonitor: self.logger = get_logger("prompt_manager.image_monitor") self.review_recorder: Optional[Callable[[str], str]] = None self.thumbnail_queue: Optional[Callable[[str], str]] = None + self.output_listener: Optional[Callable[[str], None]] = None + + def set_output_listener(self, listener: Optional[Callable[[str], None]]) -> None: + """Report every settled output's resolved path to ``listener`` (None + stops it), on the running handler and every later one (#348).""" + self.output_listener = listener + if self.handler is not None: + self.handler.output_listener = listener def set_thumbnail_queue( self, queue_preview: Optional[Callable[[str], str]] @@ -925,6 +960,7 @@ class ImageMonitor: handler = ImageGenerationHandler(self.db_manager, self.prompt_tracker) handler.review_recorder = self.review_recorder handler.thumbnail_queue = self.thumbnail_queue + handler.output_listener = self.output_listener observer = Observer() scheduled = [] for output_dir in valid_directories: