refactor(thumbnails): keep the after-scan queue's store rule and pin the write lock (#320)

A planned target keeps the plan's own containment rule (inside the
store, as the thumbnail job writes it); captures keep the serve rule.
The structural test also covers PREVIEW_FOLDER joins and the
repo-root modules; the write-lock test asserts the lock instead of
waiting on a timer; new tests cover nested roots in the after-scan
queue and a legacy upper-case preview served through the resolver.

py/api/autotag_routes.py

@@ -20,7 +20,7 @@ from .jobs import (
 
 from .maintenance import disk_plan
 from .prompts import thumbnail_url_for
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .. import model_gate
 
 try:
@@ -1371,7 +1371,8 @@ class AutotagRoutesMixin:
             files = await self._run_in_executor(
                 self._without_removed, output_path, files
             )
-            has_thumbnails = (output_path / "thumbnails").is_dir()
+            store = preview_store.store_dir(output_path)
+            has_thumbnails = store is not None and store.is_dir()
 
             images = []
             for image_path in files:

py/api/history_current.py

@@ -15,10 +15,10 @@ from typing import BinaryIO
 
 from aiohttp import web
 
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .images import _canonical, _public_path, SERVABLE_EXTENSIONS, VIDEO_EXTENSIONS
 from .images import in_trash_below_roots
-from .images import existing_media_thumbnail, media_thumbnail_rel
+from .images import existing_media_thumbnail, media_preview_path
 
 try:
     from ...utils.execution_outputs import (
@@ -326,12 +326,10 @@ class _OpenedCurrentMedia:
 
 
 def _preview_writable(root: Path, preview: Path) -> bool:
-    """A preview may be written: ``thumbnails`` and every link segment of the
-    preview's folder resolve inside *root* (the serve route's rule, #249)."""
-    folder = root / "thumbnails"
-    return _canonical(folder).is_relative_to(_canonical(root)) and _canonical(
-        preview
-    ).is_relative_to(_canonical(folder))
+    """A preview may be written: the store of *root* and every link segment
+    of the preview's folder resolve inside it (the serve route's rule, #249;
+    ``preview_store.owns``)."""
+    return preview_store.owns(root, preview)
 
 
 def _is_jxl(path) -> bool:
@@ -381,8 +379,12 @@ def _cover_target(resolved, ensure_preview, placeholder, accepts_jxl):
         return _canonical(preview)
     if preview is not None:
         return _canonical(preview)
-    preview = resolved.gallery_root / media_thumbnail_rel(relative)
-    if ensure_preview is not None and _preview_writable(resolved.gallery_root, preview):
+    preview = media_preview_path(resolved.gallery_root, relative)
+    if (
+        ensure_preview is not None
+        and preview is not None
+        and _preview_writable(resolved.gallery_root, preview)
+    ):
         ensure_preview(target, preview, is_video)
     if is_video and placeholder is not None:
         return _placeholder_media(placeholder)

py/api/images.py

@@ -34,7 +34,7 @@ from .gallery_listing import (
     WalkOutcome,
     walk_root,
 )
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .thumbnail_encode import (
     PREVIEW_HELD,
     EncoderClosed,
@@ -1068,13 +1068,20 @@ def _iter_media_files(output_path, max_files):
 
 
 def media_thumbnail_rel(rel_path, settings=None):
-    """``thumbnails/<rel>_thumb<ext>`` (POSIX) for the media file *rel_path*:
-    the name the preview gets when it is written now (#317), images and
-    videos alike (``writer_settings``' extension)."""
+    """``thumbnails/<rel>_thumb<ext>`` (POSIX URL text) for the media file
+    *rel_path*: the name the preview gets when it is written now (#317),
+    images and videos alike (``writer_settings``' extension). Where it is on
+    disk is ``media_preview_path``."""
     settings = settings or writer_settings()
     return thumbnail_settings.thumbnail_rel(rel_path, settings.extension)
 
 
+def media_preview_path(output_path, rel_path, settings=None):
+    """Where the preview of *rel_path* is written now (``preview_store``)."""
+    settings = settings or writer_settings()
+    return preview_store.preview_path(output_path, rel_path, settings.extension)
+
+
 def existing_media_thumbnail(output_path, rel_path):
     """The existing preview *rel_path* owns that a listing or cover shows
     (the format written now first, then older ones), or None. The Images
@@ -2798,13 +2805,32 @@ class ImageRoutesMixin:
                     output_dirs[0], filepath, int(root_param), len(outputs)
                 )
 
+            # Preview text is mapped by the resolver (#320); any other text,
+            # ``Thumbnails/...`` media on a case-sensitive disk included,
+            # keeps the plain join.
+            preview_parts = (
+                tuple(filepath.split("/"))
+                if preview_store.is_preview_text(filepath)
+                else None
+            )
             for output_path in output_dirs:
-                image_path = output_path / filepath
+                if preview_parts is None:
+                    image_path = output_path / filepath
+                else:
+                    image_path = preview_store.request_path(output_path, preview_parts)
+                    if image_path is None:
+                        continue
                 if not image_path.exists():
                     continue
-                # A symlink that points outside the root is a hard deny, not
-                # a fall-through to the next root.
-                if not _is_within(image_path, output_path):
+                # A symlink that points outside the root (a preview: outside
+                # the folder holding its store) is a hard deny, not a
+                # fall-through to the next root.
+                container = (
+                    output_path
+                    if preview_parts is None
+                    else preview_store.store_container(output_path)
+                )
+                if not _is_within(image_path, container):
                     return _forbidden()
                 if in_trash_below_roots(image_path, output_dirs):
                     return _file_not_found()
@@ -2988,6 +3014,7 @@ class ImageRoutesMixin:
         """Blocking thumbnail generation loop (run in executor)."""
         settings = writer_settings(settings)
         start = _time.monotonic()
+        thumbnails_dir = preview_store.given_store(output_path, thumbnails_dir)
         plan = self._plan_thumbnails(output_path, thumbnails_dir, settings=settings)
         payload = self._execute_thumbnail_plan(plan, settings, start)
         payload["thumbnails_path"] = _public_path(thumbnails_dir)
@@ -3000,7 +3027,8 @@ class ImageRoutesMixin:
         start = _time.monotonic()
         plan = self._plan_gallery_thumbnails(output_dirs, settings=settings)
         payload = self._execute_thumbnail_plan(plan, settings, start)
-        payload["thumbnails_path"] = _public_path(output_dirs[0] / "thumbnails")
+        store = preview_store.store_dir(output_dirs[0])
+        payload["thumbnails_path"] = None if store is None else _public_path(store)
         return payload
 
     def _execute_thumbnail_plan(self, plan: dict, settings, start: float) -> dict:
@@ -3043,9 +3071,12 @@ class ImageRoutesMixin:
             if stop is not None and stop():
                 break
             root = Path(root)
+            store = preview_store.store_dir(root)
+            if store is None:
+                continue
             plan = self._plan_thumbnails(
                 root,
-                root / "thumbnails",
+                store,
                 mode=mode,
                 stop=stop,
                 settings=settings,
@@ -3129,7 +3160,7 @@ class ImageRoutesMixin:
         """
         settings = writer_settings(settings)
         output_path = Path(output_path)
-        thumbnails_dir = Path(thumbnails_dir)
+        thumbnails_dir = preview_store.given_store(output_path, thumbnails_dir)
         plan = {key: 0 for key in _PLAN_COUNTS}
         plan.update(targets=[], fresh_files=[])
         folders = {}
@@ -3221,6 +3252,9 @@ class ImageRoutesMixin:
             "dir": src.parent.name,
             "type": "video" if is_video else "image",
         }
+        # Taken when the write starts: a thumbnail location switched since
+        # then drops the write at its rename (#320).
+        guard = preview_store.write_guard(guard)
         dst = Path(dst)
         if dst.suffix != settings.extension:
             dst = dst.with_suffix(settings.extension)
@@ -3501,8 +3535,8 @@ class ImageRoutesMixin:
             self.logger.debug(f"No roots to sweep for preview temp files: {exc}")
             return 0
         for output_path in output_dirs:
-            folder = Path(output_path) / "thumbnails"
-            if folder.is_symlink() or not folder.is_dir():
+            folder = preview_store.store_dir(output_path)
+            if folder is None or folder.is_symlink() or not folder.is_dir():
                 continue
             removed += sweep_stray_temps(folder)[0]
         if removed:
@@ -3520,13 +3554,16 @@ class ImageRoutesMixin:
         directories = []
         for output_path in output_dirs:
             output_path = Path(output_path)
-            thumbnails_dir = output_path / "thumbnails"
+            thumbnails_dir = preview_store.store_dir(output_path)
+            if thumbnails_dir is None:
+                continue
             if not thumbnails_dir.exists() and not thumbnails_dir.is_symlink():
                 continue
+            container = preview_store.store_container(output_path)
             if (
                 thumbnails_dir.is_symlink()
-                or not _is_within(thumbnails_dir, output_path)
-                or _canonical(thumbnails_dir) == _canonical(output_path)
+                or not _is_within(thumbnails_dir, container)
+                or _canonical(thumbnails_dir) == _canonical(container)
             ):
                 raise ValueError(
                     "Safety check failed: invalid thumbnails directory path"

py/api/preview_store.py

@@ -0,0 +1,206 @@
+"""Where previews live on disk (#320): the one resolver.
+
+Every mapping between an image source (a "root"), one of its media files and
+that file's preview on disk goes through this module. URL text stays in
+``thumbnail_settings`` (``thumbnail_rel``: ``thumbnails/<rel>_thumb<ext>``,
+the shape the serve route, ``inspection-state.mjs`` and ``admin.js`` share);
+this module answers where such a preview is on disk.
+
+The store of a source is where its previews live. With the default location
+(mode ``source``, "Inside each source") it is ``<source>/thumbnails``, exactly
+the folder every caller joined by hand before, so nothing moves.
+
+Writes and the switch: every preview writer takes ``write_guard`` when it
+starts (it records the location's ``generation``); the final rename runs
+under ``write_lock`` and only when the generation is unchanged and no hold
+applies (``thumbnail_encode.replace_unless_held``). Changing the location
+bumps the generation under the same lock, so no write that began before can
+land in the old store afterwards. Nothing reads or commits config while
+``write_lock`` is held. The hold check runs under it too, so hold checks take
+no lock (the job registry's ``running()`` is a lock-free snapshot) and
+``bump_generation`` is never called with a registry or config lock held.
+
+All functions are blocking: call them off the event loop.
+"""
+
+import os
+import threading
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+
+from . import thumbnail_settings
+
+try:
+    from ...utils.media_folders import PREVIEW_FOLDER
+except ImportError:  # tests import the package top-level
+    from utils.media_folders import PREVIEW_FOLDER
+
+SOURCE_MODE = "source"
+
+
+@dataclass(frozen=True)
+class Location:
+    """One snapshot of the thumbnail location (take one per request or job)."""
+
+    mode: str = SOURCE_MODE
+    path: str | None = None
+    available: bool = True
+    problem: str | None = None
+    generation: int = 0
+
+
+write_lock = threading.Lock()
+_generation = 0
+
+
+def current():
+    """The thumbnail location now."""
+    return Location(generation=_generation)
+
+
+def generation():
+    return _generation
+
+
+def bump_generation():
+    """Invalidate every write that started before (under ``write_lock``)."""
+    global _generation
+    with write_lock:
+        _generation += 1
+        return _generation
+
+
+def _canonical(path):
+    """Resolved and case-folded, so paths compare on every OS."""
+    return Path(os.path.normcase(os.path.realpath(str(path))))
+
+
+def _within(path, folder):
+    return _canonical(path).is_relative_to(_canonical(folder))
+
+
+# ── stores ───────────────────────────────────────────────────────────
+
+
+def store_container(root, location=None):
+    """The folder a store of *root* must stay inside: the source itself."""
+    return Path(root)
+
+
+def store_dir(root, location=None):
+    """The store of *root* (``<source>/thumbnails``)."""
+    return Path(root) / PREVIEW_FOLDER
+
+
+def ensure_store(root, location=None):
+    """``store_dir`` for a writer. Preview folders below it are made by the
+    write itself."""
+    return store_dir(root, location)
+
+
+def given_store(root, store):
+    """A caller's explicit store argument: the default spelling
+    (``<root>/thumbnails``) means the resolver's store; another folder (only
+    tests pass one) is used as given."""
+    store = Path(store)
+    if store == Path(root) / PREVIEW_FOLDER:
+        return store_dir(root)
+    return store
+
+
+def preview_path(root, media_rel, ext=None, location=None):
+    """Where the preview of *media_rel* (relative to *root*) is, or is to be
+    written, in *ext* (the extension written now when None)."""
+    store = store_dir(root, location)
+    if store is None:
+        return None
+    if ext is None:
+        from .thumbnail_encode import writer_settings
+
+        ext = writer_settings().extension
+    return store / (thumbnail_settings.preview_stem(media_rel) + ext)
+
+
+def request_path(root, parts, location=None):
+    """The disk path of the URL parts ``("thumbnails", ...)``, or None when
+    *parts* is not under an exact ``thumbnails`` first part. The text is
+    checked before (``is_preview_text``)."""
+    parts = tuple(parts)
+    if len(parts) < 2 or parts[0] != PREVIEW_FOLDER:
+        return None
+    store = store_dir(root, location)
+    if store is None:
+        return None
+    return store.joinpath(*parts[1:])
+
+
+def owns(root, path, location=None):
+    """Whether *path* lies inside the store of *root*, resolved: the store
+    stays inside its container and no link takes *path* out of the store."""
+    store = store_dir(root, location)
+    if store is None:
+        return False
+    return _within(store, store_container(root, location)) and _within(path, store)
+
+
+# ── preview text ─────────────────────────────────────────────────────
+
+
+def is_preview_text(filepath):
+    """URL text naming a preview: ``thumbnails/<...>/<name>_thumb<ext>``.
+
+    Text only, before any join: no backslash, drive or stream colon, and no
+    empty, ``.`` or ``..`` segment; ``thumbnails`` exactly as the first part
+    and nowhere else among the folders; the extension, in any case, one a
+    preview may carry (``X_thumb.JPG`` of an older release included).
+    """
+    if not filepath or "\\" in filepath or ":" in filepath:
+        return False
+    parts = filepath.split("/")
+    if len(parts) < 2 or parts[0] != PREVIEW_FOLDER:
+        return False
+    if any(part in ("", ".", "..") for part in parts):
+        return False
+    if PREVIEW_FOLDER in parts[1:-1]:
+        return False
+    stem, extension = os.path.splitext(parts[-1])
+    return (
+        stem.endswith(thumbnail_settings.THUMB_SUFFIX)
+        and len(stem) > len(thumbnail_settings.THUMB_SUFFIX)
+        and extension.lower() in thumbnail_settings.THUMBNAIL_EXTENSIONS
+    )
+
+
+# ── writes ───────────────────────────────────────────────────────────
+
+
+class _WriteGuard:
+    """``guard()`` for one preview write: true while the location has not
+    changed since the write started and *hold* (if any) allows it."""
+
+    __slots__ = ("generation", "_hold")
+
+    def __init__(self, hold=None):
+        self.generation = _generation
+        self._hold = hold
+
+    def __call__(self):
+        if self.generation != _generation:
+            return False
+        return self._hold is None or bool(self._hold())
+
+
+def write_guard(hold=None):
+    """Take when a preview write starts; hand to its final rename."""
+    if isinstance(hold, _WriteGuard):
+        return hold
+    return _WriteGuard(hold)
+
+
+@contextmanager
+def committed(guard=None):
+    """Around a preview's final rename: holds ``write_lock`` and yields
+    whether the write may land (``guard()``, re-checked under the lock)."""
+    with write_lock:
+        yield guard is None or bool(guard())

py/api/thumbnail_auto.py

@@ -67,7 +67,7 @@ from pathlib import Path
 
 from aiohttp import web
 
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .images import (
     in_trash_below_roots,
     MEDIA_EXTENSIONS,
@@ -78,7 +78,7 @@ from .images import (
     _thumbnail_is_fresh,
     _video_placeholder_image,
     _worker_threads,
-    media_thumbnail_rel,
+    media_preview_path,
 )
 from .jobs import (
     DUPLICATE_CLEANUP_JOB_KIND,
@@ -158,6 +158,36 @@ def _dst_key(dst):
     return flight_key(Path(dst).parent, (Path(dst).name,))
 
 
+class DeferredPreview:
+    """The preview of *rel* under *root*, resolved when its write starts
+    (#320): a queued capture follows the thumbnail location at write time,
+    not the one when it was queued."""
+
+    __slots__ = ("root", "rel", "strict")
+
+    def __init__(self, root, rel, *, strict=True):
+        self.root = Path(root)
+        self.rel = Path(rel)
+        # A capture refuses a store linked out of its container (the serve
+        # route's rule); a planned target keeps the plan's rule: inside the
+        # store, as the thumbnail job writes it.
+        self.strict = strict
+
+    def resolve(self):
+        """The preview's path now, or None when nothing may be written."""
+        dst = media_preview_path(self.root, self.rel)
+        if dst is None:
+            return None
+        if self.strict:
+            return dst if preview_store.owns(self.root, dst) else None
+        store = preview_store.store_dir(self.root)
+        return dst if _is_within(dst, store) else None
+
+
+def _resolved(dst):
+    return dst.resolve() if isinstance(dst, DeferredPreview) else Path(dst)
+
+
 @functools.lru_cache(maxsize=1)
 def video_placeholder_jpeg():
     """The placeholder frame as JPEG bytes (a video whose preview is missing)."""
@@ -215,10 +245,19 @@ def _candidates(parts):
 def _current_preview(root, parts):
     """The preview *parts* names, in the format written now (#317): a request
     for an older name (``_thumb.png`` before a conversion) is answered with
-    the current file."""
+    the current file. None when the store of *root* has no place for it."""
     _folder, base, _wanted = _candidates(parts)
     name = base + thumbnail_settings.THUMB_SUFFIX + writer_settings().extension
-    return Path(root).joinpath(*parts[:-1], name)
+    return preview_store.request_path(root, (*parts[:-1], name))
+
+
+def _request_key(root, parts):
+    """The flight key of the preview *parts* under *root*: the key of its
+    disk path (so the serve and history routes share one flight)."""
+    path = preview_store.request_path(root, parts)
+    if path is None:
+        return flight_key(root, parts)
+    return _dst_key(path)
 
 
 def _quick_source(root, parts):
@@ -259,12 +298,10 @@ def _thumbnail_source(root, parts):
     if in_trash_below_roots(found, [root]):
         return None
     rel = Path(*folder_parts, found.name)
-    dst = root / media_thumbnail_rel(rel)
-    if dst != _current_preview(root, parts):
+    dst = media_preview_path(root, rel)
+    if dst is None or dst != _current_preview(root, parts):
         return None
-    if not _is_within(root / "thumbnails", root) or not _is_within(
-        dst, root / "thumbnails"
-    ):
+    if not preview_store.owns(root, dst):
         dst = None
     return found, dst, found.suffix.lower() in VIDEO_EXTENSIONS
 
@@ -284,6 +321,21 @@ def _scan_folder(folder, base, extensions):
     return None
 
 
+def _deferred(roots, src, dst):
+    """The ``DeferredPreview`` of a planned ``(src, dst)``: the root whose
+    store the plan wrote *dst* into (nested roots plan one each); *dst* as
+    planned when none matches."""
+    src = Path(src)
+    for root in roots:
+        root = Path(root)
+        if not src.is_relative_to(root):
+            continue
+        rel = src.relative_to(root)
+        if media_preview_path(root, rel) == Path(dst):
+            return DeferredPreview(root, rel, strict=False)
+    return dst
+
+
 def _stop_stale_workers():
     """Stop capture workers a previous import of this module started."""
     for thread in threading.enumerate():
@@ -425,16 +477,23 @@ class AutoThumbnailer:
 
     def enqueue(self, src, dst, is_video):
         """Queue *dst*'s preview for the worker; False when already queued,
-        full, closed or its worker is refused at interpreter exit. Never
-        blocks on a write."""
-        key = _dst_key(dst)
+        full, closed, nowhere to write it or its worker is refused at
+        interpreter exit. Never blocks on a write. *dst* may be a
+        ``DeferredPreview``: its path is resolved again when the write
+        starts."""
+        now = _resolved(dst)
+        if now is None:
+            return False
+        key = _dst_key(now)
+        if not isinstance(dst, DeferredPreview):
+            dst = now
         with self._changed:
             if self._closed or key in self._queued:
                 return False
             if len(self._queue) >= self._max_queued:
                 return False
             self._queued.add(key)
-            self._queue.append((key, Path(src), Path(dst), bool(is_video)))
+            self._queue.append((key, Path(src), dst, bool(is_video)))
             self._busy = True
             started = True
             if self._start_worker and not self._worker_running:
@@ -514,6 +573,9 @@ class AutoThumbnailer:
     def _write_queued(self, item):
         _key, src, dst, is_video = item
         try:
+            dst = _resolved(dst)
+            if dst is None:
+                return  # nowhere to write it now: made on its first view
             self.ensure(src, dst, is_video)
         except Exception:  # one file never stops the worker
             self._logger.warning(
@@ -847,7 +909,7 @@ class AutoThumbnailMixin:
         for root in output_dirs:
             root = Path(root)
             outcome = await gate.run(
-                flight_key(root, parts),
+                _request_key(root, parts),
                 functools.partial(self._preview_on_demand, root, parts),
                 pool,
             )
@@ -860,7 +922,7 @@ class AutoThumbnailMixin:
                     continue
                 return self._preview_response(BUSY, quick[0], None, quick[1])
             dst = _current_preview(root, parts)
-            if outcome == READY:
+            if outcome == READY and dst is not None:
                 return self._file_response(dst)
             located = await gate.loop.run_in_executor(
                 pool, _thumbnail_source, root, parts
@@ -883,17 +945,31 @@ class AutoThumbnailMixin:
         """
         if self._auto_thumbnails_shut:
             return "skipped"
-        located = self._capture_target(image_path)
+        located = self._capture_source(image_path)
         if located is None:
             return "skipped"
-        src, dst, is_video = located
+        src, root, rel, is_video = located
+        dst = DeferredPreview(root, rel).resolve()
+        if dst is None:
+            return "skipped"
         if _thumbnail_is_fresh(src, dst):
             return "fresh"
-        if self._auto_thumbnails().enqueue(src, dst, is_video):
+        if self._auto_thumbnails().enqueue(src, DeferredPreview(root, rel), is_video):
             return "queued"
         return "skipped"
 
     def _capture_target(self, image_path):
+        """``(src, dst, is_video)`` of a new output's preview now, or None."""
+        located = self._capture_source(image_path)
+        if located is None:
+            return None
+        src, root, rel, is_video = located
+        dst = DeferredPreview(root, rel).resolve()
+        return None if dst is None else (src, dst, is_video)
+
+    def _capture_source(self, image_path):
+        """``(src, root, rel, is_video)`` of a media file under a current
+        root (not a preview, a link or in a trash folder), or None."""
         path = Path(os.path.abspath(str(image_path)))
         if path.suffix.lower() not in MEDIA_EXTENSIONS or not _regular_file(path):
             return None
@@ -906,12 +982,7 @@ class AutoThumbnailMixin:
             rel = Path(os.path.relpath(path, root))
             if "thumbnails" in rel.parts[:-1] or in_trash_below_roots(path, [root]):
                 return None
-            dst = root / media_thumbnail_rel(rel)
-            if not _is_within(root / "thumbnails", root) or not _is_within(
-                dst, root / "thumbnails"
-            ):
-                return None
-            return path, dst, path.suffix.lower() in VIDEO_EXTENSIONS
+            return path, root, rel, path.suffix.lower() in VIDEO_EXTENSIONS
         return None
 
     # ── after a scan ──────────────────────────────────────────────────
@@ -953,7 +1024,7 @@ class AutoThumbnailMixin:
             return "none"
         thumbnailer = self._auto_thumbnails()
         for src, dst, is_video in plan["targets"]:
-            thumbnailer.enqueue(src, dst, is_video)
+            thumbnailer.enqueue(src, _deferred(output_dirs, src, dst), is_video)
         self.logger.info(f"{missing:,} previews missing after the scan; queued")
         return "queued"
 

py/api/thumbnail_encode.py

@@ -46,7 +46,7 @@ from pathlib import Path
 
 from PIL import Image
 
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 
 try:
     from ...utils.logging_config import get_logger
@@ -101,10 +101,15 @@ def temp_sibling(dst, tag="part"):
 
 
 def replace_unless_held(tmp, dst, guard=None):
-    """Rename *tmp* over *dst*, unless ``guard()`` is false (PreviewWriteHeld)."""
-    if guard is not None and not guard():
-        raise PreviewWriteHeld(PREVIEW_HELD)
-    os.replace(tmp, dst)
+    """Rename *tmp* over *dst*, unless ``guard()`` is false (PreviewWriteHeld).
+
+    The check and the rename run under ``preview_store.write_lock`` (#320): a
+    guard from ``preview_store.write_guard`` also refuses a write that started
+    before the thumbnail location changed."""
+    with preview_store.committed(guard) as allowed:
+        if not allowed:
+            raise PreviewWriteHeld(PREVIEW_HELD)
+        os.replace(tmp, dst)
 
 
 # ``temp_sibling``'s names: ``.<stem>.<tag>-<pid>-<thread id><ext>``.

py/api/thumbnail_settings.py

@@ -114,9 +114,16 @@ def validate_edge(value):
 # ── names ────────────────────────────────────────────────────────────
 
 
+def preview_stem(rel_path):
+    """``<rel without extension>_thumb`` (POSIX): a preview's name inside the
+    store of its source (``preview_store``)."""
+    return f"{Path(rel_path).with_suffix('').as_posix()}{THUMB_SUFFIX}"
+
+
 def thumbnail_stem_rel(rel_path):
-    """``thumbnails/<rel without extension>_thumb`` (POSIX)."""
-    return f"thumbnails/{Path(rel_path).with_suffix('').as_posix()}{THUMB_SUFFIX}"
+    """``thumbnails/<rel without extension>_thumb`` (POSIX): URL text only.
+    Where the preview is on disk is ``preview_store``'s answer."""
+    return f"thumbnails/{preview_stem(rel_path)}"
 
 
 def thumbnail_rel(rel_path, extension=None):
@@ -207,8 +214,13 @@ def owned_in_folder(folder, stem, source_extension, preferred=None):
 
 
 def _owned(root, rel_path, preferred):
+    from . import preview_store
+
     rel_path = Path(rel_path)
-    stem = Path(root) / thumbnail_stem_rel(rel_path)
+    store = preview_store.store_dir(root)
+    if store is None:
+        return iter(())
+    stem = store / preview_stem(rel_path)
     return owned_in_folder(stem.parent, stem.name, rel_path.suffix, preferred)
 
 

tests/test_preview_store.py

@@ -0,0 +1,457 @@
+"""One resolver for where previews live on disk (#320 PR 1).
+
+``py/api/preview_store.py`` owns every mapping between an image source, a
+media file and its preview on disk. In source mode (the only mode so far)
+it returns exactly the paths the code built by hand before
+(``<source>/thumbnails/<rel>_thumb<ext>``), so nothing changes on disk. The
+structural test keeps new ``thumbnails`` path joins out of the rest of the
+backend.
+"""
+
+import asyncio
+import os
+import re
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+import unittest.mock
+from unittest.mock import MagicMock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+sys.modules.setdefault("server", MagicMock())
+
+from py.api import preview_store, thumbnail_settings  # noqa: E402
+from py.api.thumbnail_encode import (  # noqa: E402
+    PreviewWriteHeld,
+    replace_unless_held,
+)
+
+REPO = Path(__file__).resolve().parent.parent
+WAIT = 10
+
+
+class SourceModeCase(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.root = Path(self._tmp.name) / "output"
+        self.root.mkdir()
+
+
+class TestSourceModePaths(SourceModeCase):
+    def test_default_location_is_inside_each_source(self):
+        location = preview_store.current()
+        self.assertEqual(location.mode, preview_store.SOURCE_MODE)
+        self.assertTrue(location.available)
+        self.assertIsNone(location.problem)
+
+    def test_store_of_a_source_is_its_thumbnails_folder(self):
+        self.assertEqual(preview_store.store_dir(self.root), self.root / "thumbnails")
+
+    def test_preview_path_matches_the_old_join(self):
+        rel = Path("sub", "deep", "a.png")
+        self.assertEqual(
+            preview_store.preview_path(self.root, rel, ".webp"),
+            self.root / "thumbnails/sub/deep/a_thumb.webp",
+        )
+        self.assertEqual(
+            preview_store.preview_path(self.root, "a.mp4", ".jpg"),
+            self.root / "thumbnails" / "a_thumb.jpg",
+        )
+
+    def test_preview_path_defaults_to_the_written_extension(self):
+        from py.api.thumbnail_encode import writer_settings
+
+        self.assertEqual(
+            preview_store.preview_path(self.root, "a.png").suffix,
+            writer_settings().extension,
+        )
+
+    def test_request_path_maps_url_parts_to_the_old_join(self):
+        parts = ("thumbnails", "sub", "a_thumb.webp")
+        self.assertEqual(
+            preview_store.request_path(self.root, parts), self.root.joinpath(*parts)
+        )
+
+    def test_request_path_is_none_for_anything_but_preview_parts(self):
+        self.assertIsNone(preview_store.request_path(self.root, ("sub", "a.png")))
+        self.assertIsNone(preview_store.request_path(self.root, ("Thumbnails", "a")))
+        self.assertIsNone(preview_store.request_path(self.root, ("thumbnails",)))
+
+    def test_ensure_store_is_the_store(self):
+        self.assertEqual(
+            preview_store.ensure_store(self.root), self.root / "thumbnails"
+        )
+
+    def test_owns_a_preview_inside_the_store(self):
+        inside = self.root / "thumbnails" / "sub" / "a_thumb.webp"
+        self.assertTrue(preview_store.owns(self.root, inside))
+        self.assertFalse(preview_store.owns(self.root, self.root / "a.png"))
+        self.assertFalse(
+            preview_store.owns(self.root, self.root.parent / "elsewhere.webp")
+        )
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    def test_owns_refuses_a_store_linked_out_of_the_source(self):
+        outside = Path(self._tmp.name) / "outside"
+        outside.mkdir()
+        try:
+            (self.root / "thumbnails").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        self.assertFalse(
+            preview_store.owns(self.root, self.root / "thumbnails" / "a_thumb.webp")
+        )
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    def test_owns_refuses_a_preview_folder_linked_out_of_the_store(self):
+        outside = Path(self._tmp.name) / "outside"
+        outside.mkdir()
+        (self.root / "thumbnails").mkdir()
+        try:
+            (self.root / "thumbnails" / "sub").symlink_to(
+                outside, target_is_directory=True
+            )
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        self.assertFalse(
+            preview_store.owns(
+                self.root, self.root / "thumbnails" / "sub" / "a_thumb.webp"
+            )
+        )
+
+    def test_store_container_is_the_source(self):
+        self.assertEqual(preview_store.store_container(self.root), self.root)
+
+    def test_given_store_argument_is_kept_unless_it_is_the_default(self):
+        default = self.root / "thumbnails"
+        other = Path(self._tmp.name) / "custom"
+        self.assertEqual(preview_store.given_store(self.root, default), default)
+        self.assertEqual(preview_store.given_store(self.root, other), other)
+
+
+class TestPreviewText(unittest.TestCase):
+    def test_preview_text_is_an_exact_thumbnails_prefix_and_preview_name(self):
+        for text in (
+            "thumbnails/a_thumb.webp",
+            "thumbnails/sub/a_thumb.jxl",
+            "thumbnails/a_thumb.JPG",
+            "thumbnails/a_thumb.Png",
+        ):
+            with self.subTest(text=text):
+                self.assertTrue(preview_store.is_preview_text(text))
+
+    def test_other_text_is_not_preview_text(self):
+        for text in (
+            "Thumbnails/a_thumb.webp",
+            "thumbnails/a.png",
+            "thumbnails/_thumb.webp",
+            "thumbnails/a_thumb.txt",
+            "sub/thumbnails/a_thumb.webp",
+            "thumbnails/../a_thumb.webp",
+            "thumbnails//a_thumb.webp",
+            "thumbnails/./a_thumb.webp",
+            "thumbnails\\a_thumb.webp",
+            "thumbnails/c:a_thumb.webp",
+            "thumbnails/sub/thumbnails/a_thumb.webp",
+            "thumbnails",
+            "",
+        ):
+            with self.subTest(text=text):
+                self.assertFalse(preview_store.is_preview_text(text))
+
+
+class TestWritesAndTheSwitch(SourceModeCase):
+    def _temp_and_dst(self):
+        folder = self.root / "thumbnails"
+        folder.mkdir()
+        tmp = folder / ".a_thumb.part-1-1.webp"
+        tmp.write_bytes(b"new")
+        return tmp, folder / "a_thumb.webp"
+
+    def test_a_write_from_the_current_generation_lands(self):
+        tmp, dst = self._temp_and_dst()
+        replace_unless_held(tmp, dst, preview_store.write_guard())
+        self.assertEqual(dst.read_bytes(), b"new")
+
+    def test_a_write_whose_generation_changed_is_dropped(self):
+        tmp, dst = self._temp_and_dst()
+        guard = preview_store.write_guard()
+        preview_store.bump_generation()
+        with self.assertRaises(PreviewWriteHeld):
+            replace_unless_held(tmp, dst, guard)
+        self.assertFalse(dst.exists())
+
+    def test_the_hold_check_still_applies(self):
+        tmp, dst = self._temp_and_dst()
+        with self.assertRaises(PreviewWriteHeld):
+            replace_unless_held(tmp, dst, preview_store.write_guard(lambda: False))
+        self.assertFalse(dst.exists())
+
+    def test_the_check_and_the_rename_run_under_the_write_lock(self):
+        """The hold check and the rename hold ``write_lock``, the lock the
+        generation bump takes: a switch cannot fall between them."""
+        tmp, dst = self._temp_and_dst()
+        in_check = threading.Event()
+        release = threading.Event()
+        seen = {}
+
+        def slow_hold():
+            seen["locked"] = preview_store.write_lock.locked()
+            in_check.set()
+            release.wait(WAIT)
+            return True
+
+        guard = preview_store.write_guard(slow_hold)
+        writer = threading.Thread(target=replace_unless_held, args=(tmp, dst, guard))
+        writer.start()
+        try:
+            self.assertTrue(in_check.wait(WAIT))
+            self.assertTrue(seen["locked"])
+            self.assertTrue(preview_store.write_lock.locked())
+        finally:
+            release.set()
+            writer.join(WAIT)
+        self.assertFalse(preview_store.write_lock.locked())
+        self.assertEqual(dst.read_bytes(), b"new")
+        bumped = preview_store.bump_generation()
+        self.assertEqual(preview_store.generation(), bumped)
+
+
+class TestOnlyTheResolverBuildsStorePaths(unittest.TestCase):
+    """No backend module joins ``thumbnails`` onto a path by hand any more."""
+
+    ALLOWED = {
+        Path("py/api/preview_store.py"),
+        Path("py/api/thumbnail_settings.py"),  # URL text
+    }
+    # A path join with a literal "thumbnails" operand.
+    JOIN = re.compile(
+        r"""/\s*["']thumbnails["']"""
+        r"""|["']thumbnails["']\s*/"""
+        r"""|joinpath\([^)]*["']thumbnails["']"""
+        r"""|os\.path\.join\([^)]*["']thumbnails["']"""
+        r"""|["']thumbnails/"""
+        r"""|/\s*PREVIEW_FOLDER|joinpath\([^)]*PREVIEW_FOLDER"""
+        r"""|Path\([^)]*,\s*["']thumbnails["']"""
+    )
+
+    def test_no_thumbnails_join_outside_the_resolver(self):
+        offenders = []
+        files = sorted(REPO.glob("*.py"))
+        for folder in ("py", "utils", "database"):
+            files += sorted((REPO / folder).rglob("*.py"))
+        for path in files:
+            rel = path.relative_to(REPO)
+            if rel in self.ALLOWED:
+                continue
+            for number, line in enumerate(
+                path.read_text(encoding="utf-8").splitlines(), 1
+            ):
+                code = line.split("#", 1)[0]
+                if self.JOIN.search(code):
+                    offenders.append(f"{rel}:{number}: {line.strip()}")
+        self.assertEqual(offenders, [])
+
+
+class TestSettingsLookupsFollowTheStore(SourceModeCase):
+    def test_owned_thumbnails_reads_the_store(self):
+        folder = self.root / "thumbnails" / "sub"
+        folder.mkdir(parents=True)
+        (folder / "a_thumb.webp").write_bytes(b"x")
+        self.assertEqual(
+            thumbnail_settings.owned_thumbnails(self.root, Path("sub", "a.png")),
+            [folder / "a_thumb.webp"],
+        )
+
+
+try:
+    from tests.test_thumbnail_auto import AutoThumbnailCase, _entry
+    from tests.test_api_images import make_png
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_thumbnail_auto import AutoThumbnailCase, _entry
+    from test_api_images import make_png
+
+
+class RedirectedStoreCase(AutoThumbnailCase):
+    """Every caller follows the resolver: with the store of the output root
+    moved to ``<tmp>/elsewhere/output-key`` (a stand-in for a later mode),
+    previews are read, written, listed and removed there and only there."""
+
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.container = self.comfy_dir / "elsewhere"
+        self.store = self.container / "output-key"
+        self.store.mkdir(parents=True)
+        self.old_store = self.output_dir / "thumbnails"
+        root = self.output_dir
+        real_store = preview_store.store_dir
+        real_container = preview_store.store_container
+
+        def store_dir(r, location=None):
+            return self.store if Path(r) == root else real_store(r, location)
+
+        def store_container(r, location=None):
+            if Path(r) == root:
+                return self.container
+            return real_container(r, location)
+
+        for name, fake in (
+            ("store_dir", store_dir),
+            ("store_container", store_container),
+        ):
+            patcher = unittest.mock.patch.object(preview_store, name, fake)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+
+    def _preview(self, rel, ext=".webp"):
+        return self.store / (thumbnail_settings.preview_stem(rel) + ext)
+
+
+class TestEveryCallerFollowsTheStore(RedirectedStoreCase):
+    async def test_an_existing_preview_is_served_from_the_store(self):
+        make_png(self.output_dir / "sub" / "a.png", size=(64, 64))
+        preview = self._preview("sub/a.png")
+        make_png(preview, size=(32, 32))
+        url = _entry(self.api, self.output_dir, "sub/a.png")["thumbnail_url"]
+        self.assertIn("v=", url)  # the listing found it in the store
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(await resp.read(), preview.read_bytes())
+
+    async def test_a_legacy_upper_case_preview_is_served_from_the_store(self):
+        make_png(self.output_dir / "k.JPG", size=(64, 64))
+        legacy = self.store / "k_thumb.JPG"
+        legacy.write_bytes(b"legacy preview")
+        resp = await self.client.get(
+            "/prompt_manager/images/serve/thumbnails/k_thumb.JPG?root=0"
+        )
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(await resp.read(), b"legacy preview")
+
+    async def test_a_missing_preview_is_made_in_the_store_on_first_view(self):
+        make_png(self.output_dir / "sub" / "b.png", size=(640, 480))
+        url = _entry(self.api, self.output_dir, "sub/b.png")["thumbnail_url"]
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertTrue(self._preview("sub/b.png").is_file())
+        self.assertFalse(self.old_store.exists())
+
+    async def test_a_leftover_in_the_source_folder_is_not_served(self):
+        make_png(self.output_dir / "c.png", size=(640, 480))
+        leftover = self.old_store / "c_thumb.webp"
+        leftover.parent.mkdir()
+        leftover.write_bytes(b"old preview bytes")
+        resp = await self.client.get(
+            "/prompt_manager/images/serve/thumbnails/c_thumb.webp?root=0"
+        )
+        self.assertEqual(resp.status, 200)
+        self.assertNotEqual(await resp.read(), b"old preview bytes")
+        self.assertTrue(self._preview("c.png").is_file())
+
+    async def test_capture_writes_into_the_store(self):
+        path = make_png(self.output_dir / "run" / "d.png", size=(64, 64))
+        self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "queued")
+        await self._idle()
+        self.assertTrue(self._preview("run/d.png").is_file())
+        self.assertFalse(self.old_store.exists())
+
+    async def test_generate_and_clear_follow_the_store(self):
+        make_png(self.output_dir / "e.png", size=(64, 64))
+        payload = await asyncio.get_running_loop().run_in_executor(
+            None, self.api._generate_gallery_thumbnails_sync, [self.output_dir]
+        )
+        self.assertTrue(payload["success"])
+        self.assertTrue(self._preview("e.png").is_file())
+        self.assertFalse(self.old_store.exists())
+        self.assertTrue(payload["thumbnails_path"].endswith("output-key"))
+        cleared, _size = self.api._clear_gallery_thumbnails_sync([self.output_dir])
+        self.assertEqual(cleared, 1)
+        self.assertFalse(self._preview("e.png").exists())
+        self.assertTrue(self.store.is_dir())  # the store itself stays
+
+    async def test_removing_a_file_removes_its_preview_in_the_store(self):
+        from py.api.admin import _remove_thumbnail
+
+        make_png(self.output_dir / "f.png", size=(64, 64))
+        preview = self._preview("f.png")
+        preview.write_bytes(b"x")
+        self.assertTrue(_remove_thumbnail(self.output_dir, Path("f.png")))
+        self.assertFalse(preview.exists())
+
+    async def test_the_temp_sweep_walks_the_store(self):
+        stray = self.store / ".g_thumb.part-1-1.webp"
+        stray.write_bytes(b"x")
+        old = os.path.getmtime(stray) - 10 * 24 * 3600
+        os.utime(stray, (old, old))
+        self.assertEqual(self.api.sweep_thumbnail_temps(), 1)
+        self.assertFalse(stray.exists())
+
+    async def test_history_cover_writes_into_the_store(self):
+        from py.api import history_current
+
+        self.assertTrue(
+            history_current._preview_writable(self.output_dir, self._preview("h.png"))
+        )
+        self.assertFalse(
+            history_current._preview_writable(
+                self.output_dir, self.old_store / "h_thumb.webp"
+            )
+        )
+
+
+class TestCaptureResolvesAtWriteTime(RedirectedStoreCase):
+    def test_a_queued_capture_follows_a_store_that_moved_before_its_write(self):
+        from py.api import thumbnail_auto as auto
+
+        written = []
+        thumbnailer = auto.AutoThumbnailer(
+            lambda src, dst, *a, **k: written.append(dst) or {"action": "generated"},
+            lambda: True,
+            self.api.logger,
+            start_worker=False,
+        )
+        self.addCleanup(thumbnailer.close)
+        src = make_png(self.output_dir / "i.png", size=(32, 32))
+        deferred = auto.DeferredPreview(self.output_dir, Path("i.png"))
+        self.assertTrue(thumbnailer.enqueue(src, deferred, False))
+        moved = self.container / "moved"
+        moved.mkdir()
+        self.store = moved  # the location changes before the worker runs
+        thumbnailer._write_queued(thumbnailer._queue.popleft())
+        self.assertEqual(written, [moved / "i_thumb.webp"])
+
+    def test_the_after_scan_queue_maps_each_target_to_its_own_root(self):
+        from py.api import thumbnail_auto as auto
+
+        nested = self.output_dir / "inner"
+        nested.mkdir()
+        src = nested / "j.png"
+        outer = auto._deferred(
+            [self.output_dir, nested], src, self._preview("inner/j.png")
+        )
+        inner = auto._deferred(
+            [self.output_dir, nested], src, nested / "thumbnails" / "j_thumb.webp"
+        )
+        self.assertEqual(
+            (outer.root, outer.rel), (self.output_dir, Path("inner/j.png"))
+        )
+        self.assertEqual((inner.root, inner.rel), (nested, Path("j.png")))
+        stray = self.comfy_dir / "x_thumb.webp"
+        self.assertEqual(auto._deferred([self.output_dir], src, stray), stray)
+
+    def test_serve_and_history_share_one_flight_key(self):
+        from py.api import thumbnail_auto as auto
+
+        parts = ("thumbnails", "sub", "a_thumb.webp")
+        self.assertEqual(
+            auto._request_key(self.output_dir, parts),
+            auto._dst_key(self._preview("sub/a.png")),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()