From afe3733b69422665582a1e10504fba27ffab51c0 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Thu, 22 Oct 2026 22:46:02 -0700 Subject: [PATCH] fix(review): coalesce queue change events and answer queue thumbnails - The change stream sends the first notice at once and collapses the notices of the next 0.4 s into one trailing change, so a busy scan costs an open page one re-read per 0.4 s at most (COALESCE_SECONDS). - Queue items answer thumbnail_url, the same URL the Images listing builds (one shared helper, media_thumbnail_url), or null without a thumbnail. - The "duplicate and refused decision send nothing" test spies on notify_review_change instead of reading the next event. --- py/api/images.py | 29 ++++++++++---- py/api/review.py | 8 +++- py/api/review_events.py | 21 +++++++++- tests/test_api_review.py | 25 ++++++++++++ tests/test_api_review_events.py | 70 +++++++++++++++++++++++++++------ 5 files changed, 131 insertions(+), 22 deletions(-) diff --git a/py/api/images.py b/py/api/images.py index 6edfdc1..d2d67f0 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -663,6 +663,25 @@ def _iter_media_files(output_path, max_files): yield root_path / name +def media_thumbnail_url(output_path, rel_path, root_index, *, root_scope=None): + """The Images listing's thumbnail URL for *rel_path* under *output_path*, + or None when no thumbnail exists (one blocking stat). The Needs review + queue cards use the same URL (#249 F, M10).""" + rel_path = Path(rel_path) + extension = rel_path.suffix.lower() + thumb_rel = ( + f"thumbnails/{rel_path.with_suffix('').as_posix()}_thumb" + f"{'.jpg' if extension in VIDEO_EXTENSIONS else extension}" + ) + thumbnail_url = thumbnail_url_for(thumb_rel, Path(output_path) / thumb_rel) + if thumbnail_url is None: + return None + thumbnail_url += f"&root={root_index}" + if root_scope is not None: + thumbnail_url += f"&root_scope={root_scope}" + return thumbnail_url + + def _output_image_entry( media_path, output_path, root_index, public_root=None, *, root_scope=None ): @@ -678,19 +697,13 @@ def _output_image_entry( extension = media_path.suffix.lower() is_video = extension in VIDEO_EXTENSIONS - thumb_rel = ( - f"thumbnails/{rel_path.with_suffix('').as_posix()}_thumb" - f"{'.jpg' if is_video else extension}" + thumbnail_url = media_thumbnail_url( + output_path, rel_path, root_index, root_scope=root_scope ) - thumbnail_url = thumbnail_url_for(thumb_rel, output_path / thumb_rel) - if thumbnail_url is not None: - thumbnail_url += f"&root={root_index}" encoded_path = urllib.parse.quote(rel_path.as_posix(), safe="/") url = f"/prompt_manager/images/serve/{encoded_path}?root={root_index}" if root_scope is not None: url += f"&root_scope={root_scope}" - if thumbnail_url is not None: - thumbnail_url += f"&root_scope={root_scope}" entry = { "id": hashlib.sha1( diff --git a/py/api/review.py b/py/api/review.py index bdbcbae..6fb5f4f 100644 --- a/py/api/review.py +++ b/py/api/review.py @@ -29,6 +29,7 @@ import urllib.parse from aiohttp import web from .annotations import _HEX64, _TOKEN, _UUID, _capture_error, _error +from .images import media_thumbnail_url from .inspection_service import VIDEO_EXTENSIONS from .prompts import bad_request @@ -255,7 +256,7 @@ class ReviewRoutesMixin: def _review_view(self, row: dict, roots, real_roots) -> dict: """The public ReviewItem of one stored row (no path, no locator).""" found = format_review_target(row["spelled_path"], row["locator"], real_roots) - target = url = None + target = url = thumbnail_url = None if found is not None: index, relative = found target = { @@ -269,6 +270,10 @@ class ReviewRoutesMixin: f"{urllib.parse.quote(relative, safe='/')}?root={index}" f"&root_scope={target['root_scope']}" ) + # The Images listing's thumbnail, when one exists (one stat). + thumbnail_url = media_thumbnail_url( + roots[index], relative, index, root_scope=target["root_scope"] + ) filename = os.path.basename(row["spelled_path"]) return { "id": row["id"], @@ -281,6 +286,7 @@ class ReviewRoutesMixin: "candidate_count": row["candidate_count"], "limited": row["limited"], "url": url, + "thumbnail_url": thumbnail_url, "is_video": os.path.splitext(filename)[1].lower() in VIDEO_EXTENSIONS, } diff --git a/py/api/review_events.py b/py/api/review_events.py index 91a9d2a..b59b611 100644 --- a/py/api/review_events.py +++ b/py/api/review_events.py @@ -4,7 +4,9 @@ open Needs review page: ``event: ready`` when it opens, then ``event: change`` after each committed queue change (an item the scan or the monitor recorded, a decision from any tab, a spelling refresh) and after a -restore. Notices that arrive while one is being written collapse into one. +restore. The first notice is sent at once; notices in the next +``COALESCE_SECONDS`` collapse into one trailing ``change`` (M9: a busy scan +records file after file, and each event costs every open page a re-read). The page re-reads its queue on each ``change``; nothing polls. ``change`` carries no data (``{}``). ``ready`` carries the queue's version ``{library_epoch, review_revision}``, read after the stream subscribed, the @@ -32,6 +34,8 @@ except ImportError: EVENTS_URL = "/prompt_manager/review/events" HEARTBEAT_SECONDS = 25 +# After a ``change``, the next one waits this long, collapsing the notices in between. +COALESCE_SECONDS = 0.4 _CHANGE = b"event: change\ndata: {}\n\n" _KEEPALIVE = b": keepalive\n\n" @@ -77,6 +81,14 @@ class _Stream: return "change" return "idle" + async def closed_within(self, timeout: float) -> bool: + """Wait up to ``timeout``; true when the stream closed meanwhile.""" + try: + await asyncio.wait_for(self.closed.wait(), timeout) + except asyncio.TimeoutError: + return False + return True + class ReviewEventsMixin: """The Needs review change stream.""" @@ -143,7 +155,12 @@ class ReviewEventsMixin: outcome = await stream.next(HEARTBEAT_SECONDS) if outcome == "closed": break - await response.write(_CHANGE if outcome == "change" else _KEEPALIVE) + if outcome == "idle": + await response.write(_KEEPALIVE) + continue + await response.write(_CHANGE) + if await stream.closed_within(COALESCE_SECONDS): + break await response.write_eof() except (ConnectionError, RuntimeError) as exc: # the tab went away self.logger.debug(f"Needs review stream closed ({exc})") diff --git a/tests/test_api_review.py b/tests/test_api_review.py index fc82002..1075f0b 100644 --- a/tests/test_api_review.py +++ b/tests/test_api_review.py @@ -179,6 +179,7 @@ class TestQueueReads(ReviewRoutesTestCase): "candidate_count", "limited", "url", + "thumbnail_url", "is_video", }, ) @@ -374,10 +375,34 @@ class TestQueueReads(ReviewRoutesTestCase): (True, None, "two.png"), ) self.assertEqual((item["url"], item["is_video"]), (None, False)) + self.assertIsNone(item["thumbnail_url"]) detail = await self._detail(item["id"]) self.assertIs(detail["unavailable"], True) self._no_paths() + async def test_queue_cards_use_the_images_listing_thumbnail(self): + # M10: a known thumbnail is the card's media, as on the Images page; + # without one the card falls back to the original. + self._require_routes() + from py.api.images import _output_image_entry + + path = self._ambiguous("sub/f0.png") + self._record(path) + item = (await self._list())["items"][0] + self.assertIsNone(item["thumbnail_url"]) + thumb = self.output_dir / "thumbnails" / "sub" / "f0_thumb.png" + thumb.parent.mkdir(parents=True) + write_png(thumb, prompt=ambiguous_graph("cat", "dog"), color=(1, 2, 3)) + item = (await self._list())["items"][0] + scope = item["target"]["root_scope"] + listed = _output_image_entry(path, self.output_dir, 0, root_scope=scope) + self.assertIsNotNone(listed["thumbnail_url"]) + self.assertEqual(item["thumbnail_url"], listed["thumbnail_url"]) + self.assertEqual(item["url"], listed["url"]) + detail = await self._detail(item["id"]) + self.assertEqual(detail["thumbnail_url"], listed["thumbnail_url"]) + self._no_paths() + def test_target_formatting_keeps_the_recorded_spelling_on_windows(self): self._require_routes() spelled = "C:\\Out\\Sub\\Image_0001.PNG" diff --git a/tests/test_api_review_events.py b/tests/test_api_review_events.py index dba9818..22544b2 100644 --- a/tests/test_api_review_events.py +++ b/tests/test_api_review_events.py @@ -96,22 +96,70 @@ class TestReviewEvents(ReviewEventsTestCase): self.assertEqual(await self._event(stream), "change") async def test_duplicate_recording_and_refused_decision_send_nothing(self): + # A spy on the notice itself: a wrong notice would otherwise pass as + # the next real change's event. self._require_events() + from unittest.mock import patch + + from database import review as review_db + from database import review_decisions + + calls = [] + original = review_changes.notify_review_change + + def spy(db_path): + calls.append(db_path) + original(db_path) + path = self._ambiguous("two.png") self._record(path) - stream = await self._stream() - # The same bytes again: nothing changes. - self.assertEqual(await self._record_in_thread(path), "duplicate") - item, detail, observed = await self._open() - # A stale revision: refused with zero writes. - body = {**self._body(detail, observed, "unresolved"), "expected_revision": 9} - status, data = await self._decide(item["id"], body) - self.assertEqual(status, 409, data) - # The next notice is the next real change, so nothing came before it. - self._record(self._ambiguous("other.png", color=(0, 0, 9))) - self.assertEqual(await self._event(stream), "change") + with ( + patch.object(review_db, "notify_review_change", spy), + patch.object(review_decisions, "notify_review_change", spy), + ): + # The same bytes again: nothing changes. + self.assertEqual(await self._record_in_thread(path), "duplicate") + self.assertEqual(calls, [], "a duplicate recording sent a notice") + item, detail, observed = await self._open() + # A stale revision: refused with zero writes. + body = { + **self._body(detail, observed, "unresolved"), + "expected_revision": 9, + } + status, data = await self._decide(item["id"], body) + self.assertEqual(status, 409, data) + self.assertEqual(calls, [], "a refused decision sent a notice") + self._record(self._ambiguous("other.png", color=(0, 0, 9))) + self.assertEqual(len(calls), 1, "a real recording sends one notice") self.assertEqual(len(self._items()), 2) + async def test_a_burst_of_changes_coalesces_into_one_more_event(self): + # M9: a busy scan records file after file; an open page hears the + # first at once and the rest as one trailing ``change``. + self._require_events() + from unittest.mock import patch + + path = self.api.db.model.db_path + with ( + patch.object(review_events, "COALESCE_SECONDS", 0.3), + patch.object(review_events, "HEARTBEAT_SECONDS", 0.6), + ): + stream = await self._stream() + review_changes.notify_review_change(path) + self.assertEqual(await self._event(stream), "change") + for _ in range(5): + review_changes.notify_review_change(path) + await asyncio.sleep(0.02) + # Every change event up to the next keepalive: exactly one. + changes = 0 + while True: + line = await asyncio.wait_for(stream.content.readline(), READ_SECONDS) + self.assertTrue(line, "the stream ended") + if line == b": keepalive\n": + break + changes += line == b"event: change\n" + self.assertEqual(changes, 1) + async def test_decision_from_any_tab_sends_a_change(self): self._require_events() self._record(self._ambiguous())