perf(api)!: cut per-request and per-tick event bus traffic

Session update events carry the new action as action_entry instead of the full log; access lines stay out of the log ring unless --debug. Job events build and redact their dict once, the progress interval matches the 0.5 s poll, and dashboard stats are serialized once.
This commit is contained in:
imSp4rky
2026-11-02 14:52:18 -07:00
parent bb081046e2
commit f6ce600d71
9 changed files with 77 additions and 37 deletions
+6 -5
View File
@@ -28,8 +28,9 @@ routes are available in `--remote-only` mode too.
Run the server headless with `--quiet`: no banner, no Rich output, one plain log line per
event on stderr, and `api.*` loggers stay at `INFO` so remote session activity reaches the
log ring. Add `--debug` to raise the level to `DEBUG` and also write the full log to
`directories.logs` as `unshackle_serve_<time>.log`. Nothing else the dashboard shows reaches
log ring. The log ring and the event stream do not get `aiohttp.access` request lines.
Add `--debug` to raise the level to `DEBUG`, send the request lines to the log ring too,
and write the full log to `directories.logs` as `unshackle_serve_<time>.log`. Nothing else the dashboard shows reaches
the disk: stats, the log ring and the event stream live in memory and reset on restart, so a
dashboard UI that wants history must keep its own copy.
@@ -48,13 +49,13 @@ All are `GET`, except the two `drain` routes.
| `DELETE /api/dashboard/drain` | Cancels the drain and returns the `/status` body. The server accepts new work again. |
| `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the display name of the title the client is on: the last one it asked tracks for, else the first resolved title), `auth_status`, `auth_error`, `server_account` (the server profile lent to the remote session, else `null`), `server_cdm` (`false` when the client's own device licenses the remote session), `server_cdm_max_height` (the tallest video the server CDM licenses live, else `null`), `client`, `actions`, `created_at` and `last_accessed` (ISO 8601) with `created_ts` and `last_accessed_ts` (Unix epoch, the same clock as log `ts`), `age_seconds`, `idle_seconds`, `log_seq`. `client` is whatever the remote client sent as `client` in its session create request. The CLI sends `version`, `code_hash` (the commit the client runs, `null` when its source cannot be read), `platform` and `argv`. `argv` is the command line the user ran: proxy and URL userinfo, secret query parameters and credential values become `***`, home and install paths shorten as in the logs, and the line is cut at 3000 characters. An older client sends less or nothing and the field is `{}`. `log_seq` is the last sequence number in the remote session's service log - poll `/api/dashboard/sessions/{id}/logs` when it changes. `actions` is the session's request log, newest last, capped at 500: `ts`, `method`, `action` (`titles`, `tracks`, `segments`, `license`, `prompt`, …), `query`, `status`, `ms`, `bytes_in`, `bytes_out`. |
| `/api/dashboard/jobs` | Every download job with full detail, regardless of owner. Empty in `--remote-only` mode. |
| `/api/dashboard/logs` | `{"seq": N, "records": [...]}`: the last 1000 log records. `?since=<seq>` returns only newer records; `?level=WARNING` sets the minimum level; `?logger=serve` keeps one logger and its children (`aiohttp.access` is the noisiest). `ts` is a Unix epoch in seconds. |
| `/api/dashboard/logs` | `{"seq": N, "records": [...]}`: the last 1000 log records. `?since=<seq>` returns only newer records; `?level=WARNING` sets the minimum level; `?logger=serve` keeps one logger and its children. `ts` is a Unix epoch in seconds. |
| `/api/dashboard/cdm-logs` | `{"seq": N, "records": [...], "sessions": [...]}`: the last 5000 calls to the Widevine and PlayReady CDM device routes, and the open sessions of each device. `?since=<seq>` returns only newer records. See [CDM calls](#cdm-calls). |
| `/api/dashboard/sessions/{id}/logs` | One remote session's service log: `{"session_id", "last_seq", "records": [...]}`. Each record has `seq`, `ts`, `level`, `message`. `?since=<seq>` returns only newer records. 404 when the remote session is unknown. |
| `/api/dashboard/keys` | Every configured API key: what it may do and what it has done. One row per key in `serve.users`, plus `serve.api_secret` when it is set. |
| `/api/dashboard/services` | Every service configured on this server and its load state, including the ones that failed to import. See [Services](#services) for what configured means. |
| `/api/dashboard/health` | Preflight: whether this instance could finish a download. Cached 30 seconds. |
| `/api/dashboard/events` | Server-Sent Events stream. With `?since=<seq>` it returns `{"seq", "stats", "events": [...]}` instead: every event after `seq` (the server keeps the last 20000, log records included, so a poll every few seconds is safe even under a busy access log) as one JSON burst, so a UI that polls every few seconds still sees a remote session that opened and closed in between. |
| `/api/dashboard/events` | Server-Sent Events stream. With `?since=<seq>` it returns `{"seq", "stats", "events": [...]}` instead: every event after `seq` (the server keeps the last 20000, log records included, so a poll every few seconds is safe) as one JSON burst, so a UI that polls every few seconds still sees a remote session that opened and closed in between. |
`errors` counts every structured error response since the server started, for example
`{"EXAMPLE": {"SERVICE_ERROR": 4}, "-": {"SESSION_NOT_FOUND": 1}}`. The service is the one of
@@ -74,7 +75,7 @@ those it sends:
| Event | Data |
|-------|------|
| `log` | One log record: `seq`, `ts`, `level`, `logger`, `msg`. |
| `session` | `action` (`create`, `update` on every request the session makes and when titles or tracks load, or `delete`) plus the full remote session summary, so each event is a complete snapshot. A `delete` also carries `reason`: `closed` (the client ended it), `expired` (TTL sweep) or `evicted` (`max_sessions` reached). A session opened by `--skip-dl` can live only a few seconds, so a UI that wants history must keep its own record of these events. |
| `session` | `action` (`create`, `update` on every request the session makes and when titles or tracks load, or `delete`) plus the remote session summary. The `update` for a request carries that request as `action_entry` and has no `actions`: append it to the log you hold, capped at 500. Every other event is a complete snapshot with `actions`. A `delete` also carries `reason`: `closed` (the client ended it), `expired` (TTL sweep) or `evicted` (`max_sessions` reached). A session opened by `--skip-dl` can live only a few seconds, so a UI that wants history must keep its own record of these events. |
| `job` | `event` (`queued`, `progress`, `status`, or the terminal status) plus the full job object. |
| `service` | `action` (`staged`, `applied` or `failed`) and `tags`, the services it applies to. A `failed` action carries `errors` and an empty `tags`. Refetch `/api/dashboard/services` when one arrives, instead of polling for a reload. |
+3 -1
View File
@@ -183,7 +183,9 @@ async def test_session_action_log() -> None:
store.record_action(entry.session_id, {"action": "titles", "status": 200})
store.record_action("missing", {"action": "titles", "status": 200})
assert entry.summary()["actions"] == [{"action": "titles", "status": 200}]
assert queue.get_nowait()["data"]["action"] == "update" and queue.empty()
event = queue.get_nowait()["data"]
assert event["action"] == "update" and queue.empty()
assert event["action_entry"] == {"action": "titles", "status": 200} and "actions" not in event
finally:
bus.unsubscribe(queue)
@@ -82,6 +82,29 @@ def test_to_dict_full_details_redacts_error_fields_and_parameters():
assert d["parameters"]["credential"] == "***"
def test_to_dict_redacts_again_only_when_an_error_field_changes(monkeypatch):
from unshackle.core.api import download_manager
calls = []
real = download_manager._redact_text
monkeypatch.setattr(download_manager, "_redact_text", lambda text, params: calls.append(text) or real(text, params))
job = DownloadJob(
job_id="j1",
status=JobStatus.FAILED,
created_time=datetime(2026, 1, 1),
service="EXAMPLE",
title_id="t",
parameters={"credential": "user:hunter2"},
)
job.error_message = "login failed for user:hunter2"
job.to_dict(include_full_details=True)
job.to_dict(include_full_details=True)
assert len(calls) == 4
job.error_message = "still failing for user:hunter2"
assert "hunter2" not in job.to_dict(include_full_details=True)["error_message"]
assert len(calls) == 8
class _PastGate(Exception):
"""Raised by the stubbed download manager to prove a request got past the gate."""
+2
View File
@@ -424,6 +424,8 @@ def serve(
logging.getLogger("api").setLevel(logging.WARNING)
logging.getLogger("api.remote").setLevel(logging.WARNING)
ring.setFormatter(logging.Formatter("%(message)s"))
if not debug:
ring.addFilter(lambda record: record.name != "aiohttp.access")
logging.getLogger().addHandler(ring)
logging.getLogger("aiohttp.access").addFilter(_MaskAccessKey())
+16 -8
View File
@@ -93,6 +93,7 @@ class JobStatus(Enum):
# Statuses a job can never leave; such jobs are safe to remove or retry.
WORKER_STDERR_LIMIT = 64 * 1024
TERMINAL_STATUSES = (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED)
@@ -143,6 +144,8 @@ class DownloadJob:
# Guards against writing the same job to the persistent history file twice.
history_recorded: bool = False
redacted: Optional[tuple[tuple, Dict[str, Optional[str]]]] = field(default=None, repr=False, compare=False)
# X-Secret-Key that created the job; scopes per-caller access, never serialized into to_dict().
owner_key: Optional[str] = None
@@ -173,17 +176,21 @@ class DownloadJob:
if include_full_details:
# Error/stderr/traceback are free text a service may have echoed a credential or proxy
# URL into, so scrub them with the same secrets that redact_parameters masks.
raw = (self.error_message, self.error_details, self.error_traceback, self.worker_stderr)
if not self.redacted or self.redacted[0] != raw:
names = ("error_message", "error_details", "error_traceback", "worker_stderr")
self.redacted = (
raw,
{name: redact_path(_redact_text(text, self.parameters)) for name, text in zip(names, raw)},
)
result.update(
{
"parameters": redact_parameters(self.parameters),
"started_time": self.started_time.isoformat() if self.started_time else None,
"completed_time": self.completed_time.isoformat() if self.completed_time else None,
"output_files": self.output_files,
"error_message": redact_path(_redact_text(self.error_message, self.parameters)),
"error_details": redact_path(_redact_text(self.error_details, self.parameters)),
"error_code": self.error_code,
"error_traceback": redact_path(_redact_text(self.error_traceback, self.parameters)),
"worker_stderr": redact_path(_redact_text(self.worker_stderr, self.parameters)),
**self.redacted[1],
}
)
@@ -781,11 +788,12 @@ class DownloadQueueManager:
def publish(self, job: DownloadJob, event: str) -> None:
"""Fan an event carrying the job's full detail dict out to every listener."""
bus.publish("job", {"event": event, **job.to_dict(include_full_details=True)})
data = job.to_dict(include_full_details=True)
bus.publish("job", {"event": event, **data})
queues = self._subscribers.get(job.job_id)
if not queues:
return
payload = {"event": event, "data": job.to_dict(include_full_details=True)}
payload = {"event": event, "data": data}
for queue in queues:
self.put(queue, payload)
@@ -1164,7 +1172,7 @@ class DownloadQueueManager:
new_progress = float(progress_data["progress"])
if new_progress != job.progress:
job.progress = new_progress
log.info(f"Job {job.job_id} progress updated: {job.progress}%")
log.debug(f"Job {job.job_id} progress updated: {job.progress}%")
# submit_input() clears the prompt out of band, so a re-prompt with the same text
# leaves the file equal to last_published; publish the prompt change anyway.
if progress_data != last_published or job.input_prompt != prompt_before:
@@ -1196,7 +1204,7 @@ class DownloadQueueManager:
if stdout.strip():
log.debug(f"Worker stdout for job {job.job_id}: {safe_stdout}")
if stderr.strip():
job.worker_stderr = stderr.strip()
job.worker_stderr = stderr.strip()[-WORKER_STDERR_LIMIT:]
if returncode != 0:
log.warning(f"Worker stderr for job {job.job_id}: {safe_stderr}")
else:
-2
View File
@@ -140,9 +140,7 @@ def main(argv: list[str]) -> int:
try:
with progress_lock:
progress_state.update(progress_data)
log.info(f"Writing progress update: {progress_data}")
write_result(progress_path, progress_state)
log.info(f"Progress update written to {progress_path}")
except Exception as e:
log.error(f"Failed to write progress update: {e}")
+18 -18
View File
@@ -2746,7 +2746,7 @@ def configured_service_tags() -> List[str]:
return [tag for tag in tags if tag in named]
async def dashboard_status_handler(request: web.Request) -> web.Response:
def dashboard_status() -> Dict[str, Any]:
"""Server-wide stats for the developer dashboard."""
from unshackle.core.api.download_manager import get_download_manager
from unshackle.core.api.session_store import get_session_store
@@ -2754,17 +2754,20 @@ async def dashboard_status_handler(request: web.Request) -> web.Response:
store = get_session_store()
jobs = Counter(job.status.value for job in get_download_manager().list_jobs())
return web.json_response(
{
**stats.to_dict(),
"services": len(configured_service_tags()),
"sessions": store.session_count,
"max_sessions": store.max_sessions,
"session_ttl": store.ttl,
"jobs": dict(jobs),
"draining": store.draining,
}
)
return {
**stats.to_dict(),
"services": len(configured_service_tags()),
"sessions": store.session_count,
"max_sessions": store.max_sessions,
"session_ttl": store.ttl,
"jobs": dict(jobs),
"draining": store.draining,
}
async def dashboard_status_handler(request: web.Request) -> web.Response:
"""Answer the request for the dashboard's server-wide stats."""
return web.json_response(dashboard_status())
async def dashboard_drain_start_handler(request: web.Request) -> web.Response:
@@ -3143,12 +3146,10 @@ async def dashboard_events_handler(request: web.Request) -> web.StreamResponse:
since = int(request.query["since"])
except ValueError:
since = 0
status = await dashboard_status_handler(request)
status = dashboard_status()
before = bus.seq
events = bus.since(since)
return web.json_response(
{"seq": _poll_cursor(before, events), "stats": json.loads(status.text or "{}"), "events": events}
)
return web.json_response({"seq": _poll_cursor(before, events), "stats": status, "events": events})
response = web.StreamResponse(
headers={
@@ -3167,8 +3168,7 @@ async def dashboard_events_handler(request: web.Request) -> web.StreamResponse:
await response.write(f"{head}event: {event}\ndata: {payload}\n\n".encode())
async def send_stats() -> None:
status = await dashboard_status_handler(request)
await send("stats", json.loads(status.text or "{}"))
await send("stats", dashboard_status())
queue = bus.subscribe()
get_task: Optional[asyncio.Task] = None
+1 -1
View File
@@ -13,7 +13,7 @@ from unshackle.core.constants import AnyTrack
from unshackle.core.utilities import standard_resolution
JOB_PROGRESS_TERMINAL_STATES = {"Downloaded", "Decrypted", "[yellow]SKIPPED"}
PROGRESS_EMIT_INTERVAL = 0.25
PROGRESS_EMIT_INTERVAL = 0.5
# Weight for a track with no bitrate (subtitles); small vs media bitrates so subs barely move the bar.
SUBTITLE_PROGRESS_WEIGHT = 50_000.0
+8 -2
View File
@@ -117,10 +117,16 @@ def _schedule_pending_reload(service_tag: str) -> None:
schedule_pending_reload(service_tag)
def _publish(action: str, entry: SessionEntry, reason: Optional[str] = None) -> None:
def _publish(
action: str, entry: SessionEntry, reason: Optional[str] = None, action_entry: Optional[Dict[str, Any]] = None
) -> None:
"""Tell dashboard listeners about a remote session; ``action_entry`` replaces the action log with that one action."""
data = {"action": action, **entry.summary()}
if reason:
data["reason"] = reason
if action_entry is not None:
del data["actions"]
data["action_entry"] = action_entry
bus.publish("session", data)
@@ -319,7 +325,7 @@ class SessionStore:
if entry is None:
return
entry.actions.append(action)
_publish("update", entry)
_publish("update", entry, action_entry=action)
def list(self) -> List[SessionEntry]:
"""Snapshot of every live session."""