kenzuya/unshackle
1
0
Fork 2
mirror of https://github.com/unshackle-dl/unshackle.git synced 2026-11-14 21:15:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
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
Download Patch File Download Diff File Expand all files Collapse all files
unshackle/commands/serve.py
+49
View File
@@ -1,6 +1,8 @@
import asyncio
import hmac
import logging
import subprocess
from contextlib import suppress
import click
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
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(
short_help="Serve your Local Widevine/PlayReady Devices and REST API for Remote Access.",
context_settings=context_settings,
@@ -196,6 +243,7 @@ def serve(
app.on_startup.append(start_session_cleanup)
app.on_cleanup.append(stop_session_cleanup)
_install_service_refresh(app)
setup_routes(app, remote_only=remote_only)
if not remote_only:
@@ -289,6 +337,7 @@ def serve(
app.on_startup.append(start_session_cleanup)
app.on_cleanup.append(stop_session_cleanup)
_install_service_refresh(app)
if serve_widevine:
app.on_startup.append(pywidevine_serve._startup)
unshackle/core/api/download_manager.py
+23 -1
View File
@@ -710,10 +710,32 @@ class DownloadQueueManager:
self.put(queue, payload)
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)
for queue in self._subscribers.pop(job.job_id, []):
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
def put(queue: asyncio.Queue, item: Optional[Dict[str, Any]]) -> None:
unshackle/core/api/handlers.py
+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:
"""Answer the request to refresh the service repos configured in directories.services."""
from unshackle.core.service_repo import is_repo_spec, refresh_repo
"""Refresh the service repos configured in directories.services and reload the changed services."""
from unshackle.core.api.download_manager import get_download_manager
from unshackle.core.services import refresh_and_reload
try:
entries = config.directories.services
if not isinstance(entries, list):
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 [])})
busy = get_download_manager().busy_services()
repos = await asyncio.to_thread(refresh_and_reload, busy)
return web.json_response({"refreshed": all(r["updated"] for r in repos), "repos": repos})
except APIError:
unshackle/core/api/routes.py
+28 -4
View File
@@ -9,6 +9,7 @@ from aiohttp import web
from aiohttp_swagger3 import SwaggerDocs, SwaggerInfo, SwaggerUiSettings
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.handlers import (
CORS_HEADERS,
@@ -150,6 +151,11 @@ async def services(request: web.Request) -> web.Response:
schema:
type: object
properties:
load_errors:
type: array
description: Services skipped because they failed to import (startup or repo refresh)
items:
type: string
services:
type: array
items:
@@ -157,6 +163,9 @@ async def services(request: web.Request) -> web.Response:
properties:
tag:
type: string
pending_update:
type: boolean
description: Present (true) while a repo refresh waits for this service's jobs to finish
aliases:
type: array
items:
@@ -221,6 +230,8 @@ async def services(request: web.Request) -> web.Response:
"url": None,
"help": None,
}
if tag in services_module.PENDING:
service_data["pending_update"] = True
try:
service_module = Services.load(tag)
@@ -322,7 +333,7 @@ async def services(request: web.Request) -> web.Response:
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:
log.exception("Error listing services")
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
description: >
Force-sync (git pull) every service repo configured in directories.services.
`refreshed` is true when all repos synced, or when you configured no repos. The
`repos` field holds the per-repo results.
Force-sync (git pull) every service repo configured in directories.services and
re-import the services that changed, without a restart. `refreshed` is true when all
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:
'200':
description: Refresh results
@@ -1423,6 +1437,16 @@ async def maintenance_refresh_services(request: web.Request) -> web.Response:
type: array
items:
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':
description: Server error
"""
unshackle/core/services.py
+146 -28
View File
@@ -3,48 +3,66 @@ from __future__ import annotations
import inspect
import logging
import re
import sys
import threading
from pathlib import Path
from typing import Any, Iterable
import click
from unshackle.core.config import config
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.utils.redact import redact_path
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] = []
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] = []
for _dir in service_dirs:
for _path in _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))}")
def discover_services() -> tuple[list[Path], list[str]]:
"""Resolve the configured service dirs and glob every ``TAG/__init__.py``.
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:
seen[tag] = _path
SERVICES = sorted(seen.values(), key=lambda x: x.parent.stem)
service_dirs.append(Path(entry))
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:
@@ -88,10 +106,110 @@ MODULES, LOAD_ERRORS = load_services(SERVICES)
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
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:
"""Log a one-line load summary (once) and raise a single clean error on failures.