From 02be260a14fd6029b1afc1fe08e65247f52d7b68 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sun, 25 Oct 2026 21:11:31 -0700 Subject: [PATCH] fix(images): drop listings on the loop when invalidated from a worker thread (#348) A settings save (gallery roots change) and the browser-test seeds call invalidate_gallery_cache() from the executor. Listing records are only touched on the event loop, so an off-loop call now hands the drop to the loop and waits for it, keeping the call synchronous for its callers. The shared-Rescan test waits for the later Rescans to take their walk number instead of sleeping. --- py/api/__init__.py | 37 ++++++++++++++++++++++++++++++++++++- py/api/images.py | 1 + tests/test_api_images.py | 35 ++++++++++++++++++++++++++++++++++- 3 files changed, 71 insertions(+), 2 deletions(-) diff --git a/py/api/__init__.py b/py/api/__init__.py index aa1d5456..2d137763 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -31,6 +31,7 @@ import math import ntpath import os import re +import threading from pathlib import Path from aiohttp import web @@ -88,6 +89,18 @@ def _get_project_root(): return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +# How long an executor thread waits for the loop to drop the listings. +_OFF_LOOP_DROP_WAIT = 30 + + +def _running_loop(): + """The event loop running in this thread, or None.""" + try: + return asyncio.get_running_loop() + except RuntimeError: + return None + + def _public_path(path) -> str: """Render a server path for an API response without revealing the layout. @@ -349,6 +362,8 @@ class PromptManagerAPI( # background walks refreshing them (py/api/gallery_listing.py, #348). self._root_listings: dict[str, RootListing] = {} self._listing_tasks: set = set() + # The loop the walks run on, set by the first walk. + self._listing_loop = None # Workspace listing values and sort orders derived from a cached scan # (py/api/images.py _workspace_listed/_workspace_orders). self._workspace_listed_cache = {} @@ -484,7 +499,27 @@ class PromptManagerAPI( the walk replaces them anyway, and so is the favorites lookup, which holds the old list. """ - for record in self._root_listings.values(): + loop = self._listing_loop + if loop is not None and loop.is_running() and _running_loop() is not loop: + # Called from an executor thread (a settings save, a test seed): + # listing records are touched only on the loop, and the caller + # expects the drop done when this returns. + dropped = threading.Event() + + def drop(): + try: + self._drop_listings() + finally: + dropped.set() + + loop.call_soon_threadsafe(drop) + if not dropped.wait(_OFF_LOOP_DROP_WAIT): + self.logger.warning("The Images listing drop is still queued") + return + self._drop_listings() + + def _drop_listings(self): + for record in list(self._root_listings.values()): record.drop() self._workspace_order_cache = {} self._favorite_listing_cache = {} diff --git a/py/api/images.py b/py/api/images.py index f98e9a5b..519f6850 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -1356,6 +1356,7 @@ class ImageRoutesMixin: def _start_walk(self, output_path, record): """Start the root's one walk; returns its flight future (on the loop).""" loop = asyncio.get_running_loop() + self._listing_loop = loop record.walk_seq += 1 flight = loop.create_future() record.flight = flight diff --git a/tests/test_api_images.py b/tests/test_api_images.py index 8ebffa96..8559d9f5 100644 --- a/tests/test_api_images.py +++ b/tests/test_api_images.py @@ -2483,6 +2483,20 @@ class TestSharedListing(ImageAPITestCase): async def test_rescans_clicked_during_a_walk_share_the_next_one(self): make_png(self.output_dir / "a.png") await self._json("GET", self.URL) + original = self.api._listing + rescans = [] + clicked = asyncio.Event() + + async def listing(root, *, fresh=False): + if fresh: + rescans.append(root) + if len(rescans) == 3: + clicked.set() + # Runs on in this step up to its first wait, so the Rescan's + # walk number is taken before the test resumes. + return await original(root, fresh=fresh) + + self.api._listing = listing gate = self.scan.hold_next() first = asyncio.ensure_future( self._json("POST", "/prompt_manager/gallery/rescan") @@ -2494,7 +2508,7 @@ class TestSharedListing(ImageAPITestCase): asyncio.ensure_future(self._json("POST", "/prompt_manager/gallery/rescan")) for _ in range(2) ] - await asyncio.sleep(0.05) + await asyncio.wait_for(clicked.wait(), 30) gate.set() results = await asyncio.gather(first, *later) self.assertEqual([status for status, _ in results], [200] * 3) @@ -2562,6 +2576,25 @@ class TestSharedListing(ImageAPITestCase): self.assertEqual((status, data["total"]), (200, 2)) self.assertEqual(self.scan.calls, 2) + async def test_an_invalidation_from_a_worker_thread_drops_on_the_loop(self): + make_png(self.output_dir / "a.png") + await self._json("GET", self.URL) + record = self._record() + loop_thread = threading.get_ident() + threads = [] + original = record.drop + + def drop(): + threads.append(threading.get_ident()) + original() + + record.drop = drop + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, self.api.invalidate_gallery_cache) + # Dropped on the loop, before invalidate_gallery_cache returned. + self.assertEqual(threads, [loop_thread]) + self.assertIsNone(record.files) + async def test_three_discarded_walks_are_a_500(self): loop = asyncio.get_running_loop() real = self.scan.real