diff --git a/repeater/plugins/ipc.py b/repeater/plugins/ipc.py index 8d8a3e0..c3ab3d9 100644 --- a/repeater/plugins/ipc.py +++ b/repeater/plugins/ipc.py @@ -131,6 +131,7 @@ class PluginIPCServer: "catalogue_install": self._op_catalogue_install, "check_update": self._op_check_update, "update": self._op_update, + "progress": self._op_progress, "ping": self._op_ping, } @@ -367,6 +368,14 @@ class PluginIPCServer: force = bool(request.get("force_refresh") or request.get("refresh")) return self.manager.check_update(plugin_id, force_refresh=force) + def _op_progress(self, request: dict[str, Any]) -> dict[str, Any]: + plugin_id = self._require_id(request) + try: + since = int(request.get("since") or 0) + except (TypeError, ValueError): + raise PluginIPCError("since must be an integer", 400) + return self.manager.progress(plugin_id, since=since) + def _op_update(self, request: dict[str, Any]) -> dict[str, Any]: plugin_id = self._require_id(request) version = request.get("version") @@ -509,6 +518,9 @@ class PluginIPCClient: payload["version"] = version return self.call("catalogue_install", **payload) + def progress(self, plugin_id: str, *, since: int = 0) -> dict[str, Any]: + return self.call("progress", plugin_id=plugin_id, since=int(since)) + def check_update(self, plugin_id: str, *, force_refresh: bool = False) -> dict[str, Any]: return self.call("check_update", plugin_id=plugin_id, force_refresh=bool(force_refresh)) diff --git a/repeater/plugins/manager.py b/repeater/plugins/manager.py index 6bf529e..6f961ef 100644 --- a/repeater/plugins/manager.py +++ b/repeater/plugins/manager.py @@ -31,6 +31,77 @@ from .storage import PluginStorage logger = logging.getLogger("PluginManager") +PROGRESS_MAX_LINES = 500 +PROGRESS_MAX_LOGS = 32 + + +class OperationProgress: + """Bounded output log of a plugin's last install or update.""" + + def __init__(self, plugin_id: str, operation: str): + self.plugin_id = plugin_id + self.operation = operation + self.state = "running" + self.error: Optional[str] = None + self.started = time.time() + self.finished: Optional[float] = None + self._lines: list[str] = [] + self._dropped = 0 # cursors count every line ever appended + self._lock = threading.Lock() + + def append(self, line: str) -> None: + with self._lock: + self._lines.append(line) + excess = len(self._lines) - PROGRESS_MAX_LINES + if excess > 0: + del self._lines[:excess] + self._dropped += excess + + def finish(self, error: Optional[str] = None) -> None: + with self._lock: + self.state = "error" if error else "complete" + self.error = error + self.finished = time.time() + + def snapshot(self, since: int = 0) -> dict[str, Any]: + with self._lock: + start = max(0, min(int(since) - self._dropped, len(self._lines))) + return { + "id": self.plugin_id, + "operation": self.operation, + "state": self.state, + "error": self.error, + "lines": list(self._lines[start:]), + "next": self._dropped + len(self._lines), + "started": self.started, + "finished": self.finished, + } + + +def _reported(operation: str): + """Record the operation's installer output and outcome for /plugins/progress.""" + + def decorate(fn): + @wraps(fn) + def wrapped(self, plugin_id, *args, **kwargs): + progress = self._start_progress(plugin_id, operation) + previous = self.runtime.on_output + self.runtime.on_output = progress.append + try: + result = fn(self, plugin_id, *args, **kwargs) + except Exception as exc: + progress.finish(str(exc)) + raise + finally: + self.runtime.on_output = previous + progress.finish() + return result + + return wrapped + + return decorate + + def _serialized_plugin(fn): """Keep a complete lifecycle transaction exclusive, without queuing callers.""" @@ -66,6 +137,7 @@ class PluginManager: self.runtime = runtime or PluginRuntime(storage) self._lock = threading.RLock() self._operations: dict[str, tuple[int, int]] = {} + self._progress: dict[str, OperationProgress] = {} self.catalogue = catalogue_client or CatalogueClient(catalogue_url or DEFAULT_CATALOGUE_URL) self.github = github_client or GitHubReleaseClient() @@ -87,6 +159,34 @@ class PluginManager: else: self._operations[plugin_id] = (owner, depth - 1) + def _start_progress(self, plugin_id: str, operation: str) -> OperationProgress: + progress = OperationProgress(plugin_id, operation) + with self._lock: + self._progress[plugin_id] = progress + finished = sorted( + (p for p in self._progress.values() if p.state != "running"), + key=lambda p: p.finished or 0, + ) + for old in finished[: max(0, len(self._progress) - PROGRESS_MAX_LOGS)]: + del self._progress[old.plugin_id] + return progress + + def progress(self, plugin_id: str, since: int = 0) -> dict[str, Any]: + with self._lock: + progress = self._progress.get(plugin_id) + if progress is None: + return { + "id": plugin_id, + "operation": None, + "state": "idle", + "error": None, + "lines": [], + "next": 0, + "started": None, + "finished": None, + } + return progress.snapshot(since) + @contextmanager def stage_upload(self, wheel_path: str): """Own a bounded regular-file copy before IPC acknowledges processing. @@ -430,6 +530,7 @@ class PluginManager: return {"schema": catalogue.schema, "plugins": plugins_out} @_serialized_plugin + @_reported("install") def install_from_catalogue( self, plugin_id: str, @@ -568,6 +669,7 @@ class PluginManager: } @_serialized_plugin + @_reported("update") def update_plugin( self, plugin_id: str, diff --git a/repeater/plugins/runtime.py b/repeater/plugins/runtime.py index 65a949c..895703d 100644 --- a/repeater/plugins/runtime.py +++ b/repeater/plugins/runtime.py @@ -72,6 +72,7 @@ class PluginRuntime: self.crash_max_exits = crash_max_exits self._popen = popen_factory or subprocess.Popen self._run = run_factory or self._run_install_command + self.on_output: Optional[Callable[[str], None]] = None self._lock = threading.RLock() self._handles: dict[str, ProcessHandle] = {} self._runtime_state: dict[str, PluginState] = {} @@ -232,6 +233,13 @@ class PluginRuntime: rebuild_marker.unlink() shutil.rmtree(backup) + def _announce(self, line: str) -> None: + if self.on_output is not None: + try: + self.on_output(line) + except Exception: + logger.debug("install output listener failed", exc_info=True) + def _run_install_command(self, cmd, *, timeout, **kwargs): """Drain output into bounded memory and kill the entire installer group.""" # Callers construct Python/venv/pip argv; installing trusted wheels is intentional. @@ -245,12 +253,25 @@ class PluginRuntime: "installer", proc, None, process_group=proc.pid if os.name == "posix" else None ) output = bytearray() + pending = bytearray() + + def emit(line: bytes) -> None: + self._announce(line.decode("utf-8", errors="replace").rstrip("\r")) def drain(): with proc.stdout as source: while chunk := source.read1(64 * 1024): output.extend(chunk) del output[:-INSTALL_OUTPUT_MAX_BYTES] + if self.on_output is not None: + pending.extend(chunk) + *lines, rest = pending.split(b"\n") + pending[:] = rest + del pending[:-INSTALL_OUTPUT_MAX_BYTES] + for line in lines: + emit(line) + if pending: + emit(bytes(pending)) reader = threading.Thread(target=drain, name="plugin-install-output", daemon=True) reader.start() @@ -271,6 +292,7 @@ class PluginRuntime: if py.is_file(): return venv_path.parent.mkdir(parents=True, exist_ok=True) + self._announce(f"Creating environment: {venv_path.name}") result = self._run( [self.python_executable, "-m", "venv", str(venv_path)], timeout=VENV_TIMEOUT_SECONDS, @@ -288,6 +310,7 @@ class PluginRuntime: if not py.is_file(): raise RuntimeError(f"venv python missing: {py}") install_path = self._ensure_pip_wheel_filename(wheel_path) + self._announce(f"Installing {wheel_path.name} with pip") try: result = self._run( [str(py), "-m", "pip", "install", "--upgrade", str(install_path)], diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index cba364a..dd527b6 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -4701,6 +4701,50 @@ paths: description: Updated plugin status '400': description: Update unavailable or invalid + /plugins/progress: + get: + tags: [Plugins] + summary: Stream an install or update's progress + description: | + Server-sent events for a plugin's last catalogue install or update, as + /update/progress streams the repeater's own updater. JSON events: + `connected`, `line`, `status` (idle, running, complete, error), + `keepalive`, and one `done` that ends the stream. Idle streams end after + a minute. Uploaded wheels are not reported. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + parameters: + - name: id + in: query + required: true + schema: + type: string + - name: since + in: query + required: false + description: Resume from this line index + schema: + type: integer + default: 0 + - name: fresh + in: query + required: false + description: Ignore a log already finished when the stream opens; wait for the next operation. + schema: + type: boolean + default: false + responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' + '200': + description: Event stream + content: + text/event-stream: + schema: + type: string + '503': + description: Plugin manager unavailable /plugins/settings: get: tags: [Plugins] diff --git a/repeater/web/plugin_endpoints.py b/repeater/web/plugin_endpoints.py index 90613be..f96230e 100644 --- a/repeater/web/plugin_endpoints.py +++ b/repeater/web/plugin_endpoints.py @@ -2,10 +2,12 @@ from __future__ import annotations +import json import logging import re import shutil import tempfile +import time from pathlib import Path from typing import Any, Optional @@ -22,6 +24,10 @@ from repeater.plugins.storage import resolve_plugin_socket_path, resolve_plugins logger = logging.getLogger("HTTPServer") +PROGRESS_POLL_SECONDS = 0.75 +PROGRESS_MAX_SECONDS = 900.0 # the IPC completion budget +PROGRESS_IDLE_SECONDS = 60.0 + class PluginAPIEndpoints: """CherryPy nested app for /api/plugins/*.""" @@ -306,6 +312,102 @@ class PluginAPIEndpoints: raise cherrypy.HTTPError(400, "tail must be an integer") return self._handle_ipc(lambda: self._client_or_raise().logs(plugin_id, tail=tail)) + @cherrypy.expose + def progress(self, **kwargs): + """GET /api/plugins/progress?id=&since=&fresh= — SSE of an install or update.""" + if cherrypy.request.method == "OPTIONS": + return "" + if cherrypy.request.method not in ("GET", "HEAD"): + raise cherrypy.HTTPError(405) + plugin_id = self._plugin_id_from(kwargs) + try: + since = int(kwargs.get("since", 0)) + except (TypeError, ValueError): + raise cherrypy.HTTPError(400, "since must be an integer") + fresh = str(kwargs.get("fresh", "")).strip().lower() in {"1", "true", "yes", "on"} + cherrypy.response.headers["Content-Type"] = "text/event-stream" + if cherrypy.request.method == "HEAD": + return "" + cherrypy.response.headers["Cache-Control"] = "no-cache" + cherrypy.response.headers["X-Accel-Buffering"] = "no" + cherrypy.response.headers["Connection"] = "keep-alive" + return self._progress_events(plugin_id, since, fresh=fresh) + + progress._cp_config = {"response.stream": True} + + def _progress_events( + self, + plugin_id: str, + since: int, + *, + fresh: bool = False, + sleep=time.sleep, + clock=time.monotonic, + ): + def event(payload: dict) -> str: + return f"data: {json.dumps(payload)}\n\n" + + def generate(): + yield event({"type": "connected", "id": plugin_id}) + cursor, last_state, quiet, stale = since, None, 0, None + opened = clock() + while True: + try: + snap = self._client_or_raise().progress(plugin_id, since=cursor) + except GeneratorExit: + return + except Exception as exc: + yield event({"type": "done", "state": "error", "error": str(exc)}) + return + + state, started = snap.get("state"), snap.get("started") + # fresh: a log already finished when the stream opened is not this operation's + if fresh and last_state is None and state in ("complete", "error"): + stale = started + if stale is not None and started == stale: + state, snap = "idle", {"next": cursor, "operation": None, "started": None} + + changed = False + for line in snap.get("lines") or []: + yield event({"type": "line", "line": line}) + changed = True + cursor = int(snap.get("next") or cursor) + if state != last_state: + yield event( + { + "type": "status", + "state": state, + "operation": snap.get("operation"), + "started": snap.get("started"), + } + ) + last_state, changed = state, True + if state in ("complete", "error"): + yield event( + { + "type": "done", + "state": state, + "error": snap.get("error"), + "started": snap.get("started"), + } + ) + return + now = clock() + if now >= opened + PROGRESS_MAX_SECONDS: + yield event( + {"type": "done", "state": "timeout", "error": "progress stream timed out"} + ) + return + if state == "idle" and now >= opened + PROGRESS_IDLE_SECONDS: + yield event({"type": "done", "state": "idle", "error": None}) + return + quiet = 0 if changed else quiet + 1 + if quiet and quiet % 4 == 0: + yield event({"type": "keepalive"}) + sleep(PROGRESS_POLL_SECONDS) + + return generate() + @cherrypy.expose @cherrypy.tools.json_out() @cherrypy.tools.json_in(force=False) diff --git a/tests/test_plugin_progress.py b/tests/test_plugin_progress.py new file mode 100644 index 0000000..3c875fb --- /dev/null +++ b/tests/test_plugin_progress.py @@ -0,0 +1,166 @@ +import json +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from repeater.plugins.ipc import PluginIPCClient, PluginIPCServer +from repeater.plugins.manager import ( + PROGRESS_MAX_LINES, + PROGRESS_MAX_LOGS, + OperationProgress, + PluginManager, + PluginManagerError, +) +from repeater.plugins.runtime import PluginRuntime +from repeater.plugins.storage import PluginStorage +from repeater.web.plugin_endpoints import PluginAPIEndpoints + + +def test_runtime_hands_each_output_line_to_the_listener(tmp_path: Path): + runtime = PluginRuntime(PluginStorage(tmp_path / "plugins")) + heard: list[str] = [] + runtime.on_output = heard.append + script = "import sys; print('Collecting demo'); print('Installing', end=''); sys.stdout.flush()" + result = runtime._run_install_command([sys.executable, "-c", script], timeout=30) + assert result.returncode == 0 + assert heard == ["Collecting demo", "Installing"] + assert "Collecting demo" in result.stdout + + def broken(_line: str) -> None: + raise RuntimeError("listener bug") + + runtime.on_output = broken + assert ( + runtime._run_install_command([sys.executable, "-c", "print('ok')"], timeout=30).returncode + == 0 + ) + + +def test_progress_log_is_bounded_and_pages_by_cursor(): + log = OperationProgress("openhop.demo", "update") + for i in range(PROGRESS_MAX_LINES + 20): + log.append(f"line {i}") + snap = log.snapshot() + assert len(snap["lines"]) == PROGRESS_MAX_LINES + assert snap["lines"][0] == "line 20" + assert snap["next"] == PROGRESS_MAX_LINES + 20 + log.append("tail") + assert log.snapshot(since=snap["next"])["lines"] == ["tail"] + log.finish("boom") + assert (log.snapshot()["state"], log.snapshot()["error"]) == ("error", "boom") + + +def test_manager_records_the_outcome_and_restores_the_listener(tmp_path: Path): + mgr = PluginManager(PluginStorage(tmp_path / "plugins")) + assert mgr.progress("missing.plugin")["state"] == "idle" + with pytest.raises(PluginManagerError): + mgr.update_plugin("missing.plugin") + snap = mgr.progress("missing.plugin") + assert (snap["operation"], snap["state"]) == ("update", "error") + assert "plugin not found" in snap["error"] + assert mgr.runtime.on_output is None + + +def test_manager_keeps_a_bounded_number_of_logs(tmp_path: Path): + mgr = PluginManager(PluginStorage(tmp_path / "plugins")) + for i in range(PROGRESS_MAX_LOGS + 10): + with pytest.raises(PluginManagerError): + mgr.update_plugin(f"nobody.{i:03d}") + assert len(mgr._progress) == PROGRESS_MAX_LOGS + assert mgr.progress("nobody.000")["state"] == "idle" + + +def test_ipc_round_trips_progress(tmp_path: Path): + storage = PluginStorage(tmp_path / "plugins") + manager = PluginManager(storage, PluginRuntime(storage, popen_factory=lambda *a, **k: None)) + sock_path = Path(f"/tmp/oh-pm-progress-{tmp_path.name}.sock") + server = PluginIPCServer(sock_path, manager) + server.start() + try: + client = PluginIPCClient(sock_path) + assert client.progress("openhop.demo")["state"] == "idle" + with pytest.raises(PluginManagerError): + manager.update_plugin("openhop.demo") + assert client.progress("openhop.demo", since=0)["state"] == "error" + finally: + server.stop() + + +def _snap(state, lines=(), nxt=0, started=None, error=None): + return { + "state": state, + "operation": "update", + "lines": list(lines), + "next": nxt, + "error": error, + "started": started, + } + + +def _events(tmp_path, snapshots, *, ticks=None, fresh=False): + api = PluginAPIEndpoints({"storage": {"storage_dir": str(tmp_path)}}) + answers = iter(snapshots) + client = SimpleNamespace(progress=lambda _id, since=0: next(answers, snapshots[-1])) + clock = iter(ticks or [0.0] * 1000) + with patch.object(api, "_client_or_raise", return_value=client): + chunks = api._progress_events( + "openhop.demo", 0, fresh=fresh, sleep=lambda _s: None, clock=lambda: next(clock) + ) + return [json.loads(chunk[len("data: ") : -2]) for chunk in chunks] + + +def test_stream_plays_lines_status_and_one_done(tmp_path): + events = _events( + tmp_path, + [_snap("running", ["a"], 1), _snap("running", ["b", "c"], 3), _snap("complete", nxt=3)], + ) + assert [e["type"] for e in events] == [ + "connected", + "line", + "status", + "line", + "line", + "status", + "done", + ] + assert [e["line"] for e in events if e["type"] == "line"] == ["a", "b", "c"] + assert events[-1] == {"type": "done", "state": "complete", "error": None, "started": None} + + events = _events(tmp_path, [_snap("error", ["pip failed"], 1, error="pip")]) + assert events[-1]["state"] == "error" and events[-1]["error"] == "pip" + + +def test_idle_stream_keeps_alive_then_ends(tmp_path): + events = _events(tmp_path, [_snap("idle")], ticks=[0.0] + [1.0] * 12 + [100.0]) + types = [e["type"] for e in events] + assert types[:2] == ["connected", "status"] + assert types.count("keepalive") == 2 + assert events[-1] == {"type": "done", "state": "idle", "error": None} + + events = _events(tmp_path, [_snap("running", started=5.0)], ticks=[0.0, 100.0, 500.0, 901.0]) + assert events[-1]["state"] == "timeout" + + +def test_fresh_stream_ignores_a_previous_operations_log(tmp_path): + previous = _snap("complete", ["old"], 1, started=100.0) + current = _snap("running", ["new"], 1, started=200.0) + events = _events( + tmp_path, [previous, previous, current, _snap("complete", nxt=1, started=200.0)], fresh=True + ) + assert [e["type"] for e in events] == [ + "connected", + "status", + "line", + "status", + "status", + "done", + ] + assert events[1]["state"] == "idle" + assert events[2]["line"] == "new" + assert events[-1]["started"] == 200.0 + + events = _events(tmp_path, [previous]) + assert [e["type"] for e in events] == ["connected", "line", "status", "done"]