From 3b0cfeb9bb05db4bcb89c0ebf9b19234222dfc2f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8F=AD=E6=89=AC?= Date: Sat, 8 Aug 2026 11:16:39 +0800 Subject: [PATCH] fix(daemon): make build-staleness checks non-blocking on status hot paths status() and the dashboard's _server_info() awaited is_stale() directly, which shells out to git twice per call. Under load (or in CI parallelism) that can exceed a caller's own deadline, which test_daemon_event_loop_blocking.py::test_status_returns_while_deferred_db_worker_is_busy caught: status() must return within 0.1s even while a deferred DB worker is busy, and it was timing out instead. Add StalenessMonitor (leapflow.utils.build_info): a TTL-cached, non-blocking wrapper. current() always returns instantly (None on first use = 'unknown, still checking') and schedules a fire-and-forget background refresh once the TTL elapses; refresh() stays available for deterministic callers (startup warmup, tests). Wire it into RuntimeLeapService.status() and DashboardServer._server_info(), with cancel_pending() invoked from both shutdown paths. checker is passed at call time (not bound at construction) so existing monkeypatch.setattr(module, "is_stale", ...) test doubles keep working. Updates the two tests that asserted on a synchronously-patched stale verdict to force one deterministic refresh() first, and adds dedicated StalenessMonitor coverage in test_build_info.py. --- src/leapflow/daemon/service.py | 17 +++++-- src/leapflow/dashboard/server.py | 18 +++++-- src/leapflow/utils/build_info.py | 74 ++++++++++++++++++++++++++- tests/test_build_info.py | 87 ++++++++++++++++++++++++++++++++ tests/test_daemon_rpc.py | 4 ++ tests/test_dashboard_launcher.py | 4 ++ 6 files changed, 195 insertions(+), 9 deletions(-) diff --git a/src/leapflow/daemon/service.py b/src/leapflow/daemon/service.py index 08ad755..b5b6ad9 100644 --- a/src/leapflow/daemon/service.py +++ b/src/leapflow/daemon/service.py @@ -39,7 +39,7 @@ from leapflow.daemon.turn_admission import TurnAdmission from leapflow.engine import StreamEvent from leapflow.memory.protocol import MemoryQuery -from leapflow.utils.build_info import capture_build_info, is_stale +from leapflow.utils.build_info import StalenessMonitor, capture_build_info, is_stale logger = logging.getLogger(__name__) @@ -70,8 +70,13 @@ def __init__(self, settings: Any, *, mock_host: bool = False) -> None: self._started_at = time.time() # Captured once at daemon startup so status() can tell a developer # this process is stale (source changed since it started) instead of - # a fixed behavior change looking like a code defect. + # a fixed behavior change looking like a code defect. The staleness + # check itself shells out to git, so it is wrapped in a non-blocking + # cache: status() must return promptly even while other background + # work (e.g. a deferred DB worker) is busy (see + # test_daemon_event_loop_blocking.py). self._build_info = capture_build_info() + self._build_staleness = StalenessMonitor(self._build_info) self._client_count: Callable[[], int] = lambda: 0 self._client_leases: Callable[[], list[ClientLeaseSnapshot]] = lambda: [] self._approval_coordinator = ApprovalCoordinator( @@ -133,6 +138,7 @@ async def start(self) -> None: self._observation = None async def shutdown(self) -> None: + self._build_staleness.cancel_pending() if self._ctx is None: return ctx = self._ctx @@ -756,7 +762,12 @@ async def status(self, session_id: str = "") -> dict[str, Any]: self._approval_coordinator.prune_stale() clients = await asyncio.to_thread(self._safe_client_lease_summaries) host = await asyncio.to_thread(host_backend_status, ctx) - build_stale = await asyncio.to_thread(is_stale, self._build_info) + # Non-blocking: returns the last cached verdict (None on the very + # first call) and refreshes it in the background, never awaiting the + # git subprocess on this hot path. `is_stale` is looked up here (not + # bound at __init__ time) so tests can still + # `monkeypatch.setattr(service_module, "is_stale", ...)`. + build_stale = self._build_staleness.current(is_stale) return { "pid": os.getpid(), "profile": getattr(settings, "profile", "default"), diff --git a/src/leapflow/dashboard/server.py b/src/leapflow/dashboard/server.py index a2c52b0..c109de7 100644 --- a/src/leapflow/dashboard/server.py +++ b/src/leapflow/dashboard/server.py @@ -25,7 +25,7 @@ from leapflow.dashboard.service import DaemonDataProvider, DashboardViewBuilder from leapflow.dashboard.templates import TemplateLibrary from leapflow.monitor.types import EVENT_ERROR, EVENT_FINDING, EVENT_HEARTBEAT, EVENT_WATCH_STATE -from leapflow.utils.build_info import capture_build_info, is_stale +from leapflow.utils.build_info import StalenessMonitor, capture_build_info, is_stale logger = logging.getLogger(__name__) @@ -60,8 +60,11 @@ def __init__( self._upstream_task: Optional[asyncio.Task[None]] = None # Captured once at server startup so /api/server-info and the ViewSpec # meta can report whether this long-lived process is still fresh - # relative to the source tree (see leapflow.utils.build_info). + # relative to the source tree (see leapflow.utils.build_info). Wrapped + # in a non-blocking cache so a browser refresh never waits on a git + # subprocess. self._build_info = capture_build_info() + self._build_staleness = StalenessMonitor(self._build_info) # ── App wiring ───────────────────────────────────────────────────────── @@ -95,6 +98,7 @@ async def serve(self) -> None: while True: await asyncio.sleep(3600) finally: + self._build_staleness.cancel_pending() await runner.cleanup() # ── Auth ─────────────────────────────────────────────────────────────── @@ -160,8 +164,14 @@ async def _handle_server_info(self, request: Any) -> Any: return web.json_response(await self._server_info()) async def _server_info(self) -> dict[str, Any]: - """Build the {build, stale} payload shared by the view meta and the endpoint.""" - stale = await asyncio.to_thread(is_stale, self._build_info) + """Build the {build, stale} payload shared by the view meta and the endpoint. + + Non-blocking: returns the last cached verdict and refreshes it in the + background once due, rather than awaiting the git subprocess on every + request. `is_stale` is looked up here (not bound at __init__ time) so + tests can still `monkeypatch.setattr(server_module, "is_stale", ...)`. + """ + stale = self._build_staleness.current(is_stale) return {"build": self._build_info.to_dict(), "stale": stale} async def _handle_action(self, request: Any) -> Any: diff --git a/src/leapflow/utils/build_info.py b/src/leapflow/utils/build_info.py index 19b48f3..7faa15f 100644 --- a/src/leapflow/utils/build_info.py +++ b/src/leapflow/utils/build_info.py @@ -20,13 +20,14 @@ from __future__ import annotations +import asyncio import hashlib import os import subprocess import time from dataclasses import dataclass from pathlib import Path -from typing import List, Optional, Tuple +from typing import Callable, List, Optional, Tuple from leapflow.version import __version__ @@ -147,4 +148,73 @@ def is_stale(captured: BuildInfo) -> Optional[bool]: return commit != captured.commit or digest != captured.dirty_digest -__all__ = ["BuildInfo", "capture_build_info", "is_stale"] +# Refresh cadence for :class:`StalenessMonitor`. Staleness is a "restart me" +# hint for a developer, not a real-time signal, so a coarse cadence trades an +# acceptable amount of freshness for zero risk of blocking a status call. +_DEFAULT_STALENESS_TTL_S = 30.0 + + +class StalenessMonitor: + """Non-blocking, TTL-cached wrapper around a staleness check. + + ``is_stale`` re-derives its verdict via a fresh subprocess call every + time, which is exactly right for correctness but wrong for a hot path: + daemon ``status()`` RPCs and dashboard view handlers must return promptly + even while other background work (e.g. a deferred DB worker) is busy, and + a caller-facing deadline can be shorter than a loaded git invocation. + + :meth:`current` never awaits the check itself — it returns the + last-known verdict (``None`` = "unknown, still checking" before the + first refresh completes) and, once ``ttl_s`` has elapsed, schedules a + fire-and-forget background refresh via :meth:`refresh`. Callers that need + a deterministic, synchronous result (tests, a first-use warmup) can + ``await monitor.refresh(checker)`` directly. + + ``checker`` is accepted as a parameter (defaulting to :func:`is_stale`) + rather than bound at construction time, so call sites that pass their own + module-level ``is_stale`` reference keep working with + ``monkeypatch.setattr(module, "is_stale", ...)``-style test doubles. + """ + + def __init__(self, captured: BuildInfo, ttl_s: float = _DEFAULT_STALENESS_TTL_S) -> None: + self._captured = captured + self._ttl_s = ttl_s + self._value: Optional[bool] = None + self._checked_at: float = 0.0 + self._refresh_task: "Optional[asyncio.Task[Optional[bool]]]" = None + + def current(self, checker: Callable[[BuildInfo], Optional[bool]] = is_stale) -> Optional[bool]: + """Return the last-known verdict; never blocks on ``checker``. + + Schedules a background refresh once ``ttl_s`` has elapsed since the + last completed check (or immediately, on first use). + """ + due = time.time() - self._checked_at >= self._ttl_s + if due and (self._refresh_task is None or self._refresh_task.done()): + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + if loop is not None: + self._refresh_task = loop.create_task(self.refresh(checker)) + return self._value + + async def refresh(self, checker: Callable[[BuildInfo], Optional[bool]] = is_stale) -> Optional[bool]: + """Synchronously run ``checker`` off-thread and cache the verdict. + + Safe to ``await`` directly when a caller needs a deterministic result + (e.g. right after startup, or in a test) instead of racing the + background refresh that :meth:`current` schedules. + """ + self._value = await asyncio.to_thread(checker, self._captured) + self._checked_at = time.time() + return self._value + + def cancel_pending(self) -> None: + """Cancel any in-flight background refresh; call during shutdown.""" + task = self._refresh_task + if task is not None and not task.done(): + task.cancel() + + +__all__ = ["BuildInfo", "StalenessMonitor", "capture_build_info", "is_stale"] diff --git a/tests/test_build_info.py b/tests/test_build_info.py index 7c2eea5..3ef9463 100644 --- a/tests/test_build_info.py +++ b/tests/test_build_info.py @@ -7,6 +7,7 @@ from __future__ import annotations +import asyncio import os import subprocess from pathlib import Path @@ -114,6 +115,92 @@ def test_to_dict_is_plain_json_safe_and_coerces_none_to_empty_string( } +# ── StalenessMonitor: non-blocking cache around is_stale ───────────────────── + + +def _info(commit: str = "abc123") -> build_info.BuildInfo: + return build_info.BuildInfo( + version=__version__, commit=commit, dirty_digest="d1", pid=os.getpid(), started_at=0.0, + ) + + +def test_staleness_monitor_current_is_none_before_first_refresh_completes() -> None: + """The first call must return instantly, never blocking on ``checker``.""" + monitor = build_info.StalenessMonitor(_info()) + + verdict = monitor.current(lambda captured: True) + + assert verdict is None # unknown: the background refresh has not run yet + + +async def test_staleness_monitor_current_reflects_background_refresh_once_it_completes() -> None: + monitor = build_info.StalenessMonitor(_info()) + monitor.current(lambda captured: True) # schedules the background refresh + + for _ in range(50): + if monitor.current(lambda captured: True) is not None: + break + await asyncio.sleep(0.01) + + assert monitor.current(lambda captured: True) is True + + +async def test_staleness_monitor_does_not_reschedule_refresh_within_ttl() -> None: + calls = 0 + + def _checker(captured: build_info.BuildInfo) -> Optional[bool]: + nonlocal calls + calls += 1 + return False + + monitor = build_info.StalenessMonitor(_info(), ttl_s=60.0) + await monitor.refresh(_checker) + assert calls == 1 + + # Within the TTL window, repeated current() calls must not schedule + # another background refresh. + for _ in range(5): + monitor.current(_checker) + await asyncio.sleep(0) + assert calls == 1 + + +async def test_staleness_monitor_refresh_is_synchronous_and_deterministic() -> None: + monitor = build_info.StalenessMonitor(_info()) + + verdict = await monitor.refresh(lambda captured: True) + + assert verdict is True + assert monitor.current(lambda captured: True) is True + + +def test_staleness_monitor_current_without_running_loop_degrades_gracefully() -> None: + """Called from sync code (no running loop), current() must not raise.""" + monitor = build_info.StalenessMonitor(_info()) + + assert monitor.current(lambda captured: True) is None + + +def test_staleness_monitor_cancel_pending_is_noop_without_pending_refresh() -> None: + monitor = build_info.StalenessMonitor(_info()) + + monitor.cancel_pending() # must not raise when nothing is in flight + + +async def test_staleness_monitor_cancel_pending_cancels_inflight_refresh() -> None: + async def _hang() -> Optional[bool]: + await asyncio.sleep(3600) + return True + + monitor = build_info.StalenessMonitor(_info()) + monitor._refresh_task = asyncio.ensure_future(_hang()) + + monitor.cancel_pending() + + with pytest.raises(asyncio.CancelledError): + await monitor._refresh_task + + # ── _run_git: graceful degradation on subprocess failure ──────────────────── diff --git a/tests/test_daemon_rpc.py b/tests/test_daemon_rpc.py index 872edf0..b5e529e 100644 --- a/tests/test_daemon_rpc.py +++ b/tests/test_daemon_rpc.py @@ -538,6 +538,10 @@ async def test_runtime_service_status_reports_own_build_staleness(tmp_path, monk service = service_module.RuntimeLeapService(make_settings(str(tmp_path)), mock_host=True) captured_pid = service._build_info.pid monkeypatch.setattr(service_module, "is_stale", lambda info: True) + # status() itself is a non-blocking cache read (see + # test_daemon_event_loop_blocking.py); force one deterministic refresh so + # this assertion does not race the background task current() schedules. + await service._build_staleness.refresh(service_module.is_stale) status = await service.status() diff --git a/tests/test_dashboard_launcher.py b/tests/test_dashboard_launcher.py index bb6837b..8210914 100644 --- a/tests/test_dashboard_launcher.py +++ b/tests/test_dashboard_launcher.py @@ -315,6 +315,10 @@ async def test_handle_server_info_reports_captured_build_and_stale_verdict( # This process's own fingerprint check is irrelevant to the endpoint's # wiring; pin the verdict so the assertion is deterministic. monkeypatch.setattr(server_module, "is_stale", lambda info: True) + # _server_info() is a non-blocking cache read; force one deterministic + # refresh so the assertion below does not race the background task that + # current() would otherwise schedule. + await server._build_staleness.refresh(server_module.is_stale) request = SimpleNamespace(query={"token": "t"}, headers={}) response = await server._handle_server_info(request)