diff --git a/py/api/system.py b/py/api/system.py index 2362b32..605334e 100644 --- a/py/api/system.py +++ b/py/api/system.py @@ -41,6 +41,20 @@ _HEALTH_STATUS_LINES = { _StoredReport = namedtuple("_StoredReport", "generation report") +async def _settled(*reads): + """Every read's result once all of them ended; the first error raises. + + ``asyncio.gather`` answers at the first error while the other executor + reads still run, so a route could answer (and its caller replace or + delete the database) with a sibling read still inside SQLite. + """ + results = await asyncio.gather(*reads, return_exceptions=True) + for result in results: + if isinstance(result, BaseException): + raise result + return results + + def _failure(message, status=500, code=None): return web.json_response( {"success": False, "code": code, "error": message}, status=status @@ -155,7 +169,7 @@ class SystemRoutesMixin: from ..config import GalleryConfig try: - (output_dirs, auto_detected), counts = await asyncio.gather( + (output_dirs, auto_detected), counts = await _settled( self._run_in_executor(self._output_dirs_with_origin), self._run_in_executor(self.db.library_counts), ) @@ -189,7 +203,7 @@ class SystemRoutesMixin: async def get_system_statistics(self, request): try: - by_category, totals, counts = await asyncio.gather( + by_category, totals, counts = await _settled( self._run_in_executor(self.db.usage_by_category), self._run_in_executor(self.db.get_statistics), self._run_in_executor(self.db.library_counts), diff --git a/tests/test_system_api.py b/tests/test_system_api.py index 7af1955..3d649fe 100644 --- a/tests/test_system_api.py +++ b/tests/test_system_api.py @@ -416,6 +416,54 @@ class TestStatisticsRoute(SystemAPITestCase): ) +class TestReadsSettleBeforeTheAnswer(SystemAPITestCase): + """A failed read answers only after its sibling reads ended. + + ``asyncio.gather`` does not stop executor work, so an early answer left a + sibling read on the database while the caller went on (in the suite, the + test's teardown deleted the database under it: a segfault on py3.10). + """ + + def _slow(self, target, name): + real = getattr(target, name) + finished = threading.Event() + + def slow(*args, **kwargs): + threading.Event().wait(0.3) # still reading when the sibling fails + try: + return real(*args, **kwargs) + finally: + finished.set() + + return patch.object(target, name, side_effect=slow), finished + + async def _failing(self, url, failing, slow_target, slow_name): + slow, finished = self._slow(slow_target, slow_name) + with ( + slow, + patch.object(self.api.db, failing, side_effect=RuntimeError("secret")), + self.assertLogs("prompt_manager.api", "ERROR"), + ): + await self.json("GET", url, 500) + self.assertTrue(finished.is_set(), "a sibling read outlived the answer") + + async def test_statistics_failure_waits_for_its_sibling_reads(self): + await self._failing( + "/prompt_manager/system/statistics", + "usage_by_category", + self.api.db, + "get_statistics", + ) + + async def test_overview_failure_waits_for_its_sibling_reads(self): + await self._failing( + "/prompt_manager/system/overview", + "library_counts", + self.api, + "_output_dirs_with_origin", + ) + + if __name__ == "__main__": import unittest