🐛 fix(dashboard): use managed cron threadpool

(cherry picked from commit 97fabc289c)
This commit is contained in:
墨綠BG 2026-06-13 17:19:23 +08:00 committed by kshitij
parent 346e5673de
commit 49fa04a235
2 changed files with 40 additions and 9 deletions

View file

@ -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")

View file

@ -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)