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

@@ -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 = {}

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

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