refactor(api): remove dead code paths in serve/remote layer

Delete code paths that never executed in the serve/API/remote stack:

- BridgeCancelledError: raised but never caught; replace raises with RuntimeError and drop the class.
- InputBridge.error setter and _error field: written at the handler layer, never read (session.auth_error is the surfaced channel).
- cancel_event param and _check_cancel inner fn on _perform_download: the only caller always passed None (the live job.cancel_event is untouched).
- result_with_progress monkeypatch: its status pings were redundant with the file-polled progress_sink, which is now the sole path.
- session param on _resolve_manifest_data: unused in the body.

Reconcile SessionStore's hardcoded 600s auth-state TTL into a shared AUTH_INPUT_TIMEOUT constant referenced by both the InputBridge input timeout and the store, removing the magic number without changing the longer grace window.

Update input_bridge/session_store tests for the new channels and add an assertion that auth-state entries expire at the single rule.
This commit is contained in:
imSp4rky
2026-06-21 13:07:45 -06:00
parent 40eafce17b
commit 8a9a1e1bdc
8 changed files with 36 additions and 65 deletions
+2 -29
View File
@@ -160,20 +160,13 @@ def _perform_download(
service: str,
title_id: str,
params: Dict[str, Any],
cancel_event: Optional[threading.Event] = None,
progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
) -> List[str]:
"""Execute the synchronous download logic for a job."""
def _check_cancel(stage: str):
if cancel_event and cancel_event.is_set():
raise Exception(f"Job was cancelled {stage}")
from contextlib import redirect_stderr, redirect_stdout
from io import StringIO
_check_cancel("before execution started")
# Import dl.py components lazily to avoid circular deps during module import
import click
import yaml
@@ -328,8 +321,6 @@ def _perform_download(
service_module = Services.load(service)
_check_cancel("before service instantiation")
try:
import inspect
@@ -367,29 +358,11 @@ def _perform_download(
original_download_dir = config.directories.downloads
_check_cancel("before download execution")
stdout_capture = StringIO()
stderr_capture = StringIO()
# The progress_sink (dl.build_job_progress_callables) owns the percentage; status changes
# are emitted here.
if progress_callback:
progress_callback({"progress": 0.0, "status": "starting"})
original_result = dl_instance.result
def result_with_progress(*args, **kwargs):
try:
progress_callback({"status": "downloading"})
result = original_result(*args, **kwargs)
progress_callback({"progress": 100.0, "status": "completed"})
return result
except Exception as e:
progress_callback({"progress": 0.0, "status": "failed", "error": str(e)})
raise
dl_instance.result = result_with_progress
# progress_sink (dl.build_job_progress_callables) owns percentage; terminal status
# (success/failure) is reported by the worker's result file.
try:
with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
dl_instance.result(
+1 -1
View File
@@ -60,7 +60,7 @@ def main(argv: list[str]) -> int:
log.error(f"Failed to write progress update: {e}")
output_files = _perform_download(
job_id, service, title_id, params, cancel_event=None, progress_callback=progress_callback
job_id, service, title_id, params, progress_callback=progress_callback
)
result = {"status": "success", "output_files": output_files}
-1
View File
@@ -1442,7 +1442,6 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
session.auth_status = AuthStatus.FAILED
session.auth_error = str(e)
bridge.status = AuthStatus.FAILED
bridge.error = str(e)
asyncio.create_task(_run_auth())
+8 -20
View File
@@ -25,8 +25,9 @@ class AuthStatus(Enum):
FAILED = "failed"
class BridgeCancelledError(Exception):
"""Raised when the bridge is cancelled (client disconnected, session deleted)."""
# Seconds the server waits for a remote client to answer an input prompt before
# giving up; also the TTL SessionStore grants AUTHENTICATING/PENDING_INPUT sessions.
AUTH_INPUT_TIMEOUT = 600.0
@dataclass
@@ -40,12 +41,11 @@ class InputBridge:
_prompt: Optional[str] = field(default=None, init=False, repr=False)
_response: Optional[str] = field(default=None, init=False, repr=False)
_status: AuthStatus = field(default=AuthStatus.AUTHENTICATING, init=False)
_error: Optional[str] = field(default=None, init=False, repr=False)
_cancelled: bool = field(default=False, init=False, repr=False)
_response_ready: threading.Event = field(default_factory=threading.Event, init=False, repr=False)
_lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False)
def request_input(self, prompt: str, timeout: float = 600.0) -> str:
def request_input(self, prompt: str, timeout: float = AUTH_INPUT_TIMEOUT) -> str:
"""Block until the remote client submits a response for *prompt*.
Args:
@@ -57,11 +57,11 @@ class InputBridge:
Raises:
TimeoutError: If no response is received within *timeout*.
BridgeCancelledError: If the bridge was cancelled.
RuntimeError: If the bridge was cancelled.
"""
with self._lock:
if self._cancelled:
raise BridgeCancelledError("Session was cancelled")
raise RuntimeError("Session was cancelled")
self._prompt = prompt
self._response = None
self._status = AuthStatus.PENDING_INPUT
@@ -73,16 +73,15 @@ class InputBridge:
break
with self._lock:
if self._cancelled:
raise BridgeCancelledError("Session was cancelled")
raise RuntimeError("Session was cancelled")
else:
with self._lock:
self._status = AuthStatus.FAILED
self._error = "Input request timed out waiting for client response"
raise TimeoutError(f"No client response for prompt within {timeout}s")
with self._lock:
if self._cancelled:
raise BridgeCancelledError("Session was cancelled")
raise RuntimeError("Session was cancelled")
response = self._response or ""
self._prompt = None
self._response = None
@@ -115,7 +114,6 @@ class InputBridge:
with self._lock:
self._cancelled = True
self._status = AuthStatus.FAILED
self._error = "Session cancelled"
self._response_ready.set()
@property
@@ -127,13 +125,3 @@ class InputBridge:
def status(self, value: AuthStatus) -> None:
with self._lock:
self._status = value
@property
def error(self) -> Optional[str]:
with self._lock:
return self._error
@error.setter
def error(self, value: Optional[str]) -> None:
with self._lock:
self._error = value
+2 -2
View File
@@ -14,7 +14,7 @@ from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from unshackle.core.api.input_bridge import AuthStatus, InputBridge
from unshackle.core.api.input_bridge import AUTH_INPUT_TIMEOUT, AuthStatus, InputBridge
from unshackle.core.api.sanitize import sanitize_log
from unshackle.core.config import config
from unshackle.core.tracks import Track
@@ -127,7 +127,7 @@ class SessionStore:
for sid, entry in self._sessions.items():
elapsed = (now - entry.last_accessed).total_seconds()
if entry.auth_status in (AuthStatus.AUTHENTICATING, AuthStatus.PENDING_INPUT):
if elapsed > 600:
if elapsed > AUTH_INPUT_TIMEOUT:
expired.append(sid)
elif elapsed > self._ttl:
expired.append(sid)
+2 -2
View File
@@ -198,7 +198,7 @@ def _build_tracks(data: Dict[str, Any]) -> Tracks:
return tracks
def _resolve_manifest_data(tracks: Tracks, manifests: list, session: Any) -> None:
def _resolve_manifest_data(tracks: Tracks, manifests: list) -> None:
"""Re-parse serialized manifests and populate track.data for downloading.
The server serializes DASH and ISM manifest XML as zlib-compressed base64.
@@ -604,7 +604,7 @@ class RemoteService:
for k, v in result.get("session_cookies", {}).items():
self._session.cookies.set(k, v)
_resolve_manifest_data(tracks, result.get("manifests", []), self._session)
_resolve_manifest_data(tracks, result.get("manifests", []))
self._server_cdm_type = result.get("server_cdm_type", "widevine")