From f6ce600d71e56790bf6f74d005422974ec96e31f Mon Sep 17 00:00:00 2001
From: imSp4rky <george+slight@slight.me>
Date: Mon, 2 Nov 2026 14:52:18 -0700
Subject: [PATCH] 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.
---
 docs/dev/rest-api/dashboard.md                | 11 +++---
 tests/remote/unit/test_dashboard.py           |  4 ++-
 .../unit/test_download_gates_redaction.py     | 23 ++++++++++++
 unshackle/commands/serve.py                   |  2 ++
 unshackle/core/api/download_manager.py        | 24 ++++++++-----
 unshackle/core/api/download_worker.py         |  2 --
 unshackle/core/api/handlers.py                | 36 +++++++++----------
 unshackle/core/api/progress.py                |  2 +-
 unshackle/core/api/session_store.py           | 10 ++++--
 9 files changed, 77 insertions(+), 37 deletions(-)

diff --git a/docs/dev/rest-api/dashboard.md b/docs/dev/rest-api/dashboard.md
index f5ea445b..f82eedfd 100644
--- a/docs/dev/rest-api/dashboard.md
+++ b/docs/dev/rest-api/dashboard.md
@@ -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. |
 
diff --git a/tests/remote/unit/test_dashboard.py b/tests/remote/unit/test_dashboard.py
index 37174a55..ac9def94 100644
--- a/tests/remote/unit/test_dashboard.py
+++ b/tests/remote/unit/test_dashboard.py
@@ -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)
 
diff --git a/tests/remote/unit/test_download_gates_redaction.py b/tests/remote/unit/test_download_gates_redaction.py
index 1b433f20..c581a49c 100644
--- a/tests/remote/unit/test_download_gates_redaction.py
+++ b/tests/remote/unit/test_download_gates_redaction.py
@@ -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."""
 
diff --git a/unshackle/commands/serve.py b/unshackle/commands/serve.py
index 0a2aedbc..b4412621 100644
--- a/unshackle/commands/serve.py
+++ b/unshackle/commands/serve.py
@@ -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())
 
diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py
index 98ecef64..42fbb1eb 100644
--- a/unshackle/core/api/download_manager.py
+++ b/unshackle/core/api/download_manager.py
@@ -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:
diff --git a/unshackle/core/api/download_worker.py b/unshackle/core/api/download_worker.py
index ecfd6288..5ce4500b 100644
--- a/unshackle/core/api/download_worker.py
+++ b/unshackle/core/api/download_worker.py
@@ -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}")
 
diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py
index 71345535..894c6383 100644
--- a/unshackle/core/api/handlers.py
+++ b/unshackle/core/api/handlers.py
@@ -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
diff --git a/unshackle/core/api/progress.py b/unshackle/core/api/progress.py
index 106606f1..350a8216 100644
--- a/unshackle/core/api/progress.py
+++ b/unshackle/core/api/progress.py
@@ -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
diff --git a/unshackle/core/api/session_store.py b/unshackle/core/api/session_store.py
index 7f5f0025..aef68104 100644
--- a/unshackle/core/api/session_store.py
+++ b/unshackle/core/api/session_store.py
@@ -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."""