diff --git a/backend/app/api/system_activity.py b/backend/app/api/system_activity.py index 480b8c9..0d0600c 100644 --- a/backend/app/api/system_activity.py +++ b/backend/app/api/system_activity.py @@ -20,6 +20,7 @@ from sqlalchemy import desc, func, select from ..config import get_config from ..extensions import get_session from ..models import TaskRun +from ..services.scheduler_service import scheduler_status system_activity_bp = Blueprint( "system_activity", __name__, url_prefix="/api/system/activity", @@ -81,17 +82,22 @@ def _read_workers_sync() -> dict: } +async def _queues_cached() -> dict: + """Per-queue Redis LLEN, cached 2s. Shared by /queues and /summary.""" + now = time.time() + if _QUEUE_CACHE["data"] is None or (now - _QUEUE_CACHE["ts"]) > _QUEUE_CACHE_TTL: + _QUEUE_CACHE["data"] = await asyncio.to_thread(_read_queues_sync) + _QUEUE_CACHE["ts"] = now + return _QUEUE_CACHE["data"] + + @system_activity_bp.route("/queues", methods=["GET"]) async def get_queues(): """Per-queue Redis LLEN. Cached 2s. Response: {queues: {name: depth_or_null}, fetched_at: iso8601} """ - now = time.time() - if _QUEUE_CACHE["data"] is None or (now - _QUEUE_CACHE["ts"]) > _QUEUE_CACHE_TTL: - _QUEUE_CACHE["data"] = await asyncio.to_thread(_read_queues_sync) - _QUEUE_CACHE["ts"] = now - return jsonify(_QUEUE_CACHE["data"]) + return jsonify(await _queues_cached()) @system_activity_bp.route("/workers", methods=["GET"]) @@ -107,6 +113,35 @@ async def get_workers(): return jsonify(_WORKER_CACHE["data"]) +@system_activity_bp.route("/summary", methods=["GET"]) +async def get_summary(): + """One-call rollup for the always-on TopNav pipeline indicator: + scheduler health, per-queue pending depths, currently-running count, and + recent (24h) failure count. Cheap — cached queue LLENs + two TaskRun + counts — so it's safe to poll app-wide.""" + queues_data = await _queues_cached() + depths = queues_data.get("queues", {}) + queued_total = sum(v for v in depths.values() if isinstance(v, int)) + since = datetime.now(UTC) - timedelta(hours=24) + async with get_session() as session: + scheduler = await scheduler_status(session) + running = (await session.execute( + select(func.count(TaskRun.id)).where(TaskRun.status == "running") + )).scalar_one() + failing = (await session.execute( + select(func.count(TaskRun.id)) + .where(TaskRun.status.in_(["error", "timeout"])) + .where(TaskRun.finished_at >= since) + )).scalar_one() + return jsonify({ + "scheduler": scheduler, + "queues": depths, + "queued_total": queued_total, + "running": int(running), + "failing": int(failing), + }) + + @system_activity_bp.route("/runs", methods=["GET"]) async def list_runs(): """Paginated task_run history. Query params: diff --git a/frontend/src/components/PipelineStatusChip.vue b/frontend/src/components/PipelineStatusChip.vue new file mode 100644 index 0000000..0b026c0 --- /dev/null +++ b/frontend/src/components/PipelineStatusChip.vue @@ -0,0 +1,146 @@ + + + + + + + mdi-progress-download{{ running }} + + + mdi-tray-full{{ queued }} + + + mdi-alert-circle{{ failing }} + + idle + + + + + + + Scheduler + + {{ schedLabel }} + + + + Queues + All idle + + {{ q.name }}{{ q.depth }} + + + + + Running{{ running }} + Queued{{ queued }} + + Failures 24h + {{ failing }} + + + + Open downloads → + + + + + + + diff --git a/frontend/src/components/TopNav.vue b/frontend/src/components/TopNav.vue index b2d7939..07b5983 100644 --- a/frontend/src/components/TopNav.vue +++ b/frontend/src/components/TopNav.vue @@ -8,6 +8,7 @@ {{ health.icon }} + @@ -29,6 +30,7 @@ import { computed, onMounted } from 'vue' import router, { FRONT_DOOR } from '../router.js' import { useSystemStore } from '../stores/system.js' +import PipelineStatusChip from './PipelineStatusChip.vue' const system = useSystemStore() onMounted(() => system.refreshHealth()) diff --git a/frontend/src/stores/systemActivity.js b/frontend/src/stores/systemActivity.js index e152ade..7d8252f 100644 --- a/frontend/src/stores/systemActivity.js +++ b/frontend/src/stores/systemActivity.js @@ -93,11 +93,23 @@ export const useSystemActivityStore = defineStore('systemActivity', () => { runsFilter.value = { ...runsFilter.value, ...filter } } + // One-call rollup for the always-on TopNav pipeline indicator: + // { scheduler, queues, queued_total, running, failing }. + const summary = ref(null) + async function loadSummary() { + try { + summary.value = await api.get('/api/system/activity/summary') + } catch (e) { + lastError.value = e.message + } + return summary.value + } + return { - queues, workers, recentMinute, failures, + queues, workers, recentMinute, failures, summary, runs, runsCursor, runsHasMore, runsFilter, loading, lastError, loadQueues, loadWorkers, loadRecentMinute, - loadRuns, loadFailures, setFilter, + loadRuns, loadFailures, loadSummary, setFilter, } }) diff --git a/tests/test_api_system_activity.py b/tests/test_api_system_activity.py index 0a493ab..f5832ef 100644 --- a/tests/test_api_system_activity.py +++ b/tests/test_api_system_activity.py @@ -44,6 +44,25 @@ async def test_queues_returns_all_known_queues(client, monkeypatch): assert set(body["queues"].keys()) == set(activity_module._QUEUE_NAMES) +@pytest.mark.asyncio +async def test_summary_returns_rollup_shape(client, monkeypatch): + def _fake(): + return { + "queues": {**dict.fromkeys(activity_module._QUEUE_NAMES, 0), "ml": 3}, + "fetched_at": datetime.now(UTC).isoformat(), + } + monkeypatch.setattr(activity_module, "_read_queues_sync", _fake) + + resp = await client.get("/api/system/activity/summary") + assert resp.status_code == 200 + body = await resp.get_json() + assert set(body.keys()) == {"scheduler", "queues", "queued_total", "running", "failing"} + assert body["queued_total"] == 3 # only ml has a non-zero depth + assert isinstance(body["running"], int) + assert isinstance(body["failing"], int) + assert set(body["scheduler"]) == {"last_tick_at", "next_due_at", "due_now", "auto_sources"} + + @pytest.mark.asyncio async def test_queues_cached_within_ttl(client, monkeypatch): call_count = {"n": 0}