feat(api): stream download job progress over server-sent events

Adds GET /api/download/jobs/{job_id}/events with snapshot, status, progress and terminal events, query-param auth for EventSource in both serve modes, atomic progress-file writes, and stream-safe CORS and compression middleware.
This commit is contained in:
imSp4rky
2026-08-12 10:35:43 -06:00
parent 581979c5a8
commit 16ae73e3a0
11 changed files with 312 additions and 21 deletions
+14
View File
@@ -108,6 +108,17 @@ X-Secret-Key: <your-secret>
The value must exactly match one of the configured keys (`api_secret`, or one of the keys under `users`). There is no login step, no session cookie, and no token exchange for API access. The same static key is sent on each request.
!!! info "One route also accepts the key as a query parameter"
The job event stream, `GET /api/download/jobs/{job_id}/events`, also accepts the key in a `secret_key` query parameter. This is for browsers: the `EventSource` API cannot set request headers. The exception applies to that one route only, and it works in both server modes.
```javascript
new EventSource(`http://127.0.0.1:8786/api/download/jobs/${jobId}/events?secret_key=${key}`);
```
The header always wins. If the request has an `X-Secret-Key` header, the server checks that header and ignores the query parameter, so a wrong header is rejected even when the query parameter holds a valid key.
A key in a URL is easier to leak than a key in a header, because proxies and browser history can record it. Use the header wherever your client can set one.
=== "curl"
```bash
@@ -269,6 +280,7 @@ With `--caddy`, unshackle also starts `caddy run` using the `Caddyfile` located
| Item | Value |
|---|---|
| Auth header | `X-Secret-Key` |
| Query-parameter alternative | `?secret_key=` on `GET /api/download/jobs/{job_id}/events` only |
| Default bind address | `127.0.0.1:8786` |
| Exempt endpoint | `GET /api/health` |
| Failure status | `401 Unauthorized` |
@@ -293,6 +305,8 @@ The header and config are identical between modes, but the code that enforces th
Because the API-only middleware guards *all* non-health paths, the Swagger UI at `/api/docs/` is also behind `X-Secret-Key` there. In `--remote-only` mode the Swagger UI is not mounted at all.
Both middlewares read the caller's key through the shared `request_secret_key()` helper in `unshackle/core/api/handlers.py`, which returns the `X-Secret-Key` header and falls back to the `secret_key` query parameter only when the matched route is the job event stream. The integrated server needs one extra branch for this: pywidevine's `authentication` is header-only, so `serve_authentication` compares the query-parameter key itself when the header is absent, and delegates to pywidevine in every other case. That is why a present-but-wrong header is rejected in both modes.
### CORS and the `Authorization` header
A `cors_middleware` runs on every response in both modes (and answers `OPTIONS` preflight requests with an empty response). It emits:
+67
View File
@@ -492,6 +492,73 @@ Fetch a single job with full details (equivalent to a `full=true` list entry).
| `200` | - | Job returned. |
| `404` | `JOB_NOT_FOUND` | No such job. |
### `GET /api/download/jobs/{job_id}/events`
Stream the job's progress as [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events). Use this instead of polling `GET /api/download/jobs/{job_id}`.
The response has the content type `text/event-stream`. Each event has this format:
```
event: <name>
data: <job JSON>
```
The `data` field of every event is the same job object that `GET /api/download/jobs/{job_id}` returns.
| Event | When it is sent |
| --- | --- |
| `snapshot` | Immediately, as the first event. Gives the job's current state. |
| `status` | The job leaves the queue and starts to download. |
| `progress` | The worker's progress record changed. The server reads that record every 0.5 seconds, so you get at most two of these events each second, and none while the record stays the same. |
| `completed` | The job finished. The server then closes the stream. |
| `failed` | The job failed. The server then closes the stream. |
| `cancelled` | The job was cancelled. The server then closes the stream. |
The server sends a `: keep-alive` comment each 15 seconds while the job is quiet, to keep proxies from closing an idle connection. Ignore these lines.
If the job is already in a terminal state, the server sends the `snapshot` event, then the terminal event, and closes the stream immediately.
!!! tip "Authentication from a browser"
`EventSource` cannot set headers. For this endpoint only, the server also accepts the key in the `secret_key` query parameter. This works in both server modes, the default integrated server and `--api-only`:
```javascript
const events = new EventSource(
`http://127.0.0.1:8786/api/download/jobs/${jobId}/events?secret_key=${key}`
);
events.addEventListener("progress", (e) => console.log(JSON.parse(e.data).progress));
events.addEventListener("completed", () => events.close());
```
The header always wins. Send the key in the query parameter *or* in the header, not both: if an `X-Secret-Key` header is present, the server checks that header and ignores the query parameter. See [Authentication](authentication.md#how-clients-present-credentials).
=== "Request"
```bash
curl -N -H "X-Secret-Key: $KEY" \
"http://127.0.0.1:8786/api/download/jobs/b0f7c8e2-.../events"
```
=== "Response `200`"
```
event: snapshot
data: {"job_id":"b0f7c8e2-...","status":"downloading","progress":42.5,...}
event: progress
data: {"job_id":"b0f7c8e2-...","status":"downloading","progress":48.0,...}
: keep-alive
event: completed
data: {"job_id":"b0f7c8e2-...","status":"completed","progress":100.0,...}
```
| Status | Error code | Meaning |
| --- | --- | --- |
| `200` | - | Event stream started. |
| `404` | `JOB_NOT_FOUND` | No such job. |
### `DELETE /api/download/jobs/{job_id}`
Cancel or remove a job. The behaviour depends on the job's current state:
+4 -2
View File
@@ -22,8 +22,9 @@ calls you can script against:
- **Discovery**: list the services available on the server, run a service's search,
and enumerate the titles and tracks behind a title ID.
- **Downloading**: submit a download as a background *job*, then poll its status and
live progress, cancel it, retry it, or bump it to the front of the queue.
- **Downloading**: submit a download as a background *job*, then follow its status and
live progress (by polling, or by subscribing to its Server-Sent Events stream),
cancel it, retry it, or bump it to the front of the queue.
- **History and housekeeping**: read the log of finished jobs, inspect the server's
effective (redacted) configuration, check external tool versions, and clear caches
or temp files.
@@ -161,6 +162,7 @@ request and response in full.
| `GET` | `/api/download/jobs` | List jobs, with filtering and sorting. |
| `POST` | `/api/download/jobs/clear-finished` | Remove all finished jobs. |
| `GET` | `/api/download/jobs/{job_id}` | Full detail for one job. |
| `GET` | `/api/download/jobs/{job_id}/events` | Stream one job's progress as Server-Sent Events. |
| `DELETE` | `/api/download/jobs/{job_id}` | Cancel a running job, or remove a finished one. |
| `POST` | `/api/download/jobs/{job_id}/retry` | Re-queue a finished job's parameters as a new job. |
| `POST` | `/api/download/jobs/{job_id}/priority` | Move a queued job to the front. |
+5
View File
@@ -177,6 +177,11 @@ The full detail adds `started_time`, `completed_time`, `output_files`, the redac
done
```
!!! tip "Or subscribe instead of polling"
`GET /api/download/jobs/{job_id}/events` pushes the same job object as Server-Sent
Events, and closes the stream when the job ends. See
[Endpoints](endpoints.md#get-apidownloadjobsjob_idevents).
!!! note "What the percentage means"
`progress` climbs from `0` toward `90` while tracks download, then the final stretch
to `100` covers post-download work such as decryption cleanup, repackaging, and
+9 -1
View File
@@ -310,7 +310,15 @@ done
```
!!! tip "Polling cadence"
A few seconds between polls is plenty. The server writes progress updates roughly twice a second internally, but there is no push/streaming channel; you read the current snapshot each time you ask.
A few seconds between polls is plenty. The server updates a job's progress roughly twice a second internally, and each poll reads the current snapshot.
!!! tip "Or let the server push the updates"
`GET /api/download/jobs/{job_id}/events` streams the same job record as Server-Sent Events, and closes the stream when the job ends. It removes the polling loop above. See [Endpoints](endpoints.md#get-apidownloadjobsjob_idevents).
```bash
curl -N "$BASE/api/download/jobs/$JOB/events" \
-H "X-Secret-Key: $SECRET"
```
If a job ends in `failed`, the full record includes `error_message`, `error_code`, and (when the server was started with `--debug-api`) `error_traceback` and `worker_stderr` to help you diagnose it. Sensitive values in these fields are scrubbed before they are returned.
+9 -3
View File
@@ -8,6 +8,7 @@ from aiohttp import web
from unshackle.core import binaries
from unshackle.core.api import cors_middleware, setup_routes, setup_swagger
from unshackle.core.api.compression import compression_middleware
from unshackle.core.api.handlers import request_secret_key
from unshackle.core.config import config
from unshackle.core.constants import context_settings
from unshackle.core.downloaders import format_speed, parse_speed_limit, set_speed_limit
@@ -130,11 +131,11 @@ def serve(
config.serve["playready_devices"].extend(list(config.directories.prds.glob("*.prd")))
@web.middleware
async def api_key_authentication(request: web.Request, handler) -> web.Response:
async def api_key_authentication(request: web.Request, handler) -> web.StreamResponse:
"""Authenticate API requests using X-Secret-Key header."""
if request.path == "/api/health":
return await handler(request)
secret_key = request.headers.get("X-Secret-Key")
secret_key = request_secret_key(request)
if not secret_key:
return web.json_response({"status": 401, "message": "Secret Key is Empty."}, status=401)
# Constant-time compare against every configured key so a valid key can't be
@@ -238,9 +239,14 @@ def serve(
def create_serve_authentication(serve_playready_flag: bool):
@web.middleware
async def serve_authentication(request: web.Request, handler) -> web.Response:
async def serve_authentication(request: web.Request, handler) -> web.StreamResponse:
secret_key = request_secret_key(request)
if serve_playready_flag and request.path in ("/playready", "/playready/"):
response = await handler(request)
elif secret_key and not request.headers.get("X-Secret-Key"):
if not any(hmac.compare_digest(secret_key, k) for k in request.app["config"]["users"]):
return web.json_response({"status": 401, "message": "Secret Key is Invalid."}, status=401)
response = await handler(request)
else:
response = await pywidevine_serve.authentication(request, handler)
+5 -2
View File
@@ -25,10 +25,13 @@ def safe_inflate(data: bytes, max_size: int = MAX_INFLATE_BYTES) -> bytes:
@web.middleware
async def compression_middleware(request: web.Request, handler) -> web.Response:
async def compression_middleware(request: web.Request, handler) -> web.StreamResponse:
"""Compress JSON responses with gzip when the client supports it."""
response = await handler(request)
if not isinstance(response, web.Response):
return response
accept_encoding = request.headers.get("Accept-Encoding", "")
if "gzip" not in accept_encoding:
return response
@@ -37,7 +40,7 @@ async def compression_middleware(request: web.Request, handler) -> web.Response:
return response
body = response.body
if body is None or len(body) < 256:
if not isinstance(body, (bytes, bytearray)) or len(body) < 256:
return response
from unshackle.core.config import config
+55
View File
@@ -623,6 +623,7 @@ class DownloadQueueManager:
self._job_temp_files: Dict[str, Dict[str, str]] = {}
self._workers_started = False
self._shutdown_event = asyncio.Event()
self._subscribers: Dict[str, List[asyncio.Queue]] = {}
log.info(
f"Initialized download queue manager: max_concurrent={max_concurrent_downloads}, retention_hours={job_retention_hours}"
@@ -655,6 +656,48 @@ class DownloadQueueManager:
"""List all jobs."""
return list(self._jobs.values())
def subscribe(self, job_id: str) -> asyncio.Queue:
"""Register a listener for a job's events, returning the queue it receives them on."""
queue: asyncio.Queue = asyncio.Queue(maxsize=100)
self._subscribers.setdefault(job_id, []).append(queue)
return queue
def unsubscribe(self, job_id: str, queue: asyncio.Queue) -> None:
"""Drop a listener registered by subscribe()."""
queues = self._subscribers.get(job_id)
if not queues:
return
with suppress(ValueError):
queues.remove(queue)
if not queues:
del self._subscribers[job_id]
def _publish(self, job: DownloadJob, event: str) -> None:
"""Fan an event carrying the job's full detail dict out to every listener."""
queues = self._subscribers.get(job.job_id)
if not queues:
return
payload = {"event": event, "data": job.to_dict(include_full_details=True)}
for queue in queues:
self._put(queue, payload)
def _publish_terminal(self, job: DownloadJob) -> None:
"""Publish a job's final state, then close every listener's stream."""
self._publish(job, job.status.value)
for queue in self._subscribers.pop(job.job_id, []):
self._put(queue, None)
@staticmethod
def _put(queue: asyncio.Queue, item: Optional[Dict[str, Any]]) -> None:
"""Enqueue an item, discarding the oldest when a slow listener has filled the queue."""
try:
queue.put_nowait(item)
except asyncio.QueueFull:
with suppress(asyncio.QueueEmpty):
queue.get_nowait()
with suppress(asyncio.QueueFull):
queue.put_nowait(item)
def cancel_job(self, job_id: str) -> bool:
"""Cancel a job if it's queued or downloading."""
job = self._jobs.get(job_id)
@@ -666,6 +709,7 @@ class DownloadQueueManager:
job.cancel_event.set() # Signal cancellation
job.completed_time = datetime.now()
record_job_history(job) # queued jobs never reach a worker, so persist here
self._publish_terminal(job)
log.info(f"Cancelled queued job {sanitize_log(job_id)}")
return True
elif job.status == JobStatus.DOWNLOADING:
@@ -763,6 +807,11 @@ class DownloadQueueManager:
log.info("Shutting down download queue manager")
self._shutdown_event.set()
for queues in list(self._subscribers.values()):
for queue in queues:
self._put(queue, None)
self._subscribers.clear()
# Cancel all active downloads
for task in self._active_downloads.values():
task.cancel()
@@ -819,6 +868,7 @@ class DownloadQueueManager:
# Start processing the job
job.status = JobStatus.DOWNLOADING
job.started_time = datetime.now()
self._publish(job, "status")
log.info(f"Worker {worker_name} starting job {job.job_id}")
@@ -838,6 +888,7 @@ class DownloadQueueManager:
finally:
job.completed_time = datetime.now()
record_job_history(job)
self._publish_terminal(job)
if job.job_id in self._active_downloads:
del self._active_downloads[job.job_id]
@@ -913,6 +964,7 @@ class DownloadQueueManager:
stdout_bytes = b""
stderr_bytes = b""
last_published: Optional[Dict[str, Any]] = None
try:
while True:
@@ -957,6 +1009,9 @@ class DownloadQueueManager:
if new_progress != job.progress:
job.progress = new_progress
log.info(f"Job {job.job_id} progress updated: {job.progress}%")
if progress_data != last_published:
last_published = progress_data
self._publish(job, "progress")
except (FileNotFoundError, json.JSONDecodeError, ValueError) as e:
log.debug(f"Could not read progress for job {job.job_id}: {e}")
+5 -2
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import json
import logging
import os
import sys
import traceback
from pathlib import Path
@@ -20,9 +21,11 @@ def _read_payload(path: Path) -> Dict[str, Any]:
def _write_result(path: Path, payload: Dict[str, Any]) -> None:
# Atomic replace: the parent polls this file while it is being rewritten.
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as handle:
json.dump(payload, handle)
tmp = path.with_name(path.name + ".tmp")
tmp.write_text(json.dumps(payload), encoding="utf-8")
os.replace(tmp, path)
def main(argv: list[str]) -> int:
+91 -3
View File
@@ -263,7 +263,7 @@ def get_allowed_services(request: Optional[web.Request] = None) -> Optional[List
key_set: Optional[set[str]] = None
if request:
secret_key = request.headers.get("X-Secret-Key")
secret_key = request_secret_key(request)
if secret_key:
users = config.serve.get("users", {})
user_config = users.get(secret_key, {})
@@ -283,9 +283,32 @@ def get_allowed_services(request: Optional[web.Request] = None) -> Optional[List
return list(result)
JOB_EVENTS_ROUTE = "/api/download/jobs/{job_id}/events"
CORS_HEADERS = {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, OPTIONS",
"Access-Control-Allow-Headers": "Content-Type, X-Secret-Key, Authorization",
"Access-Control-Max-Age": "3600",
}
def request_secret_key(request: web.Request) -> Optional[str]:
"""The caller's key: the X-Secret-Key header, or on the events route the secret_key
query param (EventSource cannot send headers)."""
key = request.headers.get("X-Secret-Key")
if not key:
resource = request.match_info.route.resource
if resource is not None and resource.canonical == JOB_EVENTS_ROUTE:
key = request.query.get("secret_key")
return key
def caller_key(request: Optional[web.Request] = None) -> str:
"""The authenticating X-Secret-Key for a request, or 'anonymous' when unauthenticated."""
return request.headers.get("X-Secret-Key", "anonymous") if request else "anonymous"
"""The authenticating key for a request, or 'anonymous' when unauthenticated."""
if not request:
return "anonymous"
return request_secret_key(request) or "anonymous"
def owns_job(job: Any, request: Optional[web.Request] = None) -> bool:
@@ -1214,6 +1237,71 @@ async def get_download_job_handler(job_id: str, request: Optional[web.Request] =
)
async def download_job_events_handler(job_id: str, request: web.Request) -> web.StreamResponse:
"""Stream a download job's progress to the caller as Server-Sent Events."""
import json
from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager
manager = get_download_manager()
job = manager.get_job(job_id)
if not job or not owns_job(job, request):
raise APIError(
APIErrorCode.JOB_NOT_FOUND,
"Job not found",
details={"job_id": job_id},
)
response = web.StreamResponse(
headers={
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
# Stops nginx from buffering the stream.
"X-Accel-Buffering": "no",
**CORS_HEADERS,
}
)
await response.prepare(request)
async def send(event: str, data: Dict[str, Any]) -> None:
payload = json.dumps(data, separators=(",", ":"), default=str)
await response.write(f"event: {event}\ndata: {payload}\n\n".encode())
queue: Optional[asyncio.Queue] = None
get_task: Optional[asyncio.Task] = None
try:
await send("snapshot", job.to_dict(include_full_details=True))
if job.status not in TERMINAL_STATUSES:
queue = manager.subscribe(job_id)
if queue is None or job.status in TERMINAL_STATUSES:
await send(job.status.value, job.to_dict(include_full_details=True))
return response
while True:
if get_task is None:
get_task = asyncio.create_task(queue.get())
done, _ = await asyncio.wait({get_task}, timeout=15)
if not done:
await response.write(b": keep-alive\n\n")
continue
item = get_task.result()
get_task = None
if item is None:
break
await send(item["event"], item["data"])
except (ConnectionResetError, asyncio.CancelledError):
log.debug(f"SSE client disconnected from job {sanitize_log(job_id)}")
finally:
if get_task is not None:
get_task.cancel()
if queue is not None:
manager.unsubscribe(job_id, queue)
return response
async def cancel_download_job_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response:
"""Handle cancel/remove download job request."""
from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager
+48 -8
View File
@@ -11,6 +11,8 @@ from aiohttp_swagger3 import SwaggerDocs, SwaggerInfo, SwaggerUiSettings
from unshackle.core import __code_hash__, __version__
from unshackle.core.api.errors import APIError, APIErrorCode, build_error_response, handle_api_exception
from unshackle.core.api.handlers import (
CORS_HEADERS,
JOB_EVENTS_ROUTE,
cancel_download_job_handler,
clear_cache_handler,
clear_finished_download_jobs_handler,
@@ -18,6 +20,7 @@ from unshackle.core.api.handlers import (
delete_history_handler,
download_handler,
download_history_handler,
download_job_events_handler,
env_check_handler,
get_allowed_services,
get_download_job_handler,
@@ -56,26 +59,22 @@ async def cors_middleware(
else:
response = await handler(request)
# Add CORS headers
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type, X-Secret-Key, Authorization"
response.headers["Access-Control-Max-Age"] = "3600"
response.headers.update(CORS_HEADERS)
return response
log = logging.getLogger("api")
# Route handler signature: takes the request, returns a response.
Handler = Callable[[web.Request], Awaitable[web.Response]]
# Route handler signature: takes the request, returns a response (streamed or complete).
Handler = Callable[[web.Request], Awaitable[web.StreamResponse]]
def api_handler(handler: Handler) -> Handler:
"""Wrap a route handler so any raised APIError becomes a structured error response."""
@functools.wraps(handler)
async def wrapper(request: web.Request) -> web.Response:
async def wrapper(request: web.Request) -> web.StreamResponse:
try:
return await handler(request)
except APIError as e:
@@ -956,6 +955,46 @@ async def download_job_detail(request: web.Request) -> web.Response:
return await get_download_job_handler(job_id, request)
@api_handler
async def download_job_events(request: web.Request) -> web.StreamResponse:
"""
Stream download job events.
---
summary: Stream download job events
description: >
Server-Sent Events stream of a job's progress. Emits a `snapshot` event, then
`progress` and `status` events, and closes after the terminal `completed`, `failed`
or `cancelled` event. Every event carries the same full job object that
GET /api/download/jobs/{job_id} returns. A browser EventSource can authenticate with
the `secret_key` query parameter instead of the X-Secret-Key header; when both are
sent the header is used.
parameters:
- name: job_id
in: path
required: true
schema:
type: string
- name: secret_key
in: query
required: false
schema:
type: string
responses:
'200':
description: Event stream
content:
text/event-stream:
schema:
type: string
'404':
description: Job not found
'500':
description: Server error
"""
job_id = request.match_info["job_id"]
return await download_job_events_handler(job_id, request)
@api_handler
async def cancel_download_job(request: web.Request) -> web.Response:
"""
@@ -1801,6 +1840,7 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
("GET", "/api/download/jobs", download_jobs, False),
("POST", "/api/download/jobs/clear-finished", clear_finished_download_jobs, False),
("GET", "/api/download/jobs/{job_id}", download_job_detail, False),
("GET", JOB_EVENTS_ROUTE, download_job_events, False),
("DELETE", "/api/download/jobs/{job_id}", cancel_download_job, False),
("POST", "/api/download/jobs/{job_id}/retry", retry_download_job, False),
("POST", "/api/download/jobs/{job_id}/priority", prioritize_download_job, False),