fix: use async CLI Apps catalog refresh
Replace the manual thread-based catalog refresh with an asyncio task and async HTTP catalog fetches so the Settings route stays within the async WebUI model.
This commit is contained in:
@@ -461,6 +461,48 @@ class CliAppManager:
|
|||||||
_write_json(cache_path, {"_cached_at": _now(), "data": fetched})
|
_write_json(cache_path, {"_cached_at": _now(), "data": fetched})
|
||||||
return fetched
|
return fetched
|
||||||
|
|
||||||
|
async def _fetch_registry_async(
|
||||||
|
self,
|
||||||
|
url: str,
|
||||||
|
cache_path: Path,
|
||||||
|
*,
|
||||||
|
force_refresh: bool = False,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
data, cached_at = self._cached_registry(cache_path)
|
||||||
|
if (
|
||||||
|
not force_refresh
|
||||||
|
and data is not None
|
||||||
|
and _now() - cached_at < self.runtime.catalog_ttl_seconds
|
||||||
|
):
|
||||||
|
return data
|
||||||
|
|
||||||
|
try:
|
||||||
|
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
|
||||||
|
response = await client.get(url)
|
||||||
|
response.raise_for_status()
|
||||||
|
fetched = response.json()
|
||||||
|
if not isinstance(fetched, dict):
|
||||||
|
raise ValueError("registry response must be an object")
|
||||||
|
except Exception:
|
||||||
|
if data is not None:
|
||||||
|
return data
|
||||||
|
raise
|
||||||
|
|
||||||
|
_write_json(cache_path, {"_cached_at": _now(), "data": fetched})
|
||||||
|
return fetched
|
||||||
|
|
||||||
|
async def refresh_catalog_cache(self, *, force_refresh: bool = False) -> None:
|
||||||
|
for source, url, _raw_base, required in _CATALOG_SOURCES:
|
||||||
|
try:
|
||||||
|
await self._fetch_registry_async(
|
||||||
|
url,
|
||||||
|
self._cache_path(source),
|
||||||
|
force_refresh=force_refresh,
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
if required:
|
||||||
|
raise
|
||||||
|
|
||||||
def catalog(
|
def catalog(
|
||||||
self,
|
self,
|
||||||
*,
|
*,
|
||||||
|
|||||||
@@ -2,8 +2,8 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import re
|
import re
|
||||||
import threading
|
|
||||||
import time
|
import time
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
@@ -22,37 +22,26 @@ _CLI_APP_ATTACHMENT_KEYS = (
|
|||||||
"brand_color",
|
"brand_color",
|
||||||
)
|
)
|
||||||
_CATALOG_REFRESH_RETRY_SECONDS = 60.0
|
_CATALOG_REFRESH_RETRY_SECONDS = 60.0
|
||||||
_catalog_refresh_lock = threading.Lock()
|
_catalog_refresh_task: asyncio.Task[None] | None = None
|
||||||
_catalog_refresh_running = False
|
|
||||||
_catalog_refresh_last_started = 0.0
|
_catalog_refresh_last_started = 0.0
|
||||||
|
|
||||||
|
|
||||||
def _start_catalog_refresh() -> bool:
|
async def _refresh_catalog(manager: CliAppManager) -> None:
|
||||||
global _catalog_refresh_last_started, _catalog_refresh_running
|
try:
|
||||||
|
await manager.refresh_catalog_cache(force_refresh=True)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _start_catalog_refresh(manager: CliAppManager) -> bool:
|
||||||
|
global _catalog_refresh_last_started, _catalog_refresh_task
|
||||||
now = time.monotonic()
|
now = time.monotonic()
|
||||||
with _catalog_refresh_lock:
|
if _catalog_refresh_task is not None and not _catalog_refresh_task.done():
|
||||||
if _catalog_refresh_running:
|
return True
|
||||||
return True
|
if now - _catalog_refresh_last_started < _CATALOG_REFRESH_RETRY_SECONDS:
|
||||||
if now - _catalog_refresh_last_started < _CATALOG_REFRESH_RETRY_SECONDS:
|
return False
|
||||||
return False
|
_catalog_refresh_last_started = now
|
||||||
_catalog_refresh_running = True
|
_catalog_refresh_task = asyncio.create_task(_refresh_catalog(manager))
|
||||||
_catalog_refresh_last_started = now
|
|
||||||
|
|
||||||
def refresh() -> None:
|
|
||||||
global _catalog_refresh_running
|
|
||||||
try:
|
|
||||||
_manager().catalog(force_refresh=True)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
finally:
|
|
||||||
with _catalog_refresh_lock:
|
|
||||||
_catalog_refresh_running = False
|
|
||||||
|
|
||||||
threading.Thread(
|
|
||||||
target=refresh,
|
|
||||||
name="nanobot-cli-app-catalog-refresh",
|
|
||||||
daemon=True,
|
|
||||||
).start()
|
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
@@ -108,14 +97,14 @@ def _manager() -> CliAppManager:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def cli_apps_payload(*, installed_only: bool = False) -> dict[str, Any]:
|
async def cli_apps_payload(*, installed_only: bool = False) -> dict[str, Any]:
|
||||||
manager = _manager()
|
manager = _manager()
|
||||||
if installed_only:
|
if installed_only:
|
||||||
return manager.installed_payload()
|
return manager.installed_payload()
|
||||||
payload = manager.payload(cache_only=True)
|
payload = manager.payload(cache_only=True)
|
||||||
refresh_pending = False
|
refresh_pending = False
|
||||||
if not manager.catalog_cache_fresh(include_optional=True):
|
if not manager.catalog_cache_fresh(include_optional=True):
|
||||||
refresh_pending = _start_catalog_refresh()
|
refresh_pending = _start_catalog_refresh(manager)
|
||||||
if not payload["apps"]:
|
if not payload["apps"]:
|
||||||
installed = manager.installed_payload()
|
installed = manager.installed_payload()
|
||||||
if installed["apps"]:
|
if installed["apps"]:
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ request mapping and response shaping.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import inspect
|
||||||
import json
|
import json
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
from typing import Any
|
from typing import Any
|
||||||
@@ -309,10 +310,12 @@ class WebUISettingsRouter:
|
|||||||
"yes",
|
"yes",
|
||||||
}
|
}
|
||||||
try:
|
try:
|
||||||
if installed_only:
|
payload_result = (
|
||||||
payload = await asyncio.to_thread(cli_apps_payload, installed_only=True)
|
cli_apps_payload(installed_only=True)
|
||||||
else:
|
if installed_only
|
||||||
payload = await asyncio.to_thread(cli_apps_payload)
|
else cli_apps_payload()
|
||||||
|
)
|
||||||
|
payload = await payload_result if inspect.isawaitable(payload_result) else payload_result
|
||||||
except Exception:
|
except Exception:
|
||||||
self.logger.exception("failed to load CLI Apps payload")
|
self.logger.exception("failed to load CLI Apps payload")
|
||||||
return self._error_response(500, "failed to load CLI Apps")
|
return self._error_response(500, "failed to load CLI Apps")
|
||||||
|
|||||||
@@ -5,8 +5,8 @@ import functools
|
|||||||
import json
|
import json
|
||||||
import random
|
import random
|
||||||
import socket
|
import socket
|
||||||
import threading
|
|
||||||
import time
|
import time
|
||||||
|
from contextlib import suppress
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
from unittest.mock import AsyncMock, MagicMock
|
from unittest.mock import AsyncMock, MagicMock
|
||||||
@@ -484,12 +484,13 @@ async def test_cli_apps_catalog_does_not_block_other_webui_http_routes(
|
|||||||
tmp_path: Path,
|
tmp_path: Path,
|
||||||
monkeypatch: pytest.MonkeyPatch,
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
) -> None:
|
) -> None:
|
||||||
entered = threading.Event()
|
entered = asyncio.Event()
|
||||||
release = threading.Event()
|
release = asyncio.Event()
|
||||||
|
|
||||||
def slow_payload() -> dict[str, Any]:
|
async def slow_payload() -> dict[str, Any]:
|
||||||
entered.set()
|
entered.set()
|
||||||
release.wait(2.0)
|
with suppress(asyncio.TimeoutError):
|
||||||
|
await asyncio.wait_for(release.wait(), 2.0)
|
||||||
return {"apps": [], "installed_count": 0, "catalog_updated_at": None}
|
return {"apps": [], "installed_count": 0, "catalog_updated_at": None}
|
||||||
|
|
||||||
monkeypatch.setattr("nanobot.webui.settings_routes.cli_apps_payload", slow_payload)
|
monkeypatch.setattr("nanobot.webui.settings_routes.cli_apps_payload", slow_payload)
|
||||||
@@ -505,7 +506,7 @@ async def test_cli_apps_catalog_does_not_block_other_webui_http_routes(
|
|||||||
catalog_task = asyncio.create_task(
|
catalog_task = asyncio.create_task(
|
||||||
_http_get("http://127.0.0.1:29935/api/settings/cli-apps", headers=auth)
|
_http_get("http://127.0.0.1:29935/api/settings/cli-apps", headers=auth)
|
||||||
)
|
)
|
||||||
assert await asyncio.to_thread(entered.wait, 2.0)
|
assert await asyncio.wait_for(entered.wait(), 2.0)
|
||||||
assert time.perf_counter() - started < 1.0
|
assert time.perf_counter() - started < 1.0
|
||||||
|
|
||||||
workspaces_started = time.perf_counter()
|
workspaces_started = time.perf_counter()
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from nanobot.webui import cli_apps_api
|
from nanobot.webui import cli_apps_api
|
||||||
@@ -60,9 +61,13 @@ def test_cli_apps_payload_uses_cache_and_marks_refresh_pending(monkeypatch) -> N
|
|||||||
manager = _FakeManager(fresh=False)
|
manager = _FakeManager(fresh=False)
|
||||||
refreshes = []
|
refreshes = []
|
||||||
monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager)
|
monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager)
|
||||||
monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda: refreshes.append(True) or True)
|
monkeypatch.setattr(
|
||||||
|
cli_apps_api,
|
||||||
|
"_start_catalog_refresh",
|
||||||
|
lambda _manager: refreshes.append(True) or True,
|
||||||
|
)
|
||||||
|
|
||||||
payload = cli_apps_api.cli_apps_payload()
|
payload = asyncio.run(cli_apps_api.cli_apps_payload())
|
||||||
|
|
||||||
assert manager.payload_calls == [True]
|
assert manager.payload_calls == [True]
|
||||||
assert manager.fresh_checks == [True]
|
assert manager.fresh_checks == [True]
|
||||||
@@ -95,9 +100,13 @@ def test_cli_apps_payload_skips_refresh_when_cache_is_fresh(monkeypatch) -> None
|
|||||||
)
|
)
|
||||||
refreshes = []
|
refreshes = []
|
||||||
monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager)
|
monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager)
|
||||||
monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda: refreshes.append(True) or True)
|
monkeypatch.setattr(
|
||||||
|
cli_apps_api,
|
||||||
|
"_start_catalog_refresh",
|
||||||
|
lambda _manager: refreshes.append(True) or True,
|
||||||
|
)
|
||||||
|
|
||||||
payload = cli_apps_api.cli_apps_payload()
|
payload = asyncio.run(cli_apps_api.cli_apps_payload())
|
||||||
|
|
||||||
assert manager.payload_calls == [True]
|
assert manager.payload_calls == [True]
|
||||||
assert manager.fresh_checks == [True]
|
assert manager.fresh_checks == [True]
|
||||||
@@ -131,9 +140,13 @@ def test_cli_apps_payload_refreshes_when_optional_cache_is_stale(monkeypatch) ->
|
|||||||
)
|
)
|
||||||
refreshes = []
|
refreshes = []
|
||||||
monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager)
|
monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager)
|
||||||
monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda: refreshes.append(True) or True)
|
monkeypatch.setattr(
|
||||||
|
cli_apps_api,
|
||||||
|
"_start_catalog_refresh",
|
||||||
|
lambda _manager: refreshes.append(True) or True,
|
||||||
|
)
|
||||||
|
|
||||||
payload = cli_apps_api.cli_apps_payload()
|
payload = asyncio.run(cli_apps_api.cli_apps_payload())
|
||||||
|
|
||||||
assert manager.payload_calls == [True]
|
assert manager.payload_calls == [True]
|
||||||
assert manager.fresh_checks == [True]
|
assert manager.fresh_checks == [True]
|
||||||
@@ -145,9 +158,9 @@ def test_cli_apps_payload_refreshes_when_optional_cache_is_stale(monkeypatch) ->
|
|||||||
def test_cli_apps_payload_reports_not_pending_when_refresh_is_throttled(monkeypatch) -> None:
|
def test_cli_apps_payload_reports_not_pending_when_refresh_is_throttled(monkeypatch) -> None:
|
||||||
manager = _FakeManager(fresh=False)
|
manager = _FakeManager(fresh=False)
|
||||||
monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager)
|
monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager)
|
||||||
monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda: False)
|
monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda _manager: False)
|
||||||
|
|
||||||
payload = cli_apps_api.cli_apps_payload()
|
payload = asyncio.run(cli_apps_api.cli_apps_payload())
|
||||||
|
|
||||||
assert manager.payload_calls == [True]
|
assert manager.payload_calls == [True]
|
||||||
assert manager.fresh_checks == [True]
|
assert manager.fresh_checks == [True]
|
||||||
|
|||||||
Reference in New Issue
Block a user