fix(system): answer a failed System read only after its other reads finish
CI (py3.10 on ubuntu) crashed with a segfault inside SQLite in test_statistics_failure_is_an_error_envelope: the statistics and overview routes used asyncio.gather without return_exceptions, so the 500 answered on the first failure while get_statistics still ran in the executor, and the test tore the database down under it. Both routes now await every read before raising the first failure (failing before, passing after: TestFailedReadsSettleFirst). The frontend coverage merge failed on workspace/activity.mjs branches (86.3 < 87): no test asserted Activity's empty and failed-read states, so their branch was only reached in passing, depending on timing. A spec now asserts both.
This commit is contained in:
+15
-2
@@ -40,6 +40,19 @@ _HEALTH_STATUS_LINES = {
|
||||
_StoredReport = namedtuple("_StoredReport", "generation report")
|
||||
|
||||
|
||||
async def _gather_settled(*reads):
|
||||
"""Await every read, then raise the first failure.
|
||||
|
||||
Unlike a plain gather, a failure never answers while the other executor
|
||||
reads still run against the database.
|
||||
"""
|
||||
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
|
||||
@@ -154,7 +167,7 @@ class SystemRoutesMixin:
|
||||
from ..config import GalleryConfig
|
||||
|
||||
try:
|
||||
(output_dirs, auto_detected), counts = await asyncio.gather(
|
||||
(output_dirs, auto_detected), counts = await _gather_settled(
|
||||
self._run_in_executor(self._output_dirs_with_origin),
|
||||
self._run_in_executor(self.db.library_counts),
|
||||
)
|
||||
@@ -188,7 +201,7 @@ class SystemRoutesMixin:
|
||||
|
||||
async def get_system_statistics(self, request):
|
||||
try:
|
||||
by_category, totals, counts = await asyncio.gather(
|
||||
by_category, totals, counts = await _gather_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),
|
||||
|
||||
@@ -284,6 +284,19 @@ test('reload during a scan reattaches to the running job', async ({ page, reques
|
||||
await expect(pill(page)).toBeHidden();
|
||||
});
|
||||
|
||||
test('an empty Activity says no job has run, and a list that cannot be read says so', async ({ page }) => {
|
||||
// The empty and failed states were only reached in passing by other tests, so their coverage depended on timing.
|
||||
const empty = main(page).locator('section.surface > p.empty-state');
|
||||
await page.goto(`${workspace}#/activity`);
|
||||
await expect(empty).toHaveText('No jobs have run since ComfyUI started. Scans and other long operations appear here.');
|
||||
await expect(empty).not.toHaveAttribute('role');
|
||||
await expect(items(page)).toHaveCount(0);
|
||||
await page.route(url => url.pathname === '/prompt_manager/jobs', route => route.abort());
|
||||
await page.reload();
|
||||
await expect(empty).toHaveText('Activity could not be read. It is tried again shortly.');
|
||||
await expect(empty).toHaveAttribute('role', 'alert');
|
||||
});
|
||||
|
||||
test('lost server clears the pill and marks jobs ended', async ({ page, request }) => {
|
||||
// A poll interval after the restart, then failed re-reads backing off for 1 + 2 + 4 s.
|
||||
test.setTimeout(90_000);
|
||||
|
||||
@@ -416,6 +416,63 @@ class TestStatisticsRoute(SystemAPITestCase):
|
||||
)
|
||||
|
||||
|
||||
class TestFailedReadsSettleFirst(SystemAPITestCase):
|
||||
"""A failed read answers only after the route's other reads finished.
|
||||
|
||||
Answering on the first failure (``asyncio.gather`` without
|
||||
``return_exceptions``) left the other executor reads running against a
|
||||
database the caller may already be closing; on Python 3.10 CI that
|
||||
crashed the test process inside SQLite.
|
||||
"""
|
||||
|
||||
async def assert_answers_after_reads(self, url, failing, slow, error):
|
||||
raised, release, finished = (threading.Event() for _ in range(3))
|
||||
real = getattr(*slow)
|
||||
|
||||
def fail(*args, **kwargs):
|
||||
raised.set()
|
||||
raise RuntimeError("secret")
|
||||
|
||||
def held(*args, **kwargs):
|
||||
release.wait()
|
||||
try:
|
||||
return real(*args, **kwargs)
|
||||
finally:
|
||||
finished.set()
|
||||
|
||||
with (
|
||||
patch.object(*failing, side_effect=fail),
|
||||
patch.object(*slow, side_effect=held),
|
||||
self.assertLogs("prompt_manager.api", "ERROR"),
|
||||
):
|
||||
answer = asyncio.ensure_future(self.json("GET", url, 500))
|
||||
try:
|
||||
await asyncio.get_running_loop().run_in_executor(None, raised.wait)
|
||||
done, _ = await asyncio.wait({answer}, timeout=0.5)
|
||||
self.assertEqual(done, set(), "answered while a read still ran")
|
||||
finally:
|
||||
release.set()
|
||||
body = await answer
|
||||
self.assertTrue(finished.is_set())
|
||||
self.assertEqual(body, {"success": False, "code": None, "error": error})
|
||||
|
||||
async def test_statistics(self):
|
||||
await self.assert_answers_after_reads(
|
||||
"/prompt_manager/system/statistics",
|
||||
(self.api.db, "usage_by_category"),
|
||||
(self.api.db, "get_statistics"),
|
||||
"Failed to read statistics",
|
||||
)
|
||||
|
||||
async def test_overview(self):
|
||||
await self.assert_answers_after_reads(
|
||||
"/prompt_manager/system/overview",
|
||||
(self.api.db, "library_counts"),
|
||||
(self.api, "_output_dirs_with_origin"),
|
||||
"Failed to read the overview",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import unittest
|
||||
|
||||
|
||||
Reference in New Issue
Block a user