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

fix(review): coalesce queue change events and answer queue thumbnails

Browse Source 
- 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.
This commit is contained in:
Vito Sansevero
2026-10-22 22:46:02 -07:00
parent 6fa88fb2f6
commit afe3733b69
5 changed files with 131 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/api/images.py
+21 -8
View File
@@ -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(
py/api/review.py
+7 -1
View File
@@ -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,
}
py/api/review_events.py
+19 -2
View File
@@ -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})")
tests/test_api_review.py
+25
View File
@@ -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"
tests/test_api_review_events.py
+59 -11
View File
@@ -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())