fix: thumbnail runs skip fresh thumbnails before the cap and report what remains
The per-run cap of 10,000 applied to the directory walk itself, so a library with more files than that never had thumbnails generated for the rest, and a run that found 9,987 existing thumbnails spent its whole budget on skipping them. - _plan_thumbnails walks the whole tree (bounded like the gallery listing), counts files whose thumbnail is already newer as skipped without queuing them, and caps only the pending files; the leftover count is returned - start and complete payloads carry skipped and remaining; total_images counts every file considered and the processing rate uses the files actually worked on - the gallery page shows a warning naming the pending count instead of a success toast when a run stopped at the cap
This commit is contained in:
+92
-22
@@ -157,27 +157,46 @@ def _thumbnail_progress_payload(index, total, stats, result, start):
|
||||
}
|
||||
|
||||
|
||||
def _thumbnail_complete_payload(total, stats, start):
|
||||
def _thumbnail_complete_payload(total, stats, start, remaining=0, processed=None):
|
||||
"""Final payload of a thumbnail run.
|
||||
|
||||
``total`` counts every media file considered (pending, already fresh and
|
||||
left over), ``processed`` the files this run actually worked on (defaults
|
||||
to ``total``) and ``remaining`` the pending files left for the next run
|
||||
because of MAX_THUMBNAIL_FILES.
|
||||
"""
|
||||
elapsed = _time.monotonic() - start
|
||||
processed = total if processed is None else processed
|
||||
errors = stats["errors"]
|
||||
message = (
|
||||
f"Generated {stats['generated']} new thumbnails, "
|
||||
f"skipped {stats['skipped']} existing"
|
||||
)
|
||||
if remaining:
|
||||
message += f", {remaining} more to generate on the next run"
|
||||
if errors:
|
||||
message += f" ({len(errors)} errors occurred)"
|
||||
return {
|
||||
"count": stats["generated"],
|
||||
"skipped": stats["skipped"],
|
||||
"total_images": total,
|
||||
"remaining": remaining,
|
||||
"errors": errors[:10],
|
||||
"error_count": len(errors),
|
||||
"elapsed_time": round(elapsed, 2),
|
||||
"processing_rate": round(total / elapsed if elapsed > 0 else 0, 2),
|
||||
"processing_rate": round(processed / elapsed if elapsed > 0 else 0, 2),
|
||||
"message": message,
|
||||
}
|
||||
|
||||
|
||||
def _thumbnail_is_fresh(src, dst):
|
||||
"""True when ``dst`` exists and is newer than ``src`` (blocking stat)."""
|
||||
try:
|
||||
return dst.stat().st_mtime > src.stat().st_mtime
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
# Caps on request-driven work (list sizes, scans, request bodies).
|
||||
MAX_GALLERY_FILES = 50_000
|
||||
MAX_BULK_LIMIT = 5_000
|
||||
@@ -772,25 +791,40 @@ class ImageRoutesMixin:
|
||||
def _generate_thumbnails_sync(self, output_path, thumbnails_dir, thumbnail_size):
|
||||
"""Blocking thumbnail generation loop (run in executor)."""
|
||||
start = _time.monotonic()
|
||||
targets = self._thumbnail_targets(output_path, thumbnails_dir)
|
||||
plan = self._plan_thumbnails(output_path, thumbnails_dir)
|
||||
targets = plan["targets"]
|
||||
stats = _new_thumbnail_stats()
|
||||
stats["skipped"] = plan["skipped"]
|
||||
for result in self._generate_batch(targets, thumbnail_size):
|
||||
_record_thumbnail_result(stats, result)
|
||||
payload = _thumbnail_complete_payload(len(targets), stats, start)
|
||||
payload = _thumbnail_complete_payload(
|
||||
len(targets) + plan["skipped"] + plan["remaining"],
|
||||
stats,
|
||||
start,
|
||||
remaining=plan["remaining"],
|
||||
processed=len(targets),
|
||||
)
|
||||
payload["success"] = True
|
||||
payload["thumbnails_path"] = _public_path(thumbnails_dir)
|
||||
self.logger.info(payload["message"])
|
||||
return payload
|
||||
|
||||
def _thumbnail_targets(self, output_path, thumbnails_dir):
|
||||
"""Scan *output_path* for media (blocking); returns (src, dst, is_video).
|
||||
def _plan_thumbnails(self, output_path, thumbnails_dir):
|
||||
"""Decide which media files need a thumbnail (blocking walk).
|
||||
|
||||
Bounded by _iter_media_files (no symlinks, MAX_SCAN_DEPTH) and
|
||||
MAX_THUMBNAIL_FILES per request.
|
||||
Returns ``{"targets": [(src, dst, is_video)], "skipped": n, "remaining": n}``.
|
||||
Files whose thumbnail is already newer than the source are counted as
|
||||
skipped and never queued, so a large library that is mostly done only
|
||||
pays for the files that changed. The walk is bounded like the gallery
|
||||
listing (MAX_SCAN_DEPTH, MAX_GALLERY_FILES, no symlinks) and at most
|
||||
MAX_THUMBNAIL_FILES pending files are queued per run; ``remaining``
|
||||
counts the pending files beyond that cap.
|
||||
"""
|
||||
output_path = Path(output_path)
|
||||
targets = []
|
||||
for src in _iter_media_files(output_path, MAX_THUMBNAIL_FILES):
|
||||
skipped = 0
|
||||
remaining = 0
|
||||
for src in _iter_media_files(output_path, MAX_GALLERY_FILES):
|
||||
suffix = src.suffix.lower()
|
||||
is_video = suffix in VIDEO_EXTENSIONS
|
||||
rel_no_ext = src.relative_to(output_path).with_suffix("")
|
||||
@@ -799,10 +833,23 @@ class ImageRoutesMixin:
|
||||
if not _is_within(dst, thumbnails_dir):
|
||||
self.logger.warning(f"Skipping thumbnail outside safe dir: {src.name}")
|
||||
continue
|
||||
if _thumbnail_is_fresh(src, dst):
|
||||
skipped += 1
|
||||
continue
|
||||
if len(targets) >= MAX_THUMBNAIL_FILES:
|
||||
remaining += 1
|
||||
continue
|
||||
targets.append((src, dst, is_video))
|
||||
if len(targets) >= MAX_THUMBNAIL_FILES:
|
||||
self.logger.warning(f"Thumbnail scan capped at {MAX_THUMBNAIL_FILES}")
|
||||
return targets
|
||||
if remaining:
|
||||
self.logger.warning(
|
||||
f"Thumbnail run capped at {MAX_THUMBNAIL_FILES} new files; "
|
||||
f"{remaining} more will be generated on the next run"
|
||||
)
|
||||
return {"targets": targets, "skipped": skipped, "remaining": remaining}
|
||||
|
||||
def _thumbnail_targets(self, output_path, thumbnails_dir):
|
||||
"""Pending (src, dst, is_video) triples; see _plan_thumbnails."""
|
||||
return self._plan_thumbnails(output_path, thumbnails_dir)["targets"]
|
||||
|
||||
def _generate_batch(self, targets, thumbnail_size):
|
||||
"""Create thumbnails for ``targets`` on the configured worker threads.
|
||||
@@ -873,10 +920,16 @@ class ImageRoutesMixin:
|
||||
"status",
|
||||
{"phase": "scanning", "message": "Scanning for images and videos..."},
|
||||
)
|
||||
targets = await self._run_in_executor(
|
||||
self._thumbnail_targets, output_path, output_path / "thumbnails"
|
||||
plan = await self._run_in_executor(
|
||||
self._plan_thumbnails, output_path, output_path / "thumbnails"
|
||||
)
|
||||
await self._stream_thumbnails(
|
||||
response,
|
||||
plan["targets"],
|
||||
thumbnail_size,
|
||||
skipped=plan["skipped"],
|
||||
remaining=plan["remaining"],
|
||||
)
|
||||
await self._stream_thumbnails(response, targets, thumbnail_size)
|
||||
self.invalidate_gallery_cache()
|
||||
except Exception:
|
||||
self.logger.exception("Thumbnail generation failed")
|
||||
@@ -890,27 +943,38 @@ class ImageRoutesMixin:
|
||||
)
|
||||
return response
|
||||
|
||||
async def _stream_thumbnails(self, response, targets, thumbnail_size):
|
||||
"""Process *targets* in worker-thread batches, emitting SSE per file."""
|
||||
async def _stream_thumbnails(
|
||||
self, response, targets, thumbnail_size, skipped=0, remaining=0
|
||||
):
|
||||
"""Process pending *targets* in worker-thread batches, emitting SSE per file.
|
||||
|
||||
``skipped`` files already had fresh thumbnails and ``remaining`` were
|
||||
left for the next run; both only feed the start/complete payloads.
|
||||
"""
|
||||
total = len(targets)
|
||||
video_count = sum(1 for _, _, is_video in targets if is_video)
|
||||
message = (
|
||||
f"Found {total - video_count} images and {video_count} videos to process"
|
||||
)
|
||||
if skipped:
|
||||
message += f" ({skipped} already have thumbnails)"
|
||||
connected = await self._emit_sse(
|
||||
response,
|
||||
"start",
|
||||
{
|
||||
"total_images": total,
|
||||
"skipped": skipped,
|
||||
"remaining": remaining,
|
||||
"phase": "processing",
|
||||
"image_count": total - video_count,
|
||||
"video_count": video_count,
|
||||
"message": (
|
||||
f"Found {total - video_count} images and "
|
||||
f"{video_count} videos to process"
|
||||
),
|
||||
"message": message,
|
||||
},
|
||||
)
|
||||
if not connected:
|
||||
return
|
||||
stats = _new_thumbnail_stats()
|
||||
stats["skipped"] = skipped
|
||||
start = _time.monotonic()
|
||||
batch_size = _thumbnail_batch_size(_worker_threads())
|
||||
index = 0
|
||||
@@ -937,7 +1001,13 @@ class ImageRoutesMixin:
|
||||
self.logger.info("Thumbnail client disconnected; stopping early")
|
||||
return
|
||||
await asyncio.sleep(0) # let other requests run between batches
|
||||
payload = _thumbnail_complete_payload(total, stats, start)
|
||||
payload = _thumbnail_complete_payload(
|
||||
total + skipped + remaining,
|
||||
stats,
|
||||
start,
|
||||
remaining=remaining,
|
||||
processed=total,
|
||||
)
|
||||
await self._emit_sse(response, "complete", payload)
|
||||
self.logger.info(payload["message"])
|
||||
|
||||
|
||||
@@ -471,7 +471,47 @@ class TestThumbnailProgressStream(ThumbnailTestCase):
|
||||
# PIL work ran in the executor in batches (never on the loop), plus the scan.
|
||||
self.assertGreaterEqual(calls.count("_generate_batch"), 1)
|
||||
self.assertNotIn("_generate_one", calls)
|
||||
self.assertIn("_thumbnail_targets", calls)
|
||||
self.assertIn("_plan_thumbnails", calls)
|
||||
|
||||
async def test_capped_run_reports_remaining_and_the_next_run_finishes(self):
|
||||
for name in ("a", "b", "c"):
|
||||
make_png(self.output_dir / f"{name}.png", size=(64, 64))
|
||||
|
||||
with patch.object(images_module, "MAX_THUMBNAIL_FILES", 2):
|
||||
first = parse_sse(
|
||||
await (await self.client.request("POST", self.PROGRESS_URL)).text()
|
||||
)
|
||||
second = parse_sse(
|
||||
await (await self.client.request("POST", self.PROGRESS_URL)).text()
|
||||
)
|
||||
|
||||
names = [e for e, _ in first]
|
||||
self.assertEqual(names.count("progress"), 2) # only pending files are work
|
||||
done = dict(first)["complete"]
|
||||
self.assertEqual(done["count"], 2)
|
||||
self.assertEqual(done["remaining"], 1)
|
||||
self.assertEqual(done["total_images"], 3)
|
||||
self.assertIn("1 more", done["message"])
|
||||
done = dict(second)["complete"]
|
||||
self.assertEqual(done["count"], 1)
|
||||
self.assertEqual(done["skipped"], 2)
|
||||
self.assertEqual(done["remaining"], 0)
|
||||
self.assertEqual(done["total_images"], 3)
|
||||
self.assertNotIn("more", done["message"])
|
||||
|
||||
async def test_blocking_endpoint_reports_remaining(self):
|
||||
for name in ("a", "b", "c"):
|
||||
make_png(self.output_dir / f"{name}.png", size=(64, 64))
|
||||
|
||||
with patch.object(images_module, "MAX_THUMBNAIL_FILES", 2):
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/images/generate-thumbnails", json={}
|
||||
)
|
||||
data = await resp.json()
|
||||
|
||||
self.assertEqual(data["count"], 2)
|
||||
self.assertEqual(data["remaining"], 1)
|
||||
self.assertEqual(data["total_images"], 3)
|
||||
|
||||
async def test_corrupt_png_is_reported_and_loop_continues(self):
|
||||
make_png(self.output_dir / "good1.png")
|
||||
@@ -791,6 +831,36 @@ class TestMediaScanCaps(ImageAPITestCase):
|
||||
|
||||
self.assertEqual(found, ["keep.png"])
|
||||
|
||||
def test_thumbnail_plan_skips_fresh_thumbnails_before_applying_the_cap(self):
|
||||
for name in ("a", "b", "c"):
|
||||
make_png(self.output_dir / f"{name}.png")
|
||||
self.api._generate_thumbnails_sync(
|
||||
self.output_dir, self.output_dir / "thumbnails", (64, 64)
|
||||
)
|
||||
for name in ("d", "e", "f"):
|
||||
make_png(self.output_dir / f"{name}.png")
|
||||
|
||||
with patch.object(images_module, "MAX_THUMBNAIL_FILES", 2):
|
||||
plan = self.api._plan_thumbnails(
|
||||
self.output_dir, self.output_dir / "thumbnails"
|
||||
)
|
||||
|
||||
self.assertEqual(plan["skipped"], 3)
|
||||
self.assertEqual([t[0].name for t in plan["targets"]], ["d.png", "e.png"])
|
||||
self.assertEqual(plan["remaining"], 1)
|
||||
|
||||
def test_thumbnail_walk_is_bounded_like_the_gallery_walk(self):
|
||||
for name in ("a", "b", "c"):
|
||||
make_png(self.output_dir / f"{name}.png")
|
||||
|
||||
with patch.object(images_module, "MAX_GALLERY_FILES", 2):
|
||||
plan = self.api._plan_thumbnails(
|
||||
self.output_dir, self.output_dir / "thumbnails"
|
||||
)
|
||||
|
||||
self.assertEqual(len(plan["targets"]), 2)
|
||||
self.assertEqual(plan["remaining"], 0)
|
||||
|
||||
def test_thumbnail_targets_honour_file_cap(self):
|
||||
for name in ("a", "b", "c"):
|
||||
make_png(self.output_dir / f"{name}.png")
|
||||
@@ -1247,7 +1317,7 @@ class TestThumbnailUnits(ImageRouteCoverageCase):
|
||||
response.write.assert_awaited_once()
|
||||
|
||||
async def test_progress_stream_reports_internal_failure(self):
|
||||
self.api._thumbnail_targets = _raise
|
||||
self.api._plan_thumbnails = _raise
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/images/generate-thumbnails/progress"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user