nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(system): thumbnail job review fixes (#249 S5)

Browse Source 
- The legacy stream's first frame is job {id, started}, sent after the
  subscription so no runner frame is missed; it reads the registry after
  prepare, so a job started in that gap is attached to, not refused.
- Every exit sets a truthful final status line; nothing to do never
  reads "Processing file 0 of 0".
- Cancel is checked per file during the planning walk; a job cancelled
  there publishes no start frame.
- The ffmpeg fallback keeps the chosen quality (-q:v) and the aspect
  ratio; the placeholder uses the chosen JPEG settings.
- A locked preview on Windows reads "Preview in use by another program;
  the old one was kept", not the raw WinError.
- Thumbnail work no longer defers the monitor's Needs review recording:
  it writes previews only, never records.
- The disconnect test now proves the job outlives its stream (one worker,
  9 files, three batches); the attach test waits on an event.
This commit is contained in:
Vito Sansevero
2026-10-23 16:31:56 -07:00
parent 2b4b415638
commit 6f78b4f46f
8 changed files with 608 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TESTING.md
+40 -8
View File
@@ -2172,18 +2172,48 @@ cancellable at a file boundary (a file not yet started is left alone; written pr
stay), outcome groups Generated / Skipped (already fresh, listed without a log line
each) / Errors. `size` is the longest edge and `quality` the encoder (low: JPEG 70, no
optimize; medium: 85, optimized, what the legacy routes always wrote; high: 95,
optimized). `rebuild` writes fresh previews again, each through the same temp-file
optimized). Video previews take the same choice: the OpenCV frame and the placeholder
are saved with those JPEG settings, and the ffmpeg fallback (no OpenCV) maps them to
`-q:v` 8 / 4 / 2 and fits the frame inside the edge keeping its aspect ratio (`-vf
scale=E:E:force_original_aspect_ratio=decrease`, no longer a stretched square).
`rebuild` writes fresh previews again, each through the same temp-file
rename, so nothing is cleared first and a failed file keeps its old preview (a video
whose frame cannot be decoded keeps its preview instead of getting the placeholder);
originals are only read (`tests/test_thumbnail_jobs.py`).
originals are only read (`tests/test_thumbnail_jobs.py`). Cancel is also checked per
file during the planning walk: a job cancelled there stops the walk at the next file,
publishes no `start` frame and ends "Cancelled before any preview was written". Every
exit sets a final status line: "N previews written · M errors" (plus "· R more on the
next run" past the cap), "No previews to generate · N already up to date" (or "No
images or videos to preview"), "Cancelled · N previews written", or the failure; a job
with nothing to do never reads "Processing file 0 of 0" (`test_every_exit_sets_a_truthful_final_status_line`,
and "a finished thumbnail job with nothing to do says so" in `workspace-activity.spec.mjs`).
Windows: replacing a preview another process holds open (the gallery serving it, an
image viewer) fails with WinError 5 or 32; that file counts as an error reading
"Preview in use by another program; the old one was kept" (never the raw "[WinError
5]"), and its old preview stays. Any other PermissionError reads "Permission denied
writing the preview" (`test_a_locked_preview_is_a_readable_coded_error`).
Thumbnail work (the job and the gallery's generate and clear holds) is `guarded`, but it
writes only preview files and no records, so the image monitor's Needs review
recording runs beside it instead of being deferred for a whole rebuild
(`PREVIEW_ONLY_KINDS` in `py/api/review_recording.py`, `blocker(..., compatible=)`;
`test_a_running_thumbnail_job_does_not_defer_monitor_recording`). Other `guarded` and
all `records` jobs still defer it.
Deliberate change: the legacy gallery stream `POST /images/generate-thumbnails/progress`
now starts this job (missing previews, its quality preset as the size, medium encoding)
or attaches to the running one, and streams the job's unchanged `status`/`start`/
`progress`/`file_error`/`complete` frames. A client that goes away no longer stops the
work (`test_thumbnail_job_survives_client_disconnect`); the gallery's Cancel asks the
server to cancel the running thumbnail job, and a job cancelled from Activity ends the
stream with the error frame "Thumbnail generation was cancelled". The blocking
`progress`/`file_error`/`complete` frames after a first `job {id, started}` frame that
names the job and whether this stream started it. The stream reads the registry after
its headers are sent, so a job that starts in that gap is attached to, not refused. A
client that goes away no longer stops the work (`test_thumbnail_job_survives_client_disconnect`:
one worker, 9 files in three batches; the stream sees its client go after the first
batch and all 9 previews are written, where a stream that stops on disconnect leaves 1).
The gallery's Cancel (its settings panel and the startup dialog) cancels by that id,
and only a job its own stream started: a Cancel before the `job` frame keeps the stream
open until it arrives, a double Cancel posts once, and a job another tab or System
started keeps running while the gallery stops following it (`tests/js/gallery-thumbnails.test.js`).
A job cancelled from Activity ends the stream with the error frame "Thumbnail
generation was cancelled". The blocking
`POST /images/generate-thumbnails` and the clear stay legacy holds.
The workspace starts it from "Thumbnail maintenance" (`thumbnail-review.mjs`, the
@@ -2334,7 +2364,9 @@ Recorded deviations (plan D60–D67, D79–D81 renumbered D92–D99 and D111–D
job". Every other kind (the health check, a kind this page does not know) gets the
generic descriptor: title-based toasts and no tiles or Results, so it never shows the
scan's zero tiles (S1 review M6, S2 review M3). A running job without a known total
has an indeterminate `<progress>` (no `value`) and a pill without a percentage. Only a
has an indeterminate `<progress>` (no `value`) and a pill without a percentage, whose
mini bar (`aria-hidden`; the pill's label carries the state) is marked `indeterminate`
and sweeps instead of showing an empty fill (a dimmed full bar under reduced motion). Only a
running job the server can cancel shows Cancel. Tests:
`tests/js/workspace-activity-kinds.test.js`, and in `workspace-activity.spec.mjs` "a
thumbnail job shows Generated Skipped Errors, its pill verb and its toasts" and "a held
@@ -2734,7 +2766,7 @@ pass runs after its own link, best effort: a failure is logged and the link
stays. Recording follows the registry's `records`
admission (#289): the scan's recording runs under its own `records` job, and
the monitor's (one file, not a job) is refused while a `records` or `guarded`
job or hold runs. It then records nothing, logs that the file waits for the
job or hold runs, except thumbnail work, which writes no records (S5 M7). It then records nothing, logs that the file waits for the
next scan, and answers `deferred`; the next "new" scan reads the ambiguous file
again (`test_monitor_recording_is_refused_beside_record_changing_jobs`). A new item moves `review_index_revision`. A
rescan does not copy or hash an ambiguous file whose current size and
py/api/images.py
+55 -12
View File
@@ -282,6 +282,27 @@ THUMBNAIL_ENCODINGS = {
"high": {"quality": 95, "optimize": True},
}
DEFAULT_THUMBNAIL_ENCODING = THUMBNAIL_ENCODINGS["medium"]
# ffmpeg's JPEG scale (-q:v, 2 best to 31 worst) for each encoding's quality.
_FFMPEG_QSCALE = {95: 2, 85: 4, 70: 8}
# A preview that could not be replaced (#249 S5 M6). Windows refuses to rename
# over a file another process has open (WinError 5 or 32), such as a preview
# the gallery is serving; the old preview stays and the file counts as an error.
PREVIEW_IN_USE = "Preview in use by another program; the old one was kept"
PREVIEW_NOT_WRITABLE = "Permission denied writing the preview"
_WINDOWS_LOCKED = (5, 32) # ERROR_ACCESS_DENIED, ERROR_SHARING_VIOLATION
def _preview_write_error(exc):
"""The readable error for a PermissionError while writing a preview."""
if getattr(exc, "winerror", None) in _WINDOWS_LOCKED:
return PREVIEW_IN_USE
return PREVIEW_NOT_WRITABLE
def _ffmpeg_qscale(encoding):
"""ffmpeg's ``-q:v`` for a THUMBNAIL_ENCODINGS entry."""
return str(_FFMPEG_QSCALE.get(encoding["quality"], 4))
def _write_image_thumbnail(src, dst, thumbnail_size, encoding=None):
@@ -1840,12 +1861,13 @@ class ImageRoutesMixin:
return payload
def _plan_gallery_thumbnails(
self, output_dirs: list[Path], rebuild: bool = False
self, output_dirs: list[Path], rebuild: bool = False, stop=None
) -> dict:
"""Plan every gallery root with one shared per-run generation cap.
See ``_plan_thumbnails``; ``fresh_files`` keeps at most
MAX_FRESH_LISTED files across the roots.
MAX_FRESH_LISTED files across the roots. ``stop()`` true ends the walk
at the next file (a cancelled job); the partial plan is returned.
"""
combined = {
"targets": [],
@@ -1855,8 +1877,12 @@ class ImageRoutesMixin:
"fresh_files": [],
}
for root in output_dirs:
if stop is not None and stop():
break
root = Path(root)
plan = self._plan_thumbnails(root, root / "thumbnails", rebuild=rebuild)
plan = self._plan_thumbnails(
root, root / "thumbnails", rebuild=rebuild, stop=stop
)
available = max(0, MAX_THUMBNAIL_FILES - len(combined["targets"]))
combined["targets"].extend(plan["targets"][:available])
combined["skipped"] += plan["skipped"]
@@ -1895,7 +1921,7 @@ class ImageRoutesMixin:
{"success": False, "error": _safe_error(exc)}, status=500
)
def _plan_thumbnails(self, output_path, thumbnails_dir, rebuild=False):
def _plan_thumbnails(self, output_path, thumbnails_dir, rebuild=False, stop=None):
"""Decide which media files need a thumbnail (blocking walk).
Returns ``{"targets": [(src, dst, is_video)], "skipped": n,
@@ -1907,7 +1933,8 @@ class ImageRoutesMixin:
``rebuild`` every file is a target (nothing is skipped). The walk is
bounded like the gallery listing (MAX_SCAN_DEPTH, MAX_GALLERY_FILES,
no symlinks) and at most MAX_THUMBNAIL_FILES targets are queued per
run; ``remaining`` counts the targets beyond that cap.
run; ``remaining`` counts the targets beyond that cap. ``stop()`` true
ends the walk before the next file.
"""
output_path = Path(output_path)
targets = []
@@ -1916,6 +1943,8 @@ class ImageRoutesMixin:
fresh = 0
fresh_files = []
for src in _iter_media_files(output_path, MAX_GALLERY_FILES):
if stop is not None and stop():
break
suffix = src.suffix.lower()
is_video = suffix in VIDEO_EXTENSIONS
rel_no_ext = src.relative_to(output_path).with_suffix("")
@@ -2007,6 +2036,10 @@ class ImageRoutesMixin:
return {**result, "action": "generated"}
except UnidentifiedImageError:
return {**result, "action": "error", "error": "Not a valid image file"}
except PermissionError as e:
message = _preview_write_error(e)
self.logger.warning(f"Thumbnail failed for {src.name}: {message}")
return {**result, "action": "error", "error": message}
except Exception as e:
self.logger.warning(f"Thumbnail failed for {src.name}", exc_info=True)
return {**result, "action": "error", "error": _safe_error(e)}
@@ -2031,10 +2064,13 @@ class ImageRoutesMixin:
):
"""Generate thumbnail from video file. Returns True if successful.
``encoding`` is one of THUMBNAIL_ENCODINGS (medium when None). With
``keep_existing`` an existing preview is kept, and the run fails,
when no frame could be decoded, rather than replaced by the
placeholder.
``encoding`` is one of THUMBNAIL_ENCODINGS (medium when None); it
encodes the OpenCV frame, ffmpeg's frame (as ``-q:v``) and the
placeholder alike. Both decoders fit the frame inside
``thumbnail_size`` keeping its aspect ratio. With ``keep_existing`` an
existing preview is kept, and the run fails, when no frame could be
decoded, rather than replaced by the placeholder. A PermissionError
while replacing the preview propagates (``_generate_one`` names it).
"""
encoding = encoding or DEFAULT_THUMBNAIL_ENCODING
try:
@@ -2082,8 +2118,11 @@ class ImageRoutesMixin:
"00:00:01",
"-vframes",
"1",
"-s",
f"{thumbnail_size[0]}x{thumbnail_size[1]}",
"-vf",
f"scale={thumbnail_size[0]}:{thumbnail_size[1]}"
":force_original_aspect_ratio=decrease",
"-q:v",
_ffmpeg_qscale(encoding),
"-y",
str(tmp_path),
]
@@ -2148,18 +2187,22 @@ class ImageRoutesMixin:
except (OSError, AttributeError):
pass
_atomic_save(img, thumbnail_path, "JPEG", quality=85)
_atomic_save(img, thumbnail_path, "JPEG", **encoding)
self.logger.debug(
f"Generated placeholder video thumbnail: {thumbnail_path}"
)
return True
except PermissionError:
raise
except Exception as e:
self.logger.warning(
f"Failed to create placeholder thumbnail for {video_path}: {e}"
)
return False
except PermissionError:
raise
except Exception as e:
self.logger.error(
f"Video thumbnail generation failed for {video_path}: {e}"
py/api/jobs.py
+16 -7
View File
@@ -292,18 +292,24 @@ class Job:
for queue in list(self._subscribers):
queue.put_nowait(None)
async def events(self):
async def events(self, lead=None):
"""Yield the latest event, then each later one, until the job finishes.
``lead``, when given, is yielded first, after the subscription is in
place, so nothing published while the consumer sends it is missed.
Consume it under ``contextlib.aclosing`` so a consumer that stops early
(a closed socket) drops its queue at once instead of at garbage collection.
"""
queue = asyncio.Queue()
self._subscribers.add(queue)
# As of the subscription: a later event or the end reaches the queue.
snapshot, ended = self.last_event, self.finished
try:
if self.last_event is not None:
yield self.last_event
if self.finished:
if lead is not None:
yield lead
if snapshot is not None:
yield snapshot
if ended:
return
while True:
payload = await queue.get()
@@ -454,11 +460,14 @@ class JobRegistry:
self._jobs[job.id] = job
return job
def blocker(self, kind, *, admission=None, record_changing=None):
def blocker(self, kind, *, admission=None, record_changing=None, compatible=()):
"""The running job a new ``kind`` job would be refused for, or None.
The same kind refuses first; then the oldest running job whose class
the new job's admission class may not run beside.
the new job's admission class may not run beside. Running jobs of a
kind in ``compatible`` never refuse it: the caller knows that work
cannot conflict with its own (the monitor's Needs review recording
beside thumbnail work, which writes previews and no records).
"""
same_kind = self.active(kind)
if same_kind is not None:
@@ -468,7 +477,7 @@ class JobRegistry:
running = [job for job in tuple(self._jobs.values()) if job.running]
running.sort(key=lambda job: job._started_seq)
for job in running:
if job.admission in refused:
if job.admission in refused and job.kind not in compatible:
return job
return None
py/api/review_recording.py
+14 -4
View File
@@ -28,16 +28,23 @@ Recording changes records, so it follows the registry's ``records``
admission (#289). The scan records under its own ``records`` job. The
monitor's pass lasts one file and is not a job: it is refused while a
``records`` or ``guarded`` job runs (``deferred``), and the next scan reads
the file again.
the file again. Thumbnail work (the job and the gallery's generate and clear
holds, kind ``thumbnails``) is ``guarded`` but only writes and removes preview
files under ``thumbnails/`` and never touches the database, so it does not
defer the recording (#249 S5 M7): a long rebuild no longer holds Needs review
back for its whole run.
"""
import os
from pathlib import Path
from .admin import SCAN_LOG_PREFIXES, _plural, _scan_location, _scan_roots
from .jobs import THUMBNAILS_JOB_KIND
# The kind the monitor's recording checks admission under (never registered).
REVIEW_RECORDING_KIND = "review_recording"
# Guarded kinds that change no records, so the recording may run beside them.
PREVIEW_ONLY_KINDS = (THUMBNAILS_JOB_KIND,)
# record_ambiguous_file outcomes that leave the file in the pending queue.
QUEUED_OUTCOMES = ("recorded", "limited", "duplicate")
NEW_OUTCOMES = ("recorded", "limited")
@@ -185,11 +192,14 @@ class ReviewRecordingMixin:
links it); any other file is recorded when it is ambiguous with a
choice, and is ``untraced`` (its pending items without a choice
removed) when it is ambiguous without one. While a
job runs that a ``records`` job may not run beside, nothing is written
and the answer is ``deferred``. Errors propagate: the monitor logs
job runs that a ``records`` job may not run beside, other than
thumbnail work (``PREVIEW_ONLY_KINDS``), nothing is written and the
answer is ``deferred``. Errors propagate: the monitor logs
them and keeps its link.
"""
busy = self._jobs.blocker(REVIEW_RECORDING_KIND, admission="records")
busy = self._jobs.blocker(
REVIEW_RECORDING_KIND, admission="records", compatible=PREVIEW_ONLY_KINDS
)
if busy is not None:
self.logger.info(
f"Needs review was not updated for "
py/api/thumbnail_jobs.py
+73 -27
View File
@@ -19,7 +19,9 @@ old preview; originals are only read.
The legacy gallery stream ``POST /images/generate-thumbnails/progress``
starts this job (its quality preset still sizes the previews) or attaches to
the running one, and streams the job's legacy ``status``/``start``/
the running one. Its first frame, ``job {id, started}``, names that job and
whether this stream started it (the gallery cancels by that id, and only a
job it started); then come the job's legacy ``status``/``start``/
``progress``/``file_error``/``complete`` frames. A client that goes away no
longer stops the work: the job outlives the request, as a scan does.
"""
@@ -65,6 +67,8 @@ _OPTIONS_ERROR = (
f"to {MAX_THUMBNAIL_EDGE}"
)
CANCELLED_ERROR = "Thumbnail generation was cancelled"
CANCELLED_BEFORE_WRITING = "Cancelled before any preview was written"
NOTHING_TO_PREVIEW = "No images or videos to preview"
_MODE_LOG = {"missing": "missing previews", "rebuild": "every preview"}
# Outcome rows: what happened to each file's preview.
_DETAILS = {
@@ -139,6 +143,20 @@ def _frame(event, data):
return {"event": event, "data": data}
def _finished_line(plan, stats):
"""A finished job's status line: what it wrote, or that it had nothing to do."""
if not plan["targets"]:
if plan["skipped"]:
return f"No previews to generate · {plan['skipped']:,} already up to date"
return NOTHING_TO_PREVIEW
parts = [f"{_plural(stats['generated'], 'preview')} written"]
if stats["errors"]:
parts.append(_plural(len(stats["errors"]), "error"))
if plan["remaining"]:
parts.append(f"{plan['remaining']:,} more on the next run")
return " · ".join(parts)
class ThumbnailJobsMixin:
"""The thumbnail plan, the thumbnail job and the legacy stream on it."""
@@ -230,7 +248,11 @@ class ThumbnailJobsMixin:
job.finish(state)
async def _thumbnail_job_body(self, job, options):
"""Plan, then generate in worker batches; returns the final state."""
"""Plan, then generate in worker batches; returns the final state.
Every exit sets the job's final status line (I3): what was written,
that there was nothing to do, or what a cancel kept.
"""
output_dirs = await self._run_in_executor(self._get_all_output_dirs)
if not output_dirs:
message = "ComfyUI output directory not found"
@@ -252,20 +274,22 @@ class ThumbnailJobsMixin:
)
)
plan = await self._run_in_executor(
self._plan_gallery_thumbnails, output_dirs, rebuild=options.rebuild
self._plan_gallery_thumbnails,
output_dirs,
rebuild=options.rebuild,
stop=lambda: job.cancelled,
)
if job.cancelled: # M2: stopped during the walk, before any start frame
return self._end_cancelled(job, 0, CANCELLED_BEFORE_WRITING)
roots = _scan_roots(output_dirs)
self._record_fresh(job, plan, roots)
stats, start = await self._generate_thumbnail_targets(job, options, plan, roots)
self.invalidate_gallery_cache()
generated = _plural(stats["generated"], "preview")
errors = _plural(len(stats["errors"]), "error")
if job.cancelled:
job.log_line(f"Thumbnails cancelled · {generated} written")
job.publish(
_frame("error", {"error": CANCELLED_ERROR, "message": CANCELLED_ERROR})
return self._end_cancelled(
job, stats["generated"], f"Cancelled · {generated} written"
)
return "cancelled"
payload = _thumbnail_complete_payload(
len(plan["targets"]) + plan["skipped"] + plan["remaining"],
stats,
@@ -273,16 +297,27 @@ class ThumbnailJobsMixin:
remaining=plan["remaining"],
processed=len(plan["targets"]),
)
job.update(progress=100)
job.update(progress=100, status_line=_finished_line(plan, stats))
if plan["remaining"]:
job.log_line(
f"{plan['remaining']:,} more previews are left for the next run"
)
errors = _plural(len(stats["errors"]), "error")
job.log_line(f"Thumbnails finished · {generated} written · {errors}")
self.logger.info(payload["message"])
job.publish(_frame("complete", payload))
return "finished"
@staticmethod
def _end_cancelled(job, generated, status_line):
"""Close a cancelled job: its status line, log line and error frame."""
job.update(status_line=status_line)
job.log_line(f"Thumbnails cancelled · {_plural(generated, 'preview')} written")
job.publish(
_frame("error", {"error": CANCELLED_ERROR, "message": CANCELLED_ERROR})
)
return "cancelled"
def _record_fresh(self, job, plan, roots):
"""List the already fresh previews under Skipped (no log line each)."""
rows = []
@@ -310,7 +345,9 @@ class ThumbnailJobsMixin:
)
if plan["skipped"]:
message += f" ({plan['skipped']} already have thumbnails)"
job.update(total=total, status_line=_scan_status_line(0, total, 0, 0))
job.update(total=total)
if total: # never "Processing file 0 of 0" (I3)
job.update(status_line=_scan_status_line(0, total, 0, 0))
job.publish(
_frame(
"start",
@@ -409,40 +446,49 @@ class ThumbnailJobsMixin:
"Connection": "keep-alive",
},
)
running = self._jobs.active(THUMBNAILS_JOB_KIND)
job = running if running is not None and running.cancellable else None
busy = None
if job is None:
busy = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded")
await response.prepare(request)
if busy is None and job is None:
# The gallery's preset sizes the previews; medium encodes them as
# the gallery always did.
options = ThumbnailOptions("missing", thumbnail_size[0], "medium")
try:
job = self._start_thumbnail_job(options)
except JobBusy as refused:
busy = refused.running
# M3: read the registry after the last await, so a job that started
# while the headers went out is attached to, not refused.
job, started, busy = self._legacy_stream_job(thumbnail_size)
if busy is not None:
message = f"{busy.title} already running"
await self._emit_sse(
response, "error", {"error": message, "message": message}
)
return response
# No await between the start and the first read: the stream subscribes
# before the job's runner publishes its first frame.
events = job.events()
# No await between the start and the subscription: the lead frame is
# sent after it, so the stream misses none of the runner's frames.
events = job.events(lead=_frame("job", {"id": job.id, "started": started}))
try:
async for payload in events:
frame = (
f"event: {payload['event']}\ndata: {json.dumps(payload['data'])}"
)
await response.write(f"{frame}\n\n".encode("utf-8"))
except Exception as exc: # the client went away; the job goes on
except Exception as exc: # a write failed: the client went away
self.logger.info(
f"Thumbnail progress client disconnected ({exc}); "
"the thumbnails continue in the background"
)
finally:
# Also when aiohttp cancels the handler on disconnect: the job is
# its own task and goes on.
await events.aclose()
return response
def _legacy_stream_job(self, thumbnail_size):
"""``(job, started, busy)`` for the legacy stream, synchronously.
Attaches to the running cancellable thumbnail job, or starts one with
the gallery's preset as the size (missing previews, medium encoding,
as the gallery always wrote), or names the job in the way.
"""
running = self._jobs.active(THUMBNAILS_JOB_KIND)
if running is not None and running.cancellable:
return running, False, None
busy = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded")
if busy is not None:
return None, False, busy
# blocker() said free and nothing awaited since, so the start is admitted.
options = ThumbnailOptions("missing", thumbnail_size[0], "medium")
return self._start_thumbnail_job(options), True, None
tests/test_jobs.py
+53
View File
@@ -101,6 +101,35 @@ class TestJobFanOut(unittest.IsolatedAsyncioTestCase):
self.assertEqual(await _collect(job), [{"type": "complete"}])
async def test_lead_comes_first_and_nothing_published_meanwhile_is_lost(self):
"""S5 I1: the ``job`` frame leads, with the subscription already in place."""
job = Job("thumbnails", "Thumbnail generation")
job.publish({"type": "status"})
lead = {"type": "job"}
seen = []
async with contextlib.aclosing(job.events(lead=lead)) as events:
async for payload in events:
seen.append(payload)
if payload is lead:
# Published while the consumer is still sending the lead.
job.publish({"type": "progress"})
job.publish({"type": "complete"})
job.finish()
self.assertEqual(
seen,
[lead, {"type": "status"}, {"type": "progress"}, {"type": "complete"}],
)
async def test_lead_on_a_finished_job_then_its_last_event(self):
job = Job("thumbnails", "Thumbnail generation")
job.publish({"type": "complete"})
job.finish()
seen = []
async with contextlib.aclosing(job.events(lead={"type": "job"})) as events:
async for payload in events:
seen.append(payload)
self.assertEqual(seen, [{"type": "job"}, {"type": "complete"}])
async def test_departed_subscriber_is_forgotten(self):
job = Job("scan", "Scan")
job.publish({"type": "progress", "progress": 1})
@@ -151,6 +180,30 @@ class TestJobFanOut(unittest.IsolatedAsyncioTestCase):
class TestJobRegistry(unittest.IsolatedAsyncioTestCase):
async def test_a_compatible_kind_never_blocks(self):
"""S5 M7: a caller may name running kinds that cannot conflict with it."""
registry = JobRegistry()
gate = asyncio.Event()
thumbs = registry.start(
"thumbnails",
"Thumbnail generation",
admission="guarded",
runner=_gated_runner(gate),
)
self.assertIs(registry.blocker("probe", admission="records"), thumbs)
self.assertIsNone(
registry.blocker("probe", admission="records", compatible=("thumbnails",))
)
backup = registry.start(
"backup", "Backup", admission="guarded", runner=_gated_runner(gate)
)
self.assertIs(
registry.blocker("probe", admission="records", compatible=("thumbnails",)),
backup,
)
gate.set()
await asyncio.wait_for(asyncio.gather(thumbs.task, backup.task), 2)
async def test_start_lists_and_finishes_a_job(self):
registry = JobRegistry()
gate = asyncio.Event()
tests/test_review.py
+40 -2
View File
@@ -859,7 +859,9 @@ class TestMonitorRecording(ReviewTestCase):
While a ``records`` or ``guarded`` job runs, the monitor's recording
records nothing and answers ``deferred`` (its fallback link stands and
a later scan reads the ambiguous file again); a ``read`` job does not
refuse it, and it records once the blocking job has ended.
refuse it, and it records once the blocking job has ended. Thumbnail
work (the job and the gallery's holds) is ``guarded`` but writes only
previews, never records, so it does not defer the recording (S5 M7).
"""
self._require_module()
from py.api.jobs import RESTORE_JOB_KIND, THUMBNAILS_JOB_KIND
@@ -870,7 +872,7 @@ class TestMonitorRecording(ReviewTestCase):
for kind, admission in (
(RESTORE_JOB_KIND, "records"),
(THUMBNAILS_JOB_KIND, "guarded"),
("backup", "guarded"),
):
with self.subTest(admission=admission):
async with self.api._jobs.hold(
@@ -888,7 +890,43 @@ class TestMonitorRecording(ReviewTestCase):
)
async with self.api._jobs.hold("probe-read", "Read probe", admission="read"):
async with self.api._jobs.hold(
THUMBNAILS_JOB_KIND,
"Thumbnail clear (gallery page)",
admission="guarded",
):
outcome = await asyncio.to_thread(
self.api.record_review_file, str(path)
)
self.assertEqual(outcome, "recorded")
self.assertEqual(len(self._items()), 1)
async def test_a_running_thumbnail_job_does_not_defer_monitor_recording(self):
"""S5 M7: a long rebuild never holds Needs review back (no records)."""
self._require_module()
from py.api.jobs import THUMBNAILS_JOB_KIND
self._roots()
path = self.output_dir / "beside.png"
write_png(path, prompt=unlinkable_graph("cat", "dog"))
gate = asyncio.Event()
async def runner(job):
await gate.wait()
job = self.api._jobs.start(
THUMBNAILS_JOB_KIND,
"Thumbnail generation",
admission="guarded",
cancellable=True,
scope={},
runner=runner,
)
try:
outcome = await asyncio.to_thread(self.api.record_review_file, str(path))
finally:
gate.set()
await asyncio.wait_for(job.task, 5)
self.assertEqual(outcome, "recorded")
self.assertEqual(len(self._items()), 1)
tests/test_thumbnail_jobs.py
+317 -12
View File
@@ -29,6 +29,7 @@ except ImportError: # discovered with tests/ as the top-level directory
from PIL import Image # noqa: E402
import py.api.images as images_module # noqa: E402
from py.api.thumbnail_jobs import thumbnail_options # noqa: E402
from py.config import PromptManagerConfig # noqa: E402
PLAN_URL = "/prompt_manager/images/thumbnails/plan"
@@ -82,6 +83,45 @@ class ThumbnailJobCase(ThumbnailTestCase):
loop = asyncio.get_running_loop()
self.assertTrue(await loop.run_in_executor(None, event.wait, 10))
@staticmethod
def _watch_subscribers(job):
"""(subscribed, left): threading events for a stream joining or leaving."""
subscribed, left = threading.Event(), threading.Event()
class Watched(set):
def add(self, item):
super().add(item)
subscribed.set()
def discard(self, item):
super().discard(item)
left.set()
job._subscribers = Watched(job._subscribers)
return subscribed, left
def _status_lines(self):
"""Every status line any job is given, in order."""
from py.api.jobs import Job
lines = []
original = Job.update
def update(job, **kwargs):
if kwargs.get("status_line") is not None:
lines.append(kwargs["status_line"])
return original(job, **kwargs)
patcher = patch.object(Job, "update", update)
patcher.start()
self.addCleanup(patcher.stop)
return lines
def _one_worker(self):
original = PromptManagerConfig.WORKER_THREADS
PromptManagerConfig.WORKER_THREADS = 1 # batches of 4
self.addCleanup(setattr, PromptManagerConfig, "WORKER_THREADS", original)
def _run_scan(self):
gate = asyncio.Event()
@@ -399,9 +439,201 @@ class TestThumbnailJob(ThumbnailJobCase):
release.set()
await self._finished(started["job"]["id"])
async def test_every_exit_sets_a_truthful_final_status_line(self):
"""I3: finished, nothing to do, cancelled and failed each say what happened."""
lines = self._status_lines()
make_png(self.output_dir / "a.png")
(self.output_dir / "bad.png").write_bytes(b"\x89PNG nope")
_, body = await self._start()
job = await self._finished(body["job"]["id"])
self.assertEqual(job.status_line, "1 preview written · 1 error")
# Every preview fresh: the job finishes with nothing to do.
source_time = (self.output_dir / "a.png").stat().st_mtime
bad_thumb = make_png(self._thumb_path("bad"))
for thumb in (self._thumb_path("a"), bad_thumb):
os.utime(thumb, (source_time + 10, source_time + 10))
os.utime(self.output_dir / "bad.png", (source_time, source_time))
_, body = await self._start()
job = await self._finished(body["job"]["id"])
self.assertEqual((job.state, job.total), ("finished", 0))
self.assertEqual(
job.status_line, "No previews to generate · 2 already up to date"
)
self.assertNotIn("Processing file 0 of 0", lines)
async def test_an_empty_library_says_there_is_nothing_to_preview(self):
lines = self._status_lines()
_, body = await self._start()
job = await self._finished(body["job"]["id"])
self.assertEqual(
(job.state, job.status_line), ("finished", "No images or videos to preview")
)
self.assertNotIn("Processing file 0 of 0", lines)
async def test_a_cancelled_job_says_what_it_kept(self):
for i in range(6):
make_png(self.output_dir / f"p{i}.png")
self._one_worker()
real_write = images_module._write_image_thumbnail
job_ids = []
written = []
def counting(src, dst, *args, **kwargs):
real_write(src, dst, *args, **kwargs)
written.append(dst)
if len(written) == 1:
self.api._jobs.cancel(job_ids[0])
with patch.object(images_module, "_write_image_thumbnail", counting):
_, body = await self._start()
job_ids.append(body["job"]["id"])
job = await self._finished(body["job"]["id"])
self.assertEqual(job.state, "cancelled")
self.assertEqual(job.status_line, "Cancelled · 1 preview written")
async def test_an_internal_error_says_the_job_failed(self):
def broken(*_args, **_kwargs):
raise RuntimeError("boom")
self.api._plan_gallery_thumbnails = broken
_, body = await self._start()
job = await self._finished(body["job"]["id"])
self.assertEqual(
(job.state, job.status_line),
("failed", "Thumbnail generation stopped on an internal error."),
)
async def test_cancel_during_planning_stops_the_walk_and_starts_nothing(self):
"""M2: the walk checks Cancel per file; no ``start`` frame is published."""
for i in range(10):
make_png(self.output_dir / f"p{i}.png")
entered, release = threading.Event(), threading.Event()
self.addCleanup(release.set)
checked = []
real_fresh = images_module._thumbnail_is_fresh
def gated_fresh(src, dst):
checked.append(src.name)
if len(checked) == 1:
entered.set()
release.wait(10)
return real_fresh(src, dst)
with patch.object(images_module, "_thumbnail_is_fresh", gated_fresh):
stream = asyncio.ensure_future(self.client.post(self.PROGRESS_URL))
await self._wait(entered)
job = self.api._jobs.active("thumbnails")
self.assertTrue(self.api._jobs.cancel(job.id))
release.set()
events = parse_sse(await (await stream).text())
await asyncio.wait_for(job.task, 10)
self.assertEqual(checked, ["p0.png"]) # the walk stopped at the next file
self.assertEqual([name for name, _ in events], ["job", "status", "error"])
self.assertEqual(job.state, "cancelled")
self.assertEqual(job.status_line, "Cancelled before any preview was written")
self.assertEqual(job.total, 0)
self.assertFalse((self.output_dir / "thumbnails").exists())
async def test_a_locked_preview_is_a_readable_coded_error(self):
"""M6: Windows refuses to replace a preview another process has open."""
make_png(self.output_dir / "a.png")
locked = PermissionError(13, "Access is denied")
locked.winerror = 5
def refuse(*_args, **_kwargs):
raise locked
with patch.object(images_module, "_write_image_thumbnail", refuse):
_, body = await self._start(mode="rebuild")
job = await self._finished(body["job"]["id"])
self.assertEqual(job.counters["errors"], 1)
[row] = job.outcomes["errors"]
self.assertEqual(row["detail"], images_module.PREVIEW_IN_USE)
self.assertNotIn("WinError", repr(job.detail()))
self.assertTrue(any(images_module.PREVIEW_IN_USE in line for line in job.log))
async def test_a_locked_video_preview_is_the_same_coded_error(self):
clip = self.output_dir / "clip.mp4"
clip.write_bytes(b"\x00")
locked = PermissionError(13, "Access is denied")
locked.winerror = 32
def refuse(*_args, **_kwargs):
raise locked
with (
patch.dict(sys.modules, {"cv2": None}),
patch("subprocess.run", side_effect=FileNotFoundError("no ffmpeg")),
patch.object(images_module, "_atomic_save", refuse),
):
_, body = await self._start()
job = await self._finished(body["job"]["id"])
[row] = job.outcomes["errors"]
self.assertEqual(row["detail"], images_module.PREVIEW_IN_USE)
async def test_a_permission_error_off_windows_names_the_permission(self):
make_png(self.output_dir / "a.png")
def refuse(*_args, **_kwargs):
raise PermissionError(13, "Permission denied")
with patch.object(images_module, "_write_image_thumbnail", refuse):
_, body = await self._start()
job = await self._finished(body["job"]["id"])
[row] = job.outcomes["errors"]
self.assertEqual(row["detail"], images_module.PREVIEW_NOT_WRITABLE)
async def test_ffmpeg_fallback_keeps_quality_and_aspect_ratio(self):
"""M1: without OpenCV, ffmpeg keeps the chosen quality and the aspect ratio."""
(self.output_dir / "clip.mp4").write_bytes(b"\x00")
commands = []
def fake_ffmpeg(cmd, **_kwargs):
commands.append(cmd)
Path(cmd[-1]).write_bytes(b"jpeg")
return type("Done", (), {"returncode": 0, "stderr": ""})()
for quality, qscale in (("high", "2"), ("medium", "4"), ("low", "8")):
with (
patch.dict(sys.modules, {"cv2": None}),
patch("subprocess.run", side_effect=fake_ffmpeg),
):
_, body = await self._start(mode="rebuild", size=640, quality=quality)
job = await self._finished(body["job"]["id"])
self.assertEqual(job.counters["generated"], 1, quality)
cmd = commands[-1]
self.assertNotIn("-s", cmd)
self.assertEqual(
cmd[cmd.index("-vf") + 1],
"scale=640:640:force_original_aspect_ratio=decrease",
)
self.assertEqual(cmd[cmd.index("-q:v") + 1], qscale, quality)
async def test_the_placeholder_uses_the_chosen_quality(self):
(self.output_dir / "clip.mp4").write_bytes(b"\x00")
saved = []
real_save = images_module._atomic_save
def recording(img, dst, fmt=None, **kwargs):
saved.append(kwargs)
return real_save(img, dst, fmt, **kwargs)
with (
patch.dict(sys.modules, {"cv2": None}),
patch("subprocess.run", side_effect=FileNotFoundError("no ffmpeg")),
patch.object(images_module, "_atomic_save", recording),
):
_, body = await self._start(quality="low")
await self._finished(body["job"]["id"])
self.assertEqual(saved, [images_module.THUMBNAIL_ENCODINGS["low"]])
# The legacy stream's frames and their keys, as the gallery page reads them.
LEGACY_KEYS = {
"job": {"id", "started"},
"status": {"phase", "message"},
"start": {
"total_images",
@@ -450,14 +682,14 @@ class TestLegacyStreamOnTheJob(ThumbnailJobCase):
self.assertEqual(
[name for name, _ in events],
["status", "start", "progress", "progress", "complete"],
["job", "status", "start", "progress", "progress", "complete"],
)
for name, data in events:
self.assertEqual(set(data), LEGACY_KEYS[name], name)
complete = events[-1][1]
self.assertEqual((complete["count"], complete["total_images"]), (2, 2))
self.assertEqual(
events[0][1],
events[1][1],
{"phase": "scanning", "message": "Scanning for images and videos..."},
)
with Image.open(self._thumb_path("a")) as thumb:
@@ -468,6 +700,8 @@ class TestLegacyStreamOnTheJob(ThumbnailJobCase):
("thumbnails", "Thumbnail generation", "finished", True),
)
self.assertEqual(job.counters["generated"], 2)
# I1: the first frame names the job this stream started, for its Cancel.
self.assertEqual(events[0][1], {"id": job.id, "started": True})
async def test_legacy_file_errors_still_stream(self):
make_png(self.output_dir / "good.png")
@@ -482,21 +716,48 @@ class TestLegacyStreamOnTheJob(ThumbnailJobCase):
)
async def test_thumbnail_job_survives_client_disconnect(self):
for i in range(3):
"""I2: the stream sees its client go after batch 1; every batch still runs.
One worker thread makes batches of 4, so 9 files take three batches. The
first file waits until the client is gone; the second batch waits until
the stream has seen it go (its subscription dropped) and records whether
the job was still running then. A stream that stopped the work on
disconnect would leave 4 previews; the job writes all 9.
"""
for i in range(9):
make_png(self.output_dir / f"p{i}.png")
entered, release = self._gate_generation()
self._one_worker()
first, closed = threading.Event(), threading.Event()
state = {"calls": 0, "left": None, "running_when_left": None}
original = self.api._generate_one
def gated(*args, **kwargs):
state["calls"] += 1
if state["calls"] == 1:
first.set()
closed.wait(10)
elif state["calls"] == 5: # the second batch
state["running_when_left"] = state["left"].wait(10)
return original(*args, **kwargs)
self.api._generate_one = gated
self.addCleanup(closed.set)
resp = await self.client.post(self.PROGRESS_URL)
self.assertEqual(resp.status, 200)
await self._wait(entered)
await self._wait(first)
job = self.api._jobs.active("thumbnails")
_subscribed, state["left"] = self._watch_subscribers(job)
resp.close() # the gallery tab went away
release.set()
closed.set()
await self._wait(state["left"]) # the server saw the client go
self.assertEqual(job.subscriber_count, 0)
await asyncio.wait_for(job.task, 10)
self.assertEqual(job.counters["generated"], 9)
self.assertEqual(job.state, "finished")
self.assertEqual(job.counters["generated"], 3)
for i in range(3):
self.assertTrue(state["running_when_left"])
for i in range(9):
self.assertTrue(self._thumb_path(f"p{i}").is_file())
async def test_legacy_stream_attaches_to_a_running_job(self):
@@ -505,15 +766,16 @@ class TestLegacyStreamOnTheJob(ThumbnailJobCase):
_, started = await self._start(mode="missing", size=256)
await self._wait(entered)
attached = asyncio.ensure_future(self.client.post(self.PROGRESS_URL))
await asyncio.sleep(0)
job = self.api._jobs.get(started["job"]["id"])
while job.subscriber_count == 0:
await asyncio.sleep(0.01)
subscribed, _left = self._watch_subscribers(job)
attached = asyncio.ensure_future(self.client.post(self.PROGRESS_URL))
await self._wait(subscribed) # M4: an event with a ceiling, never a poll
release.set()
events = parse_sse(await (await attached).text())
self.assertEqual(events[-1][0], "complete")
# I1: the stream names the job it did not start: the gallery never cancels it.
self.assertEqual(events[0], ("job", {"id": job.id, "started": False}))
# One job: the stream followed the running one with its own options.
[listed] = self.api._jobs.list()
self.assertIs(listed, job)
@@ -536,6 +798,49 @@ class TestLegacyStreamOnTheJob(ThumbnailJobCase):
await asyncio.wait_for(job.task, 10)
self.assertEqual(job.state, "cancelled")
async def test_a_job_started_while_the_stream_prepares_is_attached(self):
"""M3: a job that starts during ``prepare`` is followed, not refused."""
from aiohttp import web
make_png(self.output_dir / "a.png")
entered, release = self._gate_generation()
real_prepare = web.StreamResponse.prepare
raced = []
async def prepare(response, request):
if not raced and type(response) is web.StreamResponse:
options, _ = thumbnail_options({"size": 256})
raced.append(self.api._start_thumbnail_job(options))
return await real_prepare(response, request)
with patch.object(web.StreamResponse, "prepare", prepare):
stream = asyncio.ensure_future(self.client.post(self.PROGRESS_URL))
await self._wait(entered)
release.set()
events = parse_sse(await (await stream).text())
[job] = raced
self.assertEqual(events[0], ("job", {"id": job.id, "started": False}))
self.assertEqual(events[-1][0], "complete")
self.assertEqual(self.api._jobs.list(), [job])
async def test_a_refused_stream_names_no_job(self):
make_png(self.output_dir / "a.png")
self._run_scan()
events = parse_sse(await (await self.client.post(self.PROGRESS_URL)).text())
self.assertEqual(
events,
[
(
"error",
{
"error": "Scan for metadata already running",
"message": "Scan for metadata already running",
},
)
],
)
if __name__ == "__main__":
unittest.main()