feat(serve): hot-reload service repos on a timer without a restart

Add serve.services_refresh_interval; refresh pulls the repos and re-imports only the changed services, staging any with active jobs until they finish. Report skipped/broken services in the log and on /api/services.
This commit is contained in:
imSp4rky
2026-08-26 23:14:23 -06:00
parent 4964869258
commit 5128fef473
11 changed files with 350 additions and 52 deletions
+12 -4
View File
@@ -62,7 +62,7 @@ If the update check fails (for example, no network), `update_available` and `lat
### `GET /api/services` ### `GET /api/services`
Show the streaming services available on this server, filtered by your allowlist. Each entry gives the service's tag, matching rules, capability flags, and its CLI parameters (useful when you make a UI that accepts service-specific options). Show the streaming services available on this server, filtered by your allowlist. Each entry gives the service's tag, matching rules, capability flags, and its CLI parameters (useful when you make a UI that accepts service-specific options). `load_errors` lists the services the server skipped because they failed to import, at startup or after a service repository refresh; the server also writes each one to its log. An entry marked `pending_update` has newer code staged: it keeps its current code until its running or queued jobs finish.
=== "Request" === "Request"
@@ -100,7 +100,8 @@ Show the streaming services available on this server, filtered by your allowlist
"has_drm": true, "has_drm": true,
"auth_methods": ["cookies"] "auth_methods": ["cookies"]
} }
] ],
"load_errors": []
} }
``` ```
@@ -118,6 +119,7 @@ Field notes:
| `needs_auth` | Whether the service overrides `authenticate()`. | | `needs_auth` | Whether the service overrides `authenticate()`. |
| `has_search` | Whether the service supports `POST /api/search`. | | `has_search` | Whether the service supports `POST /api/search`. |
| `has_drm` | Whether the service overrides a Widevine or PlayReady license hook. | | `has_drm` | Whether the service overrides a Widevine or PlayReady license hook. |
| `pending_update` | Set to `true` while a service repository refresh waits for this service's jobs to finish before it swaps in the new code. Absent otherwise. |
| `auth_methods` | Declared or inferred auth methods, e.g. `["cookies"]`, `["credentials"]`. | | `auth_methods` | Declared or inferred auth methods, e.g. `["cookies"]`, `["credentials"]`. |
| Status | Meaning | | Status | Meaning |
@@ -773,7 +775,7 @@ The `409` guard is about active file I/O, not an arbitrary lock. An in-flight jo
### `POST /api/maintenance/refresh-services` ### `POST /api/maintenance/refresh-services`
git-pull every service repository configured under `directories.services`. `refreshed` becomes `true` when all repos updated successfully, or when you configure no repository. git-pull every service repository configured under `directories.services` and re-import the services that changed, without a restart. `refreshed` becomes `true` when all repos updated successfully, or when you configure no repository. `deferred` lists the services that have a running or queued job: they keep their current code until that job finishes, then swap. `load_errors` lists the services whose new code failed to import; the server drops them until the next successful refresh.
=== "Response `200`" === "Response `200`"
@@ -781,7 +783,13 @@ git-pull every service repository configured under `directories.services`. `refr
{ {
"refreshed": true, "refreshed": true,
"repos": [ "repos": [
{ "spec": "github.com/example/services", "updated": true, "changes": ["~EXAMPLE"] } {
"spec": "github.com/example/services",
"updated": true,
"changes": ["~EXAMPLE", "+EXAMPLE2"],
"deferred": ["EXAMPLE"],
"load_errors": []
}
] ]
} }
``` ```
+1 -1
View File
@@ -183,7 +183,7 @@ request and response in full.
|---|---|---| |---|---|---|
| `POST` | `/api/maintenance/clear-cache` | Clear the cache directory. | | `POST` | `/api/maintenance/clear-cache` | Clear the cache directory. |
| `POST` | `/api/maintenance/clear-temp` | Clear the temp directory. | | `POST` | `/api/maintenance/clear-temp` | Clear the temp directory. |
| `POST` | `/api/maintenance/refresh-services` | `git pull` each configured service repository. | | `POST` | `/api/maintenance/refresh-services` | `git pull` each configured service repository and re-import the services that changed. |
### Remote-download sessions ### Remote-download sessions
+1 -1
View File
@@ -550,7 +550,7 @@ unshackle util COMMAND [ARGS]...
unshackle util refresh-services unshackle util refresh-services
``` ```
Force a refresh (git pull / hard reset) of all service repos configured under `directories.services`. Force a refresh (git pull / hard reset) of all service repos configured under `directories.services`. This updates the clones on disk. A server that is already running reloads the changed services through `POST /api/maintenance/refresh-services` or `serve.services_refresh_interval` instead.
### `util crop PATH ASPECT` ### `util crop PATH ASPECT`
@@ -76,6 +76,14 @@ directories:
services_repo_force: true services_repo_force: true
``` ```
!!! info "Refresh inside `serve`"
A running server does not restart to pick up a refresh. Both the periodic refresh
(`serve.services_refresh_interval`, see [serve](services.md#serve)) and the manual
`POST /api/maintenance/refresh-services` endpoint hard-reset the clones and then re-import
only the services that changed. A service with a running or queued download job keeps its
current code until that job finishes, then swaps; `GET /api/services` marks it with
`pending_update` in the meantime.
!!! warning "Read-only installs and reinstalls" !!! warning "Read-only installs and reinstalls"
If the services directory lives inside the installed package, a reinstall can delete the If the services directory lives inside the installed package, a reinstall can delete the
`_repos` clones. They are re-cloned on next use. On **read-only installs** you must `_repos` clones. They are re-cloned on next use. On **read-only installs** you must
+3 -1
View File
@@ -117,6 +117,7 @@ is the [REST API](../../dev/rest-api/index.md) section. These are the config key
| `max_sessions` | int | `100` | Maximum concurrent sessions. | | `max_sessions` | int | `100` | Maximum concurrent sessions. |
| `history_limit` | int | `100` | How many finished jobs to retain in history. | | `history_limit` | int | `100` | How many finished jobs to retain in history. |
| `compression_level` | int | `1` | gzip level for responses. | | `compression_level` | int | `1` | gzip level for responses. |
| `services_refresh_interval` | int (s) | `0` | How often the server pulls the git-backed service repositories in `directories.services` and hot-reloads the services that changed. `0` turns it off. A service with a running or queued job swaps to the new code as soon as its last job finishes. |
| `global_speed_limit` | str | *(unlimited)* | Server-wide download speed cap, e.g. `10M`, `1.5G` or plain bytes/sec (same format as `speed_limit`). One shared budget across all concurrent jobs; per-job speed limits are ignored while it is set. | | `global_speed_limit` | str | *(unlimited)* | Server-wide download speed cap, e.g. `10M`, `1.5G` or plain bytes/sec (same format as `speed_limit`). One shared budget across all concurrent jobs; per-job speed limits are ignored while it is set. |
| `cdm_overrides` | list or bool | *(unset)* | Allowed per-request CDM overrides: a list of permitted device names, or `true` for any. Unset rejects every override. | | `cdm_overrides` | list or bool | *(unset)* | Allowed per-request CDM overrides: a list of permitted device names, or `true` for any. Unset rejects every override. |
| `allow_job_credentials` | bool | `false` | Permit clients to supply credentials per job. | | `allow_job_credentials` | bool | `false` | Permit clients to supply credentials per job. |
@@ -141,11 +142,12 @@ serve:
api_secret: change-me api_secret: change-me
remote_only: true remote_only: true
services: [EXAMPLE1, EXAMPLE2] services: [EXAMPLE1, EXAMPLE2]
services_refresh_interval: 3600 # pull the service repositories hourly and reload what changed
# server-wide download defaults (same keys as the `dl:` block) # server-wide download defaults (same keys as the `dl:` block)
downloads: 3 downloads: 3
best_available: true best_available: true
users: users:
a1b2c3d4: # this user's API key a1b2c3d4: # this user's API key
services: [EXAMPLE1] # may only use EXAMPLE1 services: [EXAMPLE1] # may only use EXAMPLE1
e5f6a7b8: e5f6a7b8:
server_cdm: true # this key may have the server do the licensing server_cdm: true # this key may have the server do the licensing
+74
View File
@@ -0,0 +1,74 @@
"""Pins hot-reload of services: re-import on change, drop on removal, and staged swaps for busy services."""
from __future__ import annotations
import os
import shutil
import time
from pathlib import Path
import pytest
from unshackle.core import services
from unshackle.core.config import config
pytestmark = pytest.mark.unit
def write_service(root: Path, version: int) -> None:
pkg = root / "FOO"
pkg.mkdir(exist_ok=True)
init = pkg / "__init__.py"
init.write_text(f"class FOO:\n VERSION = {version}\n ALIASES = ('foo{version}',)\n")
# the bytecode cache keys on mtime and size; same-second rewrites of equal length would reuse the stale .pyc
os.utime(init, (time.time() + version, time.time() + version))
@pytest.fixture
def service_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
root = tmp_path / "svc"
root.mkdir()
write_service(root, 1)
monkeypatch.setattr(config.directories, "services", [str(root)])
monkeypatch.setattr(services, "SERVICES", [])
monkeypatch.setattr(services, "MODULES", {})
monkeypatch.setattr(services, "ALIASES", {})
monkeypatch.setattr(services, "LOAD_ERRORS", [])
monkeypatch.setattr(services, "PENDING", set())
assert services.reload_services(["FOO"]) == []
assert services.MODULES["FOO"].VERSION == 1
yield root
services.reload_services(["FOO"])
def test_reload_picks_up_new_code_and_removal(service_dir: Path):
write_service(service_dir, 2)
assert services.reload_services(["FOO"]) == []
assert services.MODULES["FOO"].VERSION == 2
assert services.ALIASES["FOO"] == ("foo2",)
assert services.Services.get_tag("foo2") == "FOO"
shutil.rmtree(service_dir / "FOO")
assert services.reload_services(["FOO"]) == []
assert "FOO" not in services.MODULES
assert "FOO" not in services.Services.get_tags()
def test_busy_service_is_staged_then_applied(service_dir: Path, monkeypatch: pytest.MonkeyPatch):
write_service(service_dir, 2)
monkeypatch.setattr(services, "repo_specs", lambda: ["example/repo"])
monkeypatch.setattr(services, "refresh_repo", lambda spec: (service_dir, ["~FOO"]))
repos = services.refresh_and_reload(busy={"FOO"})
assert repos == [
{"spec": "example/repo", "updated": True, "changes": ["~FOO"], "deferred": ["FOO"], "load_errors": []}
]
assert services.MODULES["FOO"].VERSION == 1
assert "FOO" in services.PENDING
assert services.apply_pending(busy={"FOO"}) == []
assert services.MODULES["FOO"].VERSION == 1
assert services.apply_pending(busy=set()) == ["FOO"]
assert services.MODULES["FOO"].VERSION == 2
assert services.PENDING == set()
+49
View File
@@ -1,6 +1,8 @@
import asyncio
import hmac import hmac
import logging import logging
import subprocess import subprocess
from contextlib import suppress
import click import click
from aiohttp import web from aiohttp import web
@@ -14,6 +16,51 @@ from unshackle.core.constants import context_settings
from unshackle.core.downloaders import format_speed, parse_speed_limit, set_speed_limit from unshackle.core.downloaders import format_speed, parse_speed_limit, set_speed_limit
def _install_service_refresh(app: web.Application) -> None:
"""Report service load issues, then periodically pull the service repos and hot-reload changed services."""
from unshackle.core import services
log = logging.getLogger("serve")
services.log_load_issues()
interval = int(config.serve.get("services_refresh_interval", 0) or 0)
if interval <= 0 or not services.repo_specs():
return
async def loop() -> None:
from unshackle.core.api.download_manager import get_download_manager
manager = get_download_manager()
while True:
await asyncio.sleep(interval)
try:
applied = await asyncio.to_thread(services.apply_pending, manager.busy_services())
if applied:
log.info(f"Services reloaded: {', '.join(applied)}")
for r in await asyncio.to_thread(services.refresh_and_reload, manager.busy_services()):
if r["changes"]:
log.info(f"Services refreshed {r['spec']}: {', '.join(r['changes'])}")
if r["deferred"]:
log.info(f"Services staged until their jobs finish: {', '.join(r['deferred'])}")
for err in r["load_errors"]:
log.error(f"Service reload failed: {err}")
except Exception:
log.exception("Service refresh failed")
async def start(_app: web.Application) -> None:
_app["service_refresh_task"] = asyncio.create_task(loop())
log.info(f"Service repos refresh every {interval}s")
async def stop(_app: web.Application) -> None:
task = _app.get("service_refresh_task")
if task:
task.cancel()
with suppress(asyncio.CancelledError):
await task
app.on_startup.append(start)
app.on_cleanup.append(stop)
@click.command( @click.command(
short_help="Serve your Local Widevine/PlayReady Devices and REST API for Remote Access.", short_help="Serve your Local Widevine/PlayReady Devices and REST API for Remote Access.",
context_settings=context_settings, context_settings=context_settings,
@@ -196,6 +243,7 @@ def serve(
app.on_startup.append(start_session_cleanup) app.on_startup.append(start_session_cleanup)
app.on_cleanup.append(stop_session_cleanup) app.on_cleanup.append(stop_session_cleanup)
_install_service_refresh(app)
setup_routes(app, remote_only=remote_only) setup_routes(app, remote_only=remote_only)
if not remote_only: if not remote_only:
@@ -289,6 +337,7 @@ def serve(
app.on_startup.append(start_session_cleanup) app.on_startup.append(start_session_cleanup)
app.on_cleanup.append(stop_session_cleanup) app.on_cleanup.append(stop_session_cleanup)
_install_service_refresh(app)
if serve_widevine: if serve_widevine:
app.on_startup.append(pywidevine_serve._startup) app.on_startup.append(pywidevine_serve._startup)
+23 -1
View File
@@ -710,10 +710,32 @@ class DownloadQueueManager:
self.put(queue, payload) self.put(queue, payload)
def publish_terminal(self, job: DownloadJob) -> None: def publish_terminal(self, job: DownloadJob) -> None:
"""Publish a job's final state, then close every listener's event stream.""" """Publish a job's final state, close every listener's event stream, then apply staged service reloads."""
self.publish(job, job.status.value) self.publish(job, job.status.value)
for queue in self._subscribers.pop(job.job_id, []): for queue in self._subscribers.pop(job.job_id, []):
self.put(queue, None) self.put(queue, None)
self.schedule_pending_reload()
def busy_services(self) -> set[str]:
"""Tags with a job that has not reached a terminal status; queued jobs count so none starts on half-swapped code."""
return {job.service for job in self._jobs.values() if job.status not in TERMINAL_STATUSES}
def schedule_pending_reload(self) -> None:
"""Swap in staged service updates whose last job has finished, off the event loop."""
from unshackle.core import services
if not services.PENDING:
return
async def run() -> None:
try:
applied = await asyncio.to_thread(services.apply_pending, self.busy_services())
if applied:
log.info(f"Services reloaded after job completion: {', '.join(applied)}")
except Exception:
log.exception("Applying pending service reloads failed")
asyncio.get_running_loop().create_task(run())
@staticmethod @staticmethod
def put(queue: asyncio.Queue, item: Optional[Dict[str, Any]]) -> None: def put(queue: asyncio.Queue, item: Optional[Dict[str, Any]]) -> None:
+5 -12
View File
@@ -1796,20 +1796,13 @@ async def clear_temp_handler(request: Optional[web.Request] = None) -> web.Respo
async def refresh_services_handler(request: Optional[web.Request] = None) -> web.Response: async def refresh_services_handler(request: Optional[web.Request] = None) -> web.Response:
"""Answer the request to refresh the service repos configured in directories.services.""" """Refresh the service repos configured in directories.services and reload the changed services."""
from unshackle.core.service_repo import is_repo_spec, refresh_repo from unshackle.core.api.download_manager import get_download_manager
from unshackle.core.services import refresh_and_reload
try: try:
entries = config.directories.services busy = get_download_manager().busy_services()
if not isinstance(entries, list): repos = await asyncio.to_thread(refresh_and_reload, busy)
entries = [entries]
specs = [e for e in entries if isinstance(e, str) and is_repo_spec(e)]
repos = []
for spec in specs:
dest, changes = await asyncio.to_thread(refresh_repo, spec)
repos.append({"spec": spec, "updated": dest is not None, "changes": list(changes or [])})
return web.json_response({"refreshed": all(r["updated"] for r in repos), "repos": repos}) return web.json_response({"refreshed": all(r["updated"] for r in repos), "repos": repos})
except APIError: except APIError:
+28 -4
View File
@@ -9,6 +9,7 @@ from aiohttp import web
from aiohttp_swagger3 import SwaggerDocs, SwaggerInfo, SwaggerUiSettings from aiohttp_swagger3 import SwaggerDocs, SwaggerInfo, SwaggerUiSettings
from unshackle.core import __code_hash__, __version__ from unshackle.core import __code_hash__, __version__
from unshackle.core import services as services_module
from unshackle.core.api.errors import APIError, APIErrorCode, build_error_response, handle_api_exception from unshackle.core.api.errors import APIError, APIErrorCode, build_error_response, handle_api_exception
from unshackle.core.api.handlers import ( from unshackle.core.api.handlers import (
CORS_HEADERS, CORS_HEADERS,
@@ -150,6 +151,11 @@ async def services(request: web.Request) -> web.Response:
schema: schema:
type: object type: object
properties: properties:
load_errors:
type: array
description: Services skipped because they failed to import (startup or repo refresh)
items:
type: string
services: services:
type: array type: array
items: items:
@@ -157,6 +163,9 @@ async def services(request: web.Request) -> web.Response:
properties: properties:
tag: tag:
type: string type: string
pending_update:
type: boolean
description: Present (true) while a repo refresh waits for this service's jobs to finish
aliases: aliases:
type: array type: array
items: items:
@@ -221,6 +230,8 @@ async def services(request: web.Request) -> web.Response:
"url": None, "url": None,
"help": None, "help": None,
} }
if tag in services_module.PENDING:
service_data["pending_update"] = True
try: try:
service_module = Services.load(tag) service_module = Services.load(tag)
@@ -322,7 +333,7 @@ async def services(request: web.Request) -> web.Response:
services_info.append(service_data) services_info.append(service_data)
return web.json_response({"services": services_info}) return web.json_response({"services": services_info, "load_errors": list(services_module.LOAD_ERRORS)})
except Exception as e: except Exception as e:
log.exception("Error listing services") log.exception("Error listing services")
debug_mode = request.app.get("debug_api", False) debug_mode = request.app.get("debug_api", False)
@@ -1397,9 +1408,12 @@ async def maintenance_refresh_services(request: web.Request) -> web.Response:
--- ---
summary: Refresh service repos summary: Refresh service repos
description: > description: >
Force-sync (git pull) every service repo configured in directories.services. Force-sync (git pull) every service repo configured in directories.services and
`refreshed` is true when all repos synced, or when you configured no repos. The re-import the services that changed, without a restart. `refreshed` is true when all
`repos` field holds the per-repo results. repos synced, or when you configured no repos. The `repos` field holds the per-repo
results: `deferred` names the services that keep their current code until their
running or queued jobs finish, and `load_errors` names the services whose new code
failed to import.
responses: responses:
'200': '200':
description: Refresh results description: Refresh results
@@ -1423,6 +1437,16 @@ async def maintenance_refresh_services(request: web.Request) -> web.Response:
type: array type: array
items: items:
type: string type: string
deferred:
type: array
description: Services staged until their running or queued jobs finish
items:
type: string
load_errors:
type: array
description: Services whose new code failed to import
items:
type: string
'500': '500':
description: Server error description: Server error
""" """
+146 -28
View File
@@ -3,48 +3,66 @@ from __future__ import annotations
import inspect import inspect
import logging import logging
import re import re
import sys
import threading
from pathlib import Path from pathlib import Path
from typing import Any, Iterable
import click import click
from unshackle.core.config import config from unshackle.core.config import config
from unshackle.core.service import Service from unshackle.core.service import Service
from unshackle.core.service_repo import DirtyServiceRepo, is_repo_spec, resolve_service_repo from unshackle.core.service_repo import DirtyServiceRepo, is_repo_spec, refresh_repo, resolve_service_repo
from unshackle.core.utilities import import_module_by_path from unshackle.core.utilities import import_module_by_path
from unshackle.core.utils.redact import redact_path from unshackle.core.utils.redact import redact_path
log = logging.getLogger("services") log = logging.getLogger("services")
raw = config.directories.services
if not isinstance(raw, list):
raw = [raw]
# resolve in config order - priority IS list order: the first source to define a tag is the source,
# later sources (local or repo) with the same tag are shadowed. List local last to make it a fallback.
service_dirs: list[Path] = []
DIRTY_REPOS: list[str] = [] DIRTY_REPOS: list[str] = []
for _entry in raw:
if isinstance(_entry, str) and is_repo_spec(_entry):
try:
_resolved = resolve_service_repo(_entry, force=config.services_repo_force)
except DirtyServiceRepo as e:
# local edits in the clone - record it and let check_load_errors() exit cleanly
DIRTY_REPOS.append(redact_path(str(e.path)))
_resolved = None
if _resolved:
service_dirs.append(_resolved)
else:
service_dirs.append(Path(_entry))
seen: dict[str, Path] = {}
SHADOWED: list[str] = [] SHADOWED: list[str] = []
for _dir in service_dirs:
for _path in _dir.glob("*/__init__.py"):
tag = _path.parent.stem def discover_services() -> tuple[list[Path], list[str]]:
if tag in seen: """Resolve the configured service dirs and glob every ``TAG/__init__.py``.
SHADOWED.append(f"{tag}: using {redact_path(str(seen[tag]))}, ignoring duplicate {redact_path(str(_path))}")
Priority IS list order: the first source to define a tag is the source, and discovery
shadows a later source (local or repo) with the same tag. List local last to make it a
fallback.
Returns the sorted service paths and the human-readable shadow lines.
"""
raw = config.directories.services
if not isinstance(raw, list):
raw = [raw]
service_dirs: list[Path] = []
for entry in raw:
if isinstance(entry, str) and is_repo_spec(entry):
try:
resolved = resolve_service_repo(entry, force=config.services_repo_force)
except DirtyServiceRepo as e:
# local edits in the clone - record it and let check_load_errors() exit cleanly
if (dirty := redact_path(str(e.path))) not in DIRTY_REPOS:
DIRTY_REPOS.append(dirty)
resolved = None
if resolved:
service_dirs.append(resolved)
else: else:
seen[tag] = _path service_dirs.append(Path(entry))
SERVICES = sorted(seen.values(), key=lambda x: x.parent.stem)
seen: dict[str, Path] = {}
shadowed: list[str] = []
for service_dir in service_dirs:
for path in service_dir.glob("*/__init__.py"):
tag = path.parent.stem
if tag in seen:
shadowed.append(
f"{tag}: using {redact_path(str(seen[tag]))}, ignoring duplicate {redact_path(str(path))}"
)
else:
seen[tag] = path
return sorted(seen.values(), key=lambda x: x.parent.stem), shadowed
SERVICES, SHADOWED = discover_services()
def load_service(path: Path) -> object: def load_service(path: Path) -> object:
@@ -88,10 +106,110 @@ MODULES, LOAD_ERRORS = load_services(SERVICES)
ALIASES = {tag: getattr(module, "ALIASES", ()) for tag, module in MODULES.items()} ALIASES = {tag: getattr(module, "ALIASES", ()) for tag, module in MODULES.items()}
PENDING: set[str] = set()
RELOAD_LOCK = threading.Lock()
def reload_services(tags: Iterable[str]) -> list[str]:
"""Re-import the given service tags after a repo refresh; returns load errors.
Mutates every module global in place so importers that bound ``SERVICES``,
``MODULES`` or ``ALIASES`` by name see the swap. Drops a tag that discovery no
longer finds instead of reloading it.
"""
wanted = set(tags)
if not wanted:
return []
with RELOAD_LOCK:
for tag in wanted:
for name in [n for n in sys.modules if n == tag or n.startswith(f"{tag}.")]:
del sys.modules[name]
LOAD_ERRORS[:] = [err for err in LOAD_ERRORS if not err.startswith(f"{tag}: ")]
paths, shadowed = discover_services()
SERVICES[:] = paths
SHADOWED[:] = shadowed
errors: list[str] = []
found = {path.parent.stem: path for path in paths}
for tag in wanted:
if tag not in found:
MODULES.pop(tag, None)
ALIASES.pop(tag, None)
continue
try:
MODULES[tag] = load_service(found[tag])
ALIASES[tag] = getattr(MODULES[tag], "ALIASES", ())
except Exception as e:
errors.append(str(e))
LOAD_ERRORS.extend(errors)
PENDING.difference_update(wanted)
return errors
def repo_specs() -> list[str]:
"""The git repo specs configured under ``directories.services``."""
raw = config.directories.services
entries: list[Any] = raw if isinstance(raw, list) else [raw]
return [e for e in entries if isinstance(e, str) and is_repo_spec(e)]
def refresh_and_reload(busy: set[str] | None = None) -> list[dict[str, Any]]:
"""Force-sync every configured service repo and re-import the services that changed.
Tags in ``busy`` (services with a download job that has not finished) go to ``PENDING``
instead and appear as ``deferred``; ``apply_pending`` swaps them in later.
Blocking (git subprocess): callers wrap it in ``asyncio.to_thread``.
"""
busy = busy or set()
repos: list[dict[str, Any]] = []
for spec in repo_specs():
dest, changes = refresh_repo(spec)
tags = {line[1:] for line in changes if line[:1] in "+~-!"}
if dest and "cloned (new)" in changes:
tags.update(p.parent.stem for p in dest.glob("*/__init__.py"))
deferred = sorted(tags & busy)
PENDING.update(deferred)
errors = reload_services(tags - busy)
repos.append(
{
"spec": spec,
"updated": dest is not None,
"changes": list(changes),
"deferred": deferred,
"load_errors": errors,
}
)
return repos
def apply_pending(busy: set[str] | None = None) -> list[str]:
"""Reload every staged tag whose service is no longer busy; returns the tags applied."""
ready = sorted(PENDING - (busy or set()))
if ready:
for err in reload_services(ready):
log.error("Service reload failed: %s", err)
return ready
SUMMARY_LOGGED = False SUMMARY_LOGGED = False
def log_load_issues() -> None:
"""Log every service load problem without raising - for long-running hosts like serve.
The loader skips broken services (the rest keep working), so the admin needs the log
line the CLI would otherwise render as an error box.
"""
log.info(f"Loaded {len(MODULES)} services" + (f" ({len(SHADOWED)} duplicate(s) ignored)" if SHADOWED else ""))
for shadow in SHADOWED:
log.debug("%s", shadow)
for path in DIRTY_REPOS:
log.warning(f"Service repo has local changes, not refreshed: {path}")
for err in LOAD_ERRORS:
log.error(f"Service skipped: {err}")
def check_load_errors() -> None: def check_load_errors() -> None:
"""Log a one-line load summary (once) and raise a single clean error on failures. """Log a one-line load summary (once) and raise a single clean error on failures.