From 49fa04a2356dfa10201a4e78bc0d9dc251588797 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E7=B6=A0BG?= Date: Sat, 13 Jun 2026 17:19:23 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(dashboard):=20use=20managed?= =?UTF-8?q?=20cron=20threadpool?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (cherry picked from commit 97fabc289c623549310125ac3804e21afde87c43) --- hermes_cli/web_server.py | 10 ++++- .../test_web_server_cron_profiles.py | 39 +++++++++++++++---- 2 files changed, 40 insertions(+), 9 deletions(-) diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 00931549fe3..d75b2cba616 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -21,6 +21,7 @@ from dataclasses import dataclass from datetime import datetime, timezone import hashlib import hmac +import inspect import importlib.util import json import logging @@ -93,6 +94,7 @@ try: from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from pydantic import BaseModel + from starlette.concurrency import run_in_threadpool except ImportError: # First try lazy-installing the dashboard extras. Only the user actually # running `hermes dashboard` needs fastapi+uvicorn; lazy install keeps @@ -108,6 +110,7 @@ except ImportError: from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from pydantic import BaseModel + from starlette.concurrency import run_in_threadpool except Exception: raise SystemExit( "Web UI requires fastapi and uvicorn.\n" @@ -10004,7 +10007,12 @@ def _list_cron_jobs_sync(profile: str = "all"): async def _run_cron_dashboard_io(func, *args, **kwargs): """Run cron dashboard profile/job I/O outside the FastAPI event loop.""" - return await asyncio.to_thread(func, *args, **kwargs) + if inspect.iscoroutinefunction(func): + raise TypeError("_run_cron_dashboard_io only accepts sync callables") + result = await run_in_threadpool(func, *args, **kwargs) + if inspect.isawaitable(result): + raise TypeError("_run_cron_dashboard_io sync callable returned an awaitable") + return result @app.get("/api/cron/jobs") diff --git a/tests/hermes_cli/test_web_server_cron_profiles.py b/tests/hermes_cli/test_web_server_cron_profiles.py index 0e6a268326e..185b7bac32a 100644 --- a/tests/hermes_cli/test_web_server_cron_profiles.py +++ b/tests/hermes_cli/test_web_server_cron_profiles.py @@ -1,5 +1,6 @@ """Regression tests for dashboard cron job profile routing.""" +from queue import Empty, SimpleQueue import threading import pytest @@ -24,6 +25,15 @@ def isolated_profiles(tmp_path, monkeypatch): return {"default": default_home, "worker_alpha": worker_home} +def _drain_queue(q): + values = [] + while True: + try: + values.append(q.get_nowait()) + except Empty: + return values + + def test_call_cron_for_profile_routes_storage_and_restores_globals(isolated_profiles): from cron import jobs as cron_jobs from hermes_cli import web_server @@ -174,17 +184,17 @@ async def test_cron_profile_scan_runs_off_event_loop(isolated_profiles, monkeypa ) event_loop_thread = threading.get_ident() - profile_scan_threads = [] - worker_threads = [] + profile_scan_threads = SimpleQueue() + worker_threads = SimpleQueue() original_profile_dicts = web_server._cron_profile_dicts original_find = web_server._find_cron_job_profile def tracking_profile_dicts(): - profile_scan_threads.append(threading.get_ident()) + profile_scan_threads.put(threading.get_ident()) return original_profile_dicts() def tracking_find(job_id): - worker_threads.append(threading.get_ident()) + worker_threads.put(threading.get_ident()) return original_find(job_id) monkeypatch.setattr(web_server, "_cron_profile_dicts", tracking_profile_dicts) @@ -195,10 +205,23 @@ async def test_cron_profile_scan_runs_off_event_loop(isolated_profiles, monkeypa assert any(job["id"] == worker_job["id"] for job in jobs) assert paused["profile"] == "worker_alpha" - assert profile_scan_threads - assert worker_threads - assert all(thread_id != event_loop_thread for thread_id in profile_scan_threads) - assert all(thread_id != event_loop_thread for thread_id in worker_threads) + profile_scan_thread_ids = _drain_queue(profile_scan_threads) + worker_thread_ids = _drain_queue(worker_threads) + assert profile_scan_thread_ids + assert worker_thread_ids + assert all(thread_id != event_loop_thread for thread_id in profile_scan_thread_ids) + assert all(thread_id != event_loop_thread for thread_id in worker_thread_ids) + + +@pytest.mark.asyncio +async def test_cron_dashboard_io_rejects_async_callables(): + from hermes_cli import web_server + + async def async_callable(): + return "nope" + + with pytest.raises(TypeError, match="only accepts sync callables"): + await web_server._run_cron_dashboard_io(async_callable)