fix(settings): the plugin catalog is fetched by a scheduled task, never by a page load
The operator's ruling (Background Jobs, 2026-10-08): a Steward page reads stored records; the sources are contacted by scheduled collection. An audit of every GET handler, HTMX fragment, widget poll and MCP read tool found one page load that reached out: Settings -> Plugins' catalog fragment fetched each plugin repository (5 min cache, 10 s timeout each, retried on every load when unreachable). plugin_catalog_refresh ticks every minute and fetches a repository whose copy is missing or older than an hour, waiting 5 min after a failure; saving repositories clears the cache, so the next tick fetches. The fragment only reads the cache and says which repositories are not fetched yet. Refresh is now a POST that fetches at once, as does install. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -477,6 +477,26 @@ def _register_core_tasks(app: Quart) -> None:
|
||||
)
|
||||
)
|
||||
|
||||
# Fetch the plugin catalog from the configured repositories as each copy
|
||||
# falls due (plugin_index.refresh_due), so Settings → Plugins only reads
|
||||
# what is stored (Background Jobs ruling, 2026-10-08). The repositories are
|
||||
# read on every tick, so an edit in Settings is fetched within a minute.
|
||||
async def run_catalog_refresh():
|
||||
from .core.plugin_index import refresh_due, repo_urls
|
||||
from .core.settings import get_all_settings
|
||||
async with app.db_sessionmaker() as db:
|
||||
settings = await get_all_settings(db)
|
||||
await refresh_due(repo_urls(settings))
|
||||
|
||||
app._task_registry.append(
|
||||
ScheduledTask(
|
||||
name="plugin_catalog_refresh",
|
||||
coro_factory=run_catalog_refresh,
|
||||
interval_seconds=60,
|
||||
run_on_startup=True,
|
||||
)
|
||||
)
|
||||
|
||||
# Fire due Ansible schedules (recurring playbook runs). Checks every minute;
|
||||
# each schedule's own interval gates whether it actually runs.
|
||||
async def run_ansible_schedules():
|
||||
|
||||
@@ -7,9 +7,16 @@ from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_CACHE_TTL = 300 # seconds
|
||||
# Per-URL cache: url → (raw_data, fetched_at monotonic)
|
||||
# The catalog is fetched by the scheduled plugin_catalog_refresh task, never by
|
||||
# a page load (Background Jobs ruling, 2026-10-08): a repository's copy is
|
||||
# fetched when it has none or it is older than CATALOG_MAX_AGE, and a repository
|
||||
# that failed is tried again after RETRY_AFTER. Refresh on Settings → Plugins and
|
||||
# installing a plugin fetch at once, because the operator asked.
|
||||
CATALOG_MAX_AGE = 3600 # seconds
|
||||
RETRY_AFTER = 300 # seconds
|
||||
# url → (raw index, fetched_at monotonic); url → failed_at monotonic
|
||||
_url_cache: dict[str, tuple[dict, float]] = {}
|
||||
_failed_at: dict[str, float] = {}
|
||||
|
||||
|
||||
class CatalogPlugin:
|
||||
@@ -35,59 +42,92 @@ class CatalogPlugin:
|
||||
self.tags: list[str] = data.get("tags", [])
|
||||
|
||||
|
||||
async def _fetch_one(url: str, force: bool = False) -> list[CatalogPlugin]:
|
||||
"""Fetch a single index URL, using per-URL cache."""
|
||||
def plugin_repos(settings: dict) -> list[dict]:
|
||||
"""The configured plugin repositories, [{"name", "url"}], from settings.
|
||||
|
||||
Migrates the legacy plugins.index_url single-URL field to the list format
|
||||
on first access if plugins.repositories is not yet set.
|
||||
"""
|
||||
repos = settings.get("plugins.repositories")
|
||||
if repos and isinstance(repos, list):
|
||||
return repos
|
||||
legacy = settings.get("plugins.index_url", "").strip()
|
||||
if legacy:
|
||||
return [{"name": "Default", "url": legacy}]
|
||||
return []
|
||||
|
||||
|
||||
def repo_urls(settings: dict) -> list[str]:
|
||||
return [r["url"].strip() for r in plugin_repos(settings) if (r.get("url") or "").strip()]
|
||||
|
||||
|
||||
def _plugins(raw: dict) -> list[CatalogPlugin]:
|
||||
return [CatalogPlugin(p) for p in raw.get("plugins", [])]
|
||||
|
||||
|
||||
async def _fetch_one(url: str) -> bool:
|
||||
"""Fetch one index URL into the cache. True when it was fetched."""
|
||||
import httpx
|
||||
import yaml
|
||||
|
||||
now = time.monotonic()
|
||||
if not force and url in _url_cache:
|
||||
raw, fetched_at = _url_cache[url]
|
||||
if now - fetched_at < _CACHE_TTL:
|
||||
return [CatalogPlugin(p) for p in raw.get("plugins", [])]
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
resp = await client.get(url)
|
||||
resp.raise_for_status()
|
||||
raw = yaml.safe_load(resp.text) or {}
|
||||
_url_cache[url] = (raw, now)
|
||||
logger.info("Plugin catalog fetched from %s: %d plugin(s)", url, len(raw.get("plugins", [])))
|
||||
return [CatalogPlugin(p) for p in raw.get("plugins", [])]
|
||||
except Exception:
|
||||
logger.exception("Failed to fetch plugin catalog from %s", url)
|
||||
if url in _url_cache:
|
||||
raw, _ = _url_cache[url]
|
||||
logger.warning("Returning stale cached catalog for %s", url)
|
||||
return [CatalogPlugin(p) for p in raw.get("plugins", [])]
|
||||
return []
|
||||
_failed_at[url] = time.monotonic()
|
||||
return False
|
||||
_url_cache[url] = (raw, time.monotonic())
|
||||
_failed_at.pop(url, None)
|
||||
logger.info("Plugin catalog fetched from %s: %d plugin(s)", url, len(raw.get("plugins", [])))
|
||||
return True
|
||||
|
||||
|
||||
async def fetch_catalog(
|
||||
index_urls: str | list[str],
|
||||
force: bool = False,
|
||||
) -> list[CatalogPlugin]:
|
||||
"""Fetch and merge plugin catalogs from one or more index URLs.
|
||||
|
||||
Results are cached per-URL for _CACHE_TTL seconds. Plugins with the same
|
||||
name across repos are deduplicated — the first occurrence (repo list order)
|
||||
wins. Pass force=True to bypass all caches.
|
||||
"""
|
||||
if isinstance(index_urls, str):
|
||||
index_urls = [index_urls]
|
||||
|
||||
def _merge(urls: list[str]) -> list[CatalogPlugin]:
|
||||
"""Plugins from the cached copies, deduplicated by name: the first
|
||||
repository in the list wins."""
|
||||
seen: dict[str, CatalogPlugin] = {}
|
||||
for url in index_urls:
|
||||
url = url.strip()
|
||||
if not url:
|
||||
continue
|
||||
for plugin in await _fetch_one(url, force=force):
|
||||
if plugin.name not in seen:
|
||||
seen[plugin.name] = plugin
|
||||
|
||||
for url in urls:
|
||||
if url in _url_cache:
|
||||
for plugin in _plugins(_url_cache[url][0]):
|
||||
seen.setdefault(plugin.name, plugin)
|
||||
return list(seen.values())
|
||||
|
||||
|
||||
def cached_catalog(urls: list[str]) -> tuple[list[CatalogPlugin], list[str]]:
|
||||
"""(the merged catalog from what is cached, the URLs with no copy yet).
|
||||
Reads the cache only: this is what a page shows."""
|
||||
return _merge(urls), [u for u in urls if u not in _url_cache]
|
||||
|
||||
|
||||
async def fetch_catalog(urls: list[str]) -> list[CatalogPlugin]:
|
||||
"""Fetch every URL now, then the merged catalog (a failed URL keeps its
|
||||
last copy). For an operator's Refresh or install, not for a page load."""
|
||||
for url in urls:
|
||||
await _fetch_one(url)
|
||||
return _merge(urls)
|
||||
|
||||
|
||||
async def refresh_due(urls: list[str], now: float | None = None) -> int:
|
||||
"""The scheduled task's tick: fetch each URL with no copy, or one older
|
||||
than CATALOG_MAX_AGE, unless it failed within RETRY_AFTER. Returns how
|
||||
many were fetched."""
|
||||
now = time.monotonic() if now is None else now
|
||||
fetched = 0
|
||||
for url in urls:
|
||||
cached = _url_cache.get(url)
|
||||
if cached and now - cached[1] < CATALOG_MAX_AGE:
|
||||
continue
|
||||
if url in _failed_at and now - _failed_at[url] < RETRY_AFTER:
|
||||
continue
|
||||
fetched += await _fetch_one(url)
|
||||
return fetched
|
||||
|
||||
|
||||
def clear_catalog_cache() -> None:
|
||||
"""Invalidate all in-process catalog caches."""
|
||||
"""Forget every cached copy and failure, so the next tick of the scheduled
|
||||
task fetches the repositories again (after they are edited)."""
|
||||
_url_cache.clear()
|
||||
_failed_at.clear()
|
||||
|
||||
+38
-46
@@ -7,6 +7,7 @@ from quart import Blueprint, current_app, render_template, request, redirect, ur
|
||||
|
||||
from steward.auth.middleware import require_role
|
||||
from steward.core.audit import audit_request, log_audit
|
||||
from steward.core.plugin_index import plugin_repos, repo_urls
|
||||
from steward.core.logging_setup import LEVELS as LOG_LEVELS, set_log_level
|
||||
from steward.models.users import UserRole
|
||||
from steward.core.settings import (
|
||||
@@ -606,28 +607,12 @@ async def send_report_now():
|
||||
|
||||
# ── Plugin repository helpers ─────────────────────────────────────────────────
|
||||
|
||||
def _get_plugin_repos(settings: dict) -> list[dict]:
|
||||
"""Return the list of plugin repositories from settings.
|
||||
|
||||
Migrates the legacy plugins.index_url single-URL field to the list format
|
||||
on first access if plugins.repositories is not yet set.
|
||||
"""
|
||||
repos = settings.get("plugins.repositories")
|
||||
if repos and isinstance(repos, list):
|
||||
return repos
|
||||
# Migrate legacy single URL
|
||||
legacy = settings.get("plugins.index_url", "").strip()
|
||||
if legacy:
|
||||
return [{"name": "Default", "url": legacy}]
|
||||
return []
|
||||
|
||||
|
||||
async def _save_plugin_repos(db, repos: list[dict]) -> None:
|
||||
await set_setting(db, "plugins.repositories", repos)
|
||||
|
||||
|
||||
async def _plugin_repos_partial(settings: dict):
|
||||
repos = _get_plugin_repos(settings)
|
||||
repos = plugin_repos(settings)
|
||||
return await render_template("settings/_plugin_repos.html", repos=repos)
|
||||
|
||||
|
||||
@@ -693,7 +678,7 @@ async def plugins():
|
||||
for p in discovered:
|
||||
p["_kind"] = p.get("kind") or (
|
||||
"capability" if p["_dir"] in CAPABILITY_PLUGINS else "integration")
|
||||
repos = _get_plugin_repos(settings)
|
||||
repos = plugin_repos(settings)
|
||||
# Config left behind by plugins that are no longer installed. Surfaced for an
|
||||
# explicit operator decision rather than cleaned up automatically: an external
|
||||
# plugin can be missing because /data/plugins isn't mounted or an install
|
||||
@@ -810,7 +795,7 @@ async def plugin_repos_add():
|
||||
form = await request.form
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
settings = await get_all_settings(db)
|
||||
repos = _get_plugin_repos(settings)
|
||||
repos = plugin_repos(settings)
|
||||
url = form.get("url", "").strip()
|
||||
repo_name = form.get("name", "").strip() or url
|
||||
if url:
|
||||
@@ -828,7 +813,7 @@ async def plugin_repos_add():
|
||||
async def plugin_repos_remove(idx: int):
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
settings = await get_all_settings(db)
|
||||
repos = _get_plugin_repos(settings)
|
||||
repos = plugin_repos(settings)
|
||||
if 0 <= idx < len(repos):
|
||||
repos.pop(idx)
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
@@ -845,7 +830,7 @@ async def plugin_repos_save(idx: int):
|
||||
form = await request.form
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
settings = await get_all_settings(db)
|
||||
repos = _get_plugin_repos(settings)
|
||||
repos = plugin_repos(settings)
|
||||
if 0 <= idx < len(repos):
|
||||
repos[idx] = {
|
||||
"name": form.get("name", "").strip() or repos[idx]["name"],
|
||||
@@ -864,41 +849,49 @@ async def plugin_repos_save(idx: int):
|
||||
@settings_bp.get("/plugins/catalog/")
|
||||
@require_role(UserRole.admin)
|
||||
async def plugins_catalog():
|
||||
"""HTMX partial: list plugins available in the remote catalog."""
|
||||
"""HTMX partial: the plugin catalog as the scheduled plugin_catalog_refresh
|
||||
task last fetched it. A page load never reaches a repository (Background
|
||||
Jobs ruling, 2026-10-08); Refresh does, because the operator asked."""
|
||||
return await _catalog_partial(fetch=False)
|
||||
|
||||
|
||||
@settings_bp.post("/plugins/catalog/refresh")
|
||||
@require_role(UserRole.admin)
|
||||
async def plugins_catalog_refresh():
|
||||
"""HTMX partial: fetch every configured repository now, then the catalog."""
|
||||
return await _catalog_partial(fetch=True)
|
||||
|
||||
|
||||
async def _catalog_partial(fetch: bool):
|
||||
from steward.core.plugin_index import cached_catalog, fetch_catalog
|
||||
from steward.core.plugin_manager import _LOADED_PLUGINS
|
||||
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
settings = await get_all_settings(db)
|
||||
repos = _get_plugin_repos(settings)
|
||||
repo_urls = [r["url"] for r in repos if r.get("url")]
|
||||
|
||||
if not repo_urls:
|
||||
urls = repo_urls(settings)
|
||||
if not urls:
|
||||
return await render_template(
|
||||
"settings/_catalog_partial.html",
|
||||
catalog=[],
|
||||
installed_names=set(),
|
||||
loaded_names=set(),
|
||||
catalog=[], installed_names=set(), installed_versions={}, loaded_names=set(),
|
||||
error="No plugin repositories configured.",
|
||||
)
|
||||
|
||||
from steward.core.plugin_index import fetch_catalog
|
||||
from steward.core.plugin_manager import _LOADED_PLUGINS
|
||||
|
||||
force = request.args.get("refresh") == "1"
|
||||
try:
|
||||
catalog = await fetch_catalog(repo_urls, force=force)
|
||||
error = None if catalog else "Catalog is empty or could not be fetched."
|
||||
except Exception as exc:
|
||||
catalog = []
|
||||
error = str(exc)
|
||||
if fetch:
|
||||
await fetch_catalog(urls)
|
||||
catalog, missing = cached_catalog(urls)
|
||||
error = None
|
||||
if missing:
|
||||
error = (f"Not fetched yet: {', '.join(missing)}. Steward fetches each repository within a "
|
||||
"minute of starting or of a change here, then hourly; Refresh fetches now.")
|
||||
elif not catalog:
|
||||
error = "The configured repositories list no plugins."
|
||||
|
||||
discovered = _discover_plugins()
|
||||
installed_names = {p["_dir"] for p in discovered}
|
||||
installed_versions = {p["_dir"]: p.get("version", "0.0.0") for p in discovered}
|
||||
|
||||
return await render_template(
|
||||
"settings/_catalog_partial.html",
|
||||
catalog=catalog,
|
||||
installed_names=installed_names,
|
||||
installed_versions=installed_versions,
|
||||
installed_names={p["_dir"] for p in discovered},
|
||||
installed_versions={p["_dir"]: p.get("version", "0.0.0") for p in discovered},
|
||||
loaded_names=_LOADED_PLUGINS,
|
||||
error=error,
|
||||
)
|
||||
@@ -920,8 +913,7 @@ async def install_plugin(name: str):
|
||||
|
||||
async with current_app.db_sessionmaker() as db:
|
||||
settings = await get_all_settings(db)
|
||||
repo_urls = [r["url"] for r in _get_plugin_repos(settings) if r.get("url")]
|
||||
entry = next((p for p in await fetch_catalog(repo_urls) if p.name == name), None)
|
||||
entry = next((p for p in await fetch_catalog(repo_urls(settings)) if p.name == name), None)
|
||||
if entry is None:
|
||||
return await render_template(
|
||||
"settings/_install_result.html",
|
||||
|
||||
@@ -160,7 +160,7 @@
|
||||
<div style="display:flex;align-items:center;justify-content:space-between;margin-bottom:0.75rem;">
|
||||
<div class="section-title">Plugin Catalog</div>
|
||||
<button class="btn btn-ghost btn-sm" style="font-size:0.78rem;"
|
||||
hx-get="/settings/plugins/catalog/?refresh=1"
|
||||
hx-post="/settings/plugins/catalog/refresh"
|
||||
hx-target="#plugin-catalog"
|
||||
hx-swap="outerHTML">
|
||||
Refresh
|
||||
|
||||
@@ -101,7 +101,7 @@ async def test_route_takes_url_and_checksum_from_the_catalog(client):
|
||||
checksum_sha256="abc")
|
||||
install = AsyncMock(return_value=(False, "stopped here"))
|
||||
with patch("steward.settings.routes.get_all_settings", new=AsyncMock(return_value={})), \
|
||||
patch("steward.settings.routes._get_plugin_repos", return_value=[{"url": "https://catalog/"}]), \
|
||||
patch("steward.settings.routes.repo_urls", return_value=["https://catalog/"]), \
|
||||
patch("steward.core.plugin_index.fetch_catalog", new=AsyncMock(return_value=[entry])), \
|
||||
patch("steward.core.plugin_manager.download_and_install_plugin", new=install), \
|
||||
patch("steward.settings.routes.log_audit", new=AsyncMock()):
|
||||
@@ -114,3 +114,64 @@ async def test_route_takes_url_and_checksum_from_the_catalog(client):
|
||||
resp = await client.post("/settings/plugins/install/other", headers={"X-CSRF-Token": "t"})
|
||||
assert "No configured plugin repository lists this plugin" in await resp.get_data(as_text=True)
|
||||
install.assert_not_awaited()
|
||||
|
||||
|
||||
# ── The catalog is fetched by a scheduled task, never by a page load ─────────
|
||||
# Background Jobs ruling (operator, 2026-10-08): a Steward page reads stored
|
||||
# records; the sources are contacted by scheduled collection, or because the
|
||||
# operator asked (Refresh, install).
|
||||
|
||||
from steward.core import plugin_index as pi # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fresh_catalog():
|
||||
pi.clear_catalog_cache()
|
||||
yield
|
||||
pi.clear_catalog_cache()
|
||||
|
||||
|
||||
def _fetcher(fetched, fail=()):
|
||||
async def fetch(url):
|
||||
fetched.append(url)
|
||||
if url in fail:
|
||||
pi._failed_at[url] = 0.0
|
||||
return False
|
||||
pi._url_cache[url] = ({"plugins": [{"name": f"p-{url[-1]}", "version": "1.0"}]}, 0.0)
|
||||
return True
|
||||
return fetch
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_tick_fetches_what_is_due_and_waits_out_a_failure(fresh_catalog):
|
||||
fetched = []
|
||||
with patch.object(pi, "_fetch_one", new=_fetcher(fetched, fail={"https://r/b"})):
|
||||
assert await pi.refresh_due(["https://r/a", "https://r/b"], now=10.0) == 1
|
||||
assert fetched == ["https://r/a", "https://r/b"]
|
||||
fetched.clear()
|
||||
# a fresh copy is left alone; the failure is not retried before RETRY_AFTER
|
||||
assert await pi.refresh_due(["https://r/a", "https://r/b"], now=pi.RETRY_AFTER - 1.0) == 0
|
||||
assert fetched == []
|
||||
assert await pi.refresh_due(["https://r/a", "https://r/b"], now=pi.RETRY_AFTER + 1.0) == 0
|
||||
assert fetched == ["https://r/b"]
|
||||
fetched.clear()
|
||||
assert await pi.refresh_due(["https://r/a"], now=pi.CATALOG_MAX_AGE + 1.0) == 1
|
||||
assert fetched == ["https://r/a"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_catalog_page_reads_the_cache_and_only_refresh_fetches(client, fresh_catalog):
|
||||
async with client.session_transaction() as sess:
|
||||
sess.update(user_id="u", user_role="admin", username="admin", _csrf="t")
|
||||
fetched = []
|
||||
with patch("steward.settings.routes.get_all_settings", new=AsyncMock(return_value={})), \
|
||||
patch("steward.settings.routes.repo_urls", return_value=["https://r/a"]), \
|
||||
patch.object(pi, "_fetch_one", new=_fetcher(fetched)):
|
||||
page = await (await client.get("/settings/plugins/catalog/")).get_data(as_text=True)
|
||||
assert fetched == [] and "Not fetched yet: https://r/a" in page
|
||||
page = await (await client.post("/settings/plugins/catalog/refresh",
|
||||
headers={"X-CSRF-Token": "t"})).get_data(as_text=True)
|
||||
assert fetched == ["https://r/a"] and "p-a" in page and "Not fetched yet" not in page
|
||||
# the next page load shows the stored copy without fetching again
|
||||
page = await (await client.get("/settings/plugins/catalog/")).get_data(as_text=True)
|
||||
assert fetched == ["https://r/a"] and "p-a" in page
|
||||
|
||||
Reference in New Issue
Block a user