diff --git a/TESTING.md b/TESTING.md index 6868a096..7881dc3b 100644 --- a/TESTING.md +++ b/TESTING.md @@ -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 `` (no `value`) and a pill without a percentage. Only a + has an indeterminate `` (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 diff --git a/py/api/images.py b/py/api/images.py index 2cfd40df..4ec86a13 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -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}" diff --git a/py/api/jobs.py b/py/api/jobs.py index 9aa7ec2a..aec92eb9 100644 --- a/py/api/jobs.py +++ b/py/api/jobs.py @@ -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 diff --git a/py/api/review_recording.py b/py/api/review_recording.py index 26c2de2f..7cb60c6d 100644 --- a/py/api/review_recording.py +++ b/py/api/review_recording.py @@ -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 " diff --git a/py/api/thumbnail_jobs.py b/py/api/thumbnail_jobs.py index b75ab8b3..9c0d6b63 100644 --- a/py/api/thumbnail_jobs.py +++ b/py/api/thumbnail_jobs.py @@ -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 diff --git a/tests/test_jobs.py b/tests/test_jobs.py index 59ba655e..9176ba6a 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -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() diff --git a/tests/test_review.py b/tests/test_review.py index 70776c02..ad41001e 100644 --- a/tests/test_review.py +++ b/tests/test_review.py @@ -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) diff --git a/tests/test_thumbnail_jobs.py b/tests/test_thumbnail_jobs.py index a602b6a3..9b9d5ba2 100644 --- a/tests/test_thumbnail_jobs.py +++ b/tests/test_thumbnail_jobs.py @@ -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()