diff --git a/docs/dev/rest-api/dashboard.md b/docs/dev/rest-api/dashboard.md index 809a9c62..f5ea445b 100644 --- a/docs/dev/rest-api/dashboard.md +++ b/docs/dev/rest-api/dashboard.md @@ -39,11 +39,13 @@ unshackle serve --remote-only --quiet ## Endpoints -All are `GET`. +All are `GET`, except the two `drain` routes. | Path | Returns | |------|---------| -| `/api/dashboard/status` | Version, `code_hash`, bind host and port, `mode` (`full`, `api_only`, `remote_only`), uptime, `requests_total`, `requests_rejected` (401s), `requests_by_key` (label → count), `errors` (error counts by service tag, then by `error_code`), configured service count (the same services as `/api/dashboard/services`), session count, `max_sessions` (`null` when the operator configured no limit), `session_ttl`, job counts by status. The service count includes services that failed to import - use `/api/dashboard/services` for the number that loaded. | +| `/api/dashboard/status` | Version, `code_hash`, bind host and port, `mode` (`full`, `api_only`, `remote_only`), uptime, `requests_total`, `requests_rejected` (401s), `requests_by_key` (label → count), `errors` (error counts by service tag, then by `error_code`), configured service count (the same services as `/api/dashboard/services`), session count, `max_sessions` (`null` when the operator configured no limit), `session_ttl`, job counts by status. `draining` is `true` while a drain is on. The service count includes services that failed to import - use `/api/dashboard/services` for the number that loaded. | +| `POST /api/dashboard/drain` | Starts a drain before a restart and returns the `/status` body. The server then refuses each new remote session, download job and job retry with `SERVER_DRAINING` (409). Open remote sessions and jobs continue, and each open remote session gets a warning in its service log, which the client shows to the user. Poll `/status` until `sessions` and the queued and downloading jobs are 0, then restart. The drain ends when the process stops. A second call changes nothing. | +| `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=` 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. | diff --git a/docs/dev/rest-api/endpoints.md b/docs/dev/rest-api/endpoints.md index a633bcbf..bb8e02d6 100644 --- a/docs/dev/rest-api/endpoints.md +++ b/docs/dev/rest-api/endpoints.md @@ -131,7 +131,7 @@ Field notes: | Status | Meaning | | --- | --- | | `200` | Service list returned. | -| `500` | Unexpected server error (`INTERNAL_ERROR`). | +| `422` | Unexpected server error (`INTERNAL_ERROR`). | ### `POST /api/search` @@ -296,7 +296,7 @@ These keys are sent only when the title carries them, so a title without them se | `400` | `INVALID_SERVICE` | Unknown or disallowed service. | | `401` | `AUTH_FAILED` | Authentication with the service failed. | | `404` | `NOT_FOUND` | Title not found. | -| `500` | `INTERNAL_ERROR` | Unexpected error. | +| `422` | `INTERNAL_ERROR` | Unexpected error. | ### `POST /api/list-tracks` @@ -1290,7 +1290,7 @@ The server serialises every structured error raised inside a handler to the same The [error code catalogue](errors.md#error-code-catalogue) gives the default HTTP status of each error code. -When a generic (uncategorized) exception reaches a handler, unshackle inspects its message and type to pick a code. For example, messages mentioning "connection" or "timeout" become `NETWORK_ERROR` (503, retryable), "rate limit" or "429" become `RATE_LIMITED` (429, retryable), and anything unrecognized collapses to `INTERNAL_ERROR` (500). +When a generic (uncategorized) exception reaches a handler, unshackle inspects its message and type to pick a code. For example, messages mentioning "connection" or "timeout" become `NETWORK_ERROR` (424, retryable), "rate limit" or "429" become `RATE_LIMITED` (429, retryable), and anything unrecognized collapses to `INTERNAL_ERROR` (422). --- diff --git a/docs/dev/rest-api/errors.md b/docs/dev/rest-api/errors.md index d50ca08d..55b65774 100644 --- a/docs/dev/rest-api/errors.md +++ b/docs/dev/rest-api/errors.md @@ -100,15 +100,16 @@ When it is absent, assume the request will keep failing until you change somethi Every code below is a member of the `APIErrorCode` enumeration. The **Status** column is the default HTTP status the server returns for that code. A handler can override -the status in special cases (for example, a network error is always forced to `503`), -but the defaults hold for the overwhelming majority of responses. +the status in special cases, but the defaults hold for the overwhelming majority of responses. + +The server never answers with a `5xx` status. A proxy such as Cloudflare can replace the +body of a `5xx` with its own page, and the JSON error is then lost. Server-side failures +use `422`, and the ones that a retry can fix use `424`. Read `error_code` to tell them apart. +A `5xx` that a client receives comes from a proxy in front of the server, for example +while the server restarts. ### Client errors (4xx) -`SERVICE_ERROR` and `DRM_ERROR` are upstream failures, but they use `422`, not `502`. -A proxy such as Cloudflare replaces the body of a `502` or `504` with its own page, and -the JSON error is then lost. - | Code | Status | Meaning | |---|---|---| | `INVALID_INPUT` | 400 | A required field is missing, or the request body could not be parsed. Also used as the generic "your input was malformed" code. | @@ -125,6 +126,7 @@ the JSON error is then lost. | `SESSION_NOT_FOUND` | 404 | The remote-dl session does not exist or has expired. | | `TRACK_NOT_FOUND` | 404 | A referenced track ID is not present in the remote session. | | `CONFLICT` | 409 | The target resource is in a state that disallows the action (retrying a running job, prioritising a non-queued job, clearing cache during an active download). | +| `SERVER_DRAINING` | 409 | The server is about to restart and starts no new remote session or download job. Open work continues. Retryable after the restart. | | `SERVICE_ERROR` | 422 | The streaming service's API returned an error, or the service does not support the requested operation. | | `DRM_ERROR` | 422 | DRM or license acquisition failed. | | `RATE_LIMITED` | 429 | The streaming service is rate-limiting requests. Retryable. | @@ -138,19 +140,20 @@ the JSON error is then lost. the window is fixed, so an early retry only spends the next window. See [Authentication](authentication.md#rate-limit-rejections). -### Server errors (5xx) +### Server errors (422 and 424) | Code | Status | Meaning | |---|---|---| -| `INTERNAL_ERROR` | 500 | An unexpected server-side error. This is the fallback for any exception the server could not categorise. | -| `DOWNLOAD_ERROR` | 500 | The download process itself failed. | -| `WORKER_ERROR` | 500 | The download worker subprocess errored out. | -| `NETWORK_ERROR` | 503 | A network connectivity problem (connection, timeout, DNS, TLS). Retryable. | -| `SERVICE_UNAVAILABLE` | 503 | The service is temporarily unavailable or under maintenance. Retryable. | +| `INTERNAL_ERROR` | 422 | An unexpected server-side error. This is the fallback for any exception the server could not categorise, and for an exception that no handler caught. | +| `DOWNLOAD_ERROR` | 422 | The download process itself failed. | +| `WORKER_ERROR` | 422 | The download worker subprocess errored out. | +| `NETWORK_ERROR` | 424 | A network connectivity problem (connection, timeout, DNS, TLS). Retryable. | +| `SERVICE_UNAVAILABLE` | 424 | The service is temporarily unavailable or under maintenance. Retryable. | !!! note "Fallback status" If a code is ever emitted without a mapped status, the server falls back to - `500`. In practice every code above has an explicit mapping. + `422`. In practice every code above has an explicit mapping. A middleware + in every server mode also changes any `5xx` answer, such as one from a CDM subapplication, to `422`. ## How the server produces errors @@ -169,14 +172,14 @@ The classification is keyword-based, checked in order: | If the error mentions... | It becomes | Retryable | |---|---|---| | `auth`, `login`, `credential`, `unauthorized`, `forbidden`, `token` | `AUTH_FAILED` (401) | no | -| `connection`, `timeout`, `network`, `unreachable`, `socket`, `dns`, `resolve` (or `ConnectionError`, `TimeoutError`, `URLError`, `SSLError`) | `NETWORK_ERROR` (503) | yes | +| `connection`, `timeout`, `network`, `unreachable`, `socket`, `dns`, `resolve` (or `ConnectionError`, `TimeoutError`, `URLError`, `SSLError`) | `NETWORK_ERROR` (424) | yes | | `geofence`, `geoblock`, `region`, `not available in`, `territory` | `GEOFENCE` (403) | no | | `not found`, `404`, `does not exist`, `invalid id` | `NOT_FOUND` (404) | no | | `rate limit`, `too many requests`, `429`, `throttle` | `RATE_LIMITED` (429) | yes | | `drm`, `license`, `widevine`, `playready`, `decrypt` | `DRM_ERROR` (422) | no | -| `service unavailable`, `503`, `maintenance`, `temporarily unavailable` | `SERVICE_UNAVAILABLE` (503) | yes | +| `service unavailable`, `503`, `maintenance`, `temporarily unavailable` | `SERVICE_UNAVAILABLE` (424) | yes | | `invalid`, `malformed`, `validation` (or `ValueError`, `ValidationError`) | `INVALID_INPUT` (400) | no | -| anything else | `INTERNAL_ERROR` (500) | no | +| anything else | `INTERNAL_ERROR` (422) | no | Categorised errors include a `reason` tag inside `details` (such as `"network_connectivity"` or `"drm_failure"`) so a client can react without parsing diff --git a/docs/dev/rest-api/quickstart.md b/docs/dev/rest-api/quickstart.md index e344d5d4..f523c4a9 100644 --- a/docs/dev/rest-api/quickstart.md +++ b/docs/dev/rest-api/quickstart.md @@ -443,8 +443,8 @@ The `error_code` maps to an HTTP status. The ones you will meet most often: | `CONFLICT` | 409 | Illegal state transition (e.g. retry a running job) | | `SERVICE_ERROR` | 422 | The service raised an error | | `RATE_LIMITED` | 429 | Upstream throttling (retryable) | -| `INTERNAL_ERROR` | 500 | Unexpected server-side failure | -| `NETWORK_ERROR` | 503 | Connection/timeout reaching the service (retryable) | +| `INTERNAL_ERROR` | 422 | Unexpected server-side failure | +| `NETWORK_ERROR` | 424 | Connection/timeout reaching the service (retryable) | The [error code catalogue](errors.md#error-code-catalogue) lists every code. diff --git a/tests/remote/unit/test_dashboard.py b/tests/remote/unit/test_dashboard.py index d110953c..37174a55 100644 --- a/tests/remote/unit/test_dashboard.py +++ b/tests/remote/unit/test_dashboard.py @@ -281,6 +281,37 @@ async def test_dashboard_session_logs(aiohttp_client, dashboard_cfg) -> None: await store.delete(entry.session_id) +async def test_drain_refuses_new_work_and_warns_open_sessions(aiohttp_client, dashboard_cfg) -> None: + from unshackle.core.api.download_manager import get_download_manager + from unshackle.core.api.errors import APIError + from unshackle.core.api.session_log import SessionLogBuffer + from unshackle.core.api.session_store import get_session_store + + store = get_session_store() + entry = await store.create("EXAMPLE", object()) + entry.log_buffer = SessionLogBuffer() + client = await aiohttp_client(make_app()) + headers = {"X-Secret-Key": "dash-secret"} + try: + body = await (await client.post("/api/dashboard/drain", headers=headers)).json() + assert body["draining"] is True and body["sessions"] == 1 + await client.post("/api/dashboard/drain", headers=headers) + assert [r["level"] for r in entry.log_buffer.since(0)] == ["WARNING"] + + refused = await client.post("/api/session/create", json={}) + assert refused.status == 409 + assert (await refused.json())["error_code"] == "SERVER_DRAINING" + with pytest.raises(APIError, match="restarts for an update"): + get_download_manager().create_job("EXAMPLE", "t") + + body = await (await client.delete("/api/dashboard/drain", headers=headers)).json() + assert body["draining"] is False + assert (await client.post("/api/session/create", json={})).status == 400 + finally: + store.stop_drain() + await store.delete(entry.session_id) + + async def test_dashboard_log_read_does_not_touch_the_session() -> None: from unshackle.core.api.session_store import SessionStore diff --git a/tests/remote/unit/test_errors.py b/tests/remote/unit/test_errors.py index 6026281b..8e1be734 100644 --- a/tests/remote/unit/test_errors.py +++ b/tests/remote/unit/test_errors.py @@ -40,20 +40,21 @@ def test_api_error_default_http_status_per_code() -> None: APIErrorCode.TRACK_NOT_FOUND: 404, APIErrorCode.CONFLICT: 409, APIErrorCode.RATE_LIMITED: 429, - APIErrorCode.INTERNAL_ERROR: 500, - APIErrorCode.DOWNLOAD_ERROR: 500, - APIErrorCode.WORKER_ERROR: 500, + APIErrorCode.INTERNAL_ERROR: 422, + APIErrorCode.DOWNLOAD_ERROR: 422, + APIErrorCode.WORKER_ERROR: 422, APIErrorCode.SERVICE_ERROR: 422, APIErrorCode.DRM_ERROR: 422, - APIErrorCode.NETWORK_ERROR: 503, - APIErrorCode.SERVICE_UNAVAILABLE: 503, + APIErrorCode.NETWORK_ERROR: 424, + APIErrorCode.SERVICE_UNAVAILABLE: 424, + APIErrorCode.SERVER_DRAINING: 409, } # Lock that the cases above are the *complete* live set; a new/removed code trips this. assert {c.name for c in cases} == {c.name for c in APIErrorCode} for code, expected in cases.items(): assert APIError(code, "x").http_status == expected, code - # the edge proxy replaces a 502 or 504 body, so no code may default to either - assert not {502, 504} & set(cases.values()) + # the edge proxy can replace a 5xx body, so no code may default to one + assert all(status < 500 for status in cases.values()) def test_api_error_explicit_http_status_overrides_default() -> None: @@ -88,7 +89,7 @@ def test_build_error_response_retryable_flag() -> None: def test_build_error_response_from_generic_exception() -> None: resp = build_error_response(RuntimeError("oops")) - assert resp.status == 500 + assert resp.status == 422 body = _body(resp) assert body["error_code"] == "INTERNAL_ERROR" assert body["message"] == f"oops (ref {body['ref']})" @@ -161,15 +162,15 @@ def test_handle_api_exception_categorizes_generic() -> None: "exc, status, code, retryable", [ (Exception("Invalid credentials provided"), 401, "AUTH_FAILED", False), - (Exception("Connection refused"), 503, "NETWORK_ERROR", True), - (TimeoutError("read timeout"), 503, "NETWORK_ERROR", True), + (Exception("Connection refused"), 424, "NETWORK_ERROR", True), + (TimeoutError("read timeout"), 424, "NETWORK_ERROR", True), (Exception("Not available in your region"), 403, "GEOFENCE", False), (Exception("Title not found"), 404, "NOT_FOUND", False), (Exception("HTTP 429 too many requests"), 429, "RATE_LIMITED", True), (Exception("DRM license fetch failed"), 422, "DRM_ERROR", False), - (Exception("503 service unavailable"), 503, "SERVICE_UNAVAILABLE", True), + (Exception("503 service unavailable"), 424, "SERVICE_UNAVAILABLE", True), (ValueError("malformed body"), 400, "INVALID_INPUT", False), - (RuntimeError("totally novel failure xyz"), 500, "INTERNAL_ERROR", False), + (RuntimeError("totally novel failure xyz"), 422, "INTERNAL_ERROR", False), ], ) def test_error_contract_generic_exception(exc: Exception, status: int, code: str, retryable: bool) -> None: @@ -191,3 +192,29 @@ def test_error_contract_apierror_passthrough_full_shape() -> None: assert body["message"] == f"slow down (ref {body['ref']})" assert body["details"] == {"retry_after": 5} assert body["retryable"] is True + + +async def test_outer_middleware_never_answers_5xx(aiohttp_client) -> None: + from aiohttp import web + + from unshackle.core.api.routes import no_5xx_middleware + + async def broken(request: web.Request) -> web.Response: + raise KeyError(request.headers["X-Secret-Key"]) + + async def subapp_500(_request: web.Request) -> web.Response: + return web.json_response({"status": 500, "message": "device error"}, status=500) + + app = web.Application(middlewares=[no_5xx_middleware]) + app.router.add_get("/broken", broken) + app.router.add_get("/subapp", subapp_500) + client = await aiohttp_client(app) + + resp = await client.get("/broken", headers={"X-Secret-Key": "leaky-key"}) + assert resp.status == 422 + body = await resp.json() + assert body["error_code"] == "INTERNAL_ERROR" and "leaky-key" not in json.dumps(body) + + resp = await client.get("/subapp") + assert resp.status == 422 and (await resp.json())["message"] == "device error" + assert (await client.get("/missing")).status == 404 diff --git a/tests/remote/unit/test_remote_service_helpers.py b/tests/remote/unit/test_remote_service_helpers.py index 157eb395..3ee60c05 100644 --- a/tests/remote/unit/test_remote_service_helpers.py +++ b/tests/remote/unit/test_remote_service_helpers.py @@ -558,11 +558,12 @@ def test_session_create_writes_nested_cache_and_skips_colliding_keys(tmp_path, m "create_service_instance", lambda *args, **kwargs: (SimpleNamespace(log=logging.getLogger("Svc")), None, None), ) + monkeypatch.setattr("unshackle.core.api.session_store.refuse_while_draining", lambda: None) monkeypatch.setattr("unshackle.core.api.session_store.get_session_store", lambda: (_ for _ in ()).throw(Stop())) cache = {"x": blob('"flat"'), "x.json/y": blob('"collides"'), "MSL/keys.v2": blob('"dotted"')} response = asyncio.run(handlers.session_create_handler({"service": "TESTSVC", "title_id": "t", "cache": cache})) - assert response.status == 500 # the Stop stand-in fired after the cache write + assert response.status == 422 # the Stop stand-in fired after the cache write written = {p.name: p.read_text() for p in tmp_path.rglob("*.json") if p.is_file()} assert written == {"x.json": '"flat"', "keys.v2.json": '"dotted"'} diff --git a/unshackle/commands/serve.py b/unshackle/commands/serve.py index b3bc7c0d..1da14140 100644 --- a/unshackle/commands/serve.py +++ b/unshackle/commands/serve.py @@ -16,7 +16,7 @@ from aiohttp import web import unshackle.core.config as config_module from unshackle.core import binaries -from unshackle.core.api import cors_middleware, setup_routes, setup_swagger +from unshackle.core.api import cors_middleware, no_5xx_middleware, setup_routes, setup_swagger from unshackle.core.api.compression import compression_middleware from unshackle.core.api.events import publish_refresh_events, publish_service_event from unshackle.core.api.handlers import ( @@ -495,12 +495,19 @@ def serve( log.info("Starting REST API server (pywidevine/pyplayready CDM disabled)") if no_key: app = web.Application( - middlewares=[cors_middleware, stats_middleware, dashboard_authentication, compression_middleware] + middlewares=[ + cors_middleware, + no_5xx_middleware, + stats_middleware, + dashboard_authentication, + compression_middleware, + ] ) else: app = web.Application( middlewares=[ cors_middleware, + no_5xx_middleware, stats_middleware, dashboard_authentication, api_key_authentication, @@ -569,6 +576,7 @@ def serve( app = web.Application( middlewares=[ cors_middleware, + no_5xx_middleware, stats_middleware, cdm_log_middleware, dashboard_authentication, @@ -580,6 +588,7 @@ def serve( app = web.Application( middlewares=[ cors_middleware, + no_5xx_middleware, stats_middleware, cdm_log_middleware, dashboard_authentication, diff --git a/unshackle/core/api/__init__.py b/unshackle/core/api/__init__.py index 83698765..b38b32ef 100644 --- a/unshackle/core/api/__init__.py +++ b/unshackle/core/api/__init__.py @@ -1,3 +1,3 @@ -from unshackle.core.api.routes import cors_middleware, setup_routes, setup_swagger +from unshackle.core.api.routes import cors_middleware, no_5xx_middleware, setup_routes, setup_swagger -__all__ = ["setup_routes", "setup_swagger", "cors_middleware"] +__all__ = ["setup_routes", "setup_swagger", "cors_middleware", "no_5xx_middleware"] diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py index b3537b83..f914053e 100644 --- a/unshackle/core/api/download_manager.py +++ b/unshackle/core/api/download_manager.py @@ -735,6 +735,9 @@ class DownloadQueueManager: def create_job(self, service: str, title_id: str, owner_key: Optional[str] = None, **parameters) -> DownloadJob: """Make a new download job and add it to the queue.""" + from unshackle.core.api.session_store import refuse_while_draining + + refuse_while_draining() job_id = str(uuid.uuid4()) job = DownloadJob( job_id=job_id, diff --git a/unshackle/core/api/errors.py b/unshackle/core/api/errors.py index 7b798ca3..b554b435 100644 --- a/unshackle/core/api/errors.py +++ b/unshackle/core/api/errors.py @@ -47,6 +47,7 @@ class APIErrorCode(str, Enum): DRM_ERROR = "DRM_ERROR" DOWNLOAD_ERROR = "DOWNLOAD_ERROR" SERVICE_UNAVAILABLE = "SERVICE_UNAVAILABLE" + SERVER_DRAINING = "SERVER_DRAINING" WORKER_ERROR = "WORKER_ERROR" @@ -79,7 +80,11 @@ class APIError(Exception): @staticmethod def default_http_status(error_code: APIErrorCode) -> int: - """Map error codes to default HTTP status codes.""" + """Map error codes to default HTTP status codes. + + No code maps to a 5xx: a CDN in front of the server can replace a 5xx body with its own page. + Server-side failures are 422, and the ones worth a retry are 424, which the client retries. + """ status_map = { APIErrorCode.INVALID_INPUT: 400, APIErrorCode.INVALID_SERVICE: 400, @@ -95,16 +100,17 @@ class APIError(Exception): APIErrorCode.SESSION_NOT_FOUND: 404, APIErrorCode.TRACK_NOT_FOUND: 404, APIErrorCode.CONFLICT: 409, + APIErrorCode.SERVER_DRAINING: 409, APIErrorCode.SERVICE_ERROR: 422, APIErrorCode.DRM_ERROR: 422, + APIErrorCode.INTERNAL_ERROR: 422, + APIErrorCode.DOWNLOAD_ERROR: 422, + APIErrorCode.WORKER_ERROR: 422, + APIErrorCode.NETWORK_ERROR: 424, + APIErrorCode.SERVICE_UNAVAILABLE: 424, APIErrorCode.RATE_LIMITED: 429, - APIErrorCode.INTERNAL_ERROR: 500, - APIErrorCode.NETWORK_ERROR: 503, - APIErrorCode.SERVICE_UNAVAILABLE: 503, - APIErrorCode.DOWNLOAD_ERROR: 500, - APIErrorCode.WORKER_ERROR: 500, } - return status_map.get(error_code, 500) + return status_map.get(error_code, 422) def build_error_response( @@ -136,7 +142,7 @@ def build_error_response( error_code = APIErrorCode.INTERNAL_ERROR.value message = str(error) or "An unexpected error occurred" details = {} - http_status = 500 + http_status = APIError.default_http_status(APIErrorCode.INTERNAL_ERROR) retryable = False ref = secrets.token_hex(3) @@ -235,7 +241,6 @@ def categorize_exception( message=f"Network error occurred: {redact_secrets(str(exc))}", details={**context, "reason": "network_connectivity"}, retryable=True, - http_status=503, ) if any(keyword in exc_str for keyword in ["not found", "404", "does not exist", "invalid id"]): @@ -269,7 +274,6 @@ def categorize_exception( message=f"Service temporarily unavailable: {redact_secrets(str(exc))}", details={**context, "reason": "service_unavailable"}, retryable=True, - http_status=503, ) if any(keyword in exc_str for keyword in ["invalid", "malformed", "validation"]) or exc_type in [ diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index a0d51a5b..b2f4e4a2 100644 --- a/unshackle/core/api/handlers.py +++ b/unshackle/core/api/handlers.py @@ -2716,10 +2716,27 @@ async def dashboard_status_handler(request: web.Request) -> web.Response: "max_sessions": store.max_sessions, "session_ttl": store.ttl, "jobs": dict(jobs), + "draining": store.draining, } ) +async def dashboard_drain_start_handler(request: web.Request) -> web.Response: + """Start a drain before a restart; answer with the status body.""" + from unshackle.core.api.session_store import get_session_store + + get_session_store().start_drain() + return await dashboard_status_handler(request) + + +async def dashboard_drain_stop_handler(request: web.Request) -> web.Response: + """Cancel the drain; answer with the status body.""" + from unshackle.core.api.session_store import get_session_store + + get_session_store().stop_drain() + return await dashboard_status_handler(request) + + async def dashboard_sessions_handler(request: web.Request) -> web.Response: """Every live remote session, not owner-scoped.""" from unshackle.core.api.session_store import get_session_store @@ -3857,8 +3874,9 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req remote session on the server with the authenticated service instance, fetches all titles and tracks, and returns everything the client needs for track selection. """ - from unshackle.core.api.session_store import get_session_store + from unshackle.core.api.session_store import get_session_store, refuse_while_draining + refuse_while_draining() service_tag = data.get("service") title_id = data.get("title_id") profile = client_profile(data) @@ -5691,12 +5709,12 @@ async def tunnel_check_handler(request: web.Request) -> web.Response: try: info = await asyncio.to_thread(verify_proxy_exit, session) except ConnectionError as e: - raise APIError(APIErrorCode.INVALID_PROXY, str(e), http_status=502) + raise APIError(APIErrorCode.INVALID_PROXY, str(e), http_status=424) if not info: raise APIError( APIErrorCode.INVALID_PROXY, "The server could not look up an IP address through the tunnel.", - http_status=502, + http_status=424, ) ms = round((time.perf_counter() - started) * 1000) return web.json_response({"ip": info.get("ip"), "country": info.get("country"), "ms": ms}) diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index 0f83d616..128ab64f 100644 --- a/unshackle/core/api/routes.py +++ b/unshackle/core/api/routes.py @@ -22,6 +22,8 @@ from unshackle.core.api.handlers import ( clear_finished_download_jobs_handler, clear_temp_handler, dashboard_cdm_logs_handler, + dashboard_drain_start_handler, + dashboard_drain_stop_handler, dashboard_events_handler, dashboard_health_handler, dashboard_jobs_handler, @@ -69,6 +71,7 @@ from unshackle.core.api.handlers import ( tunnel_check_handler, tunnel_handler, ) +from unshackle.core.api.sanitize import sanitize_log from unshackle.core.drm import cbcs from unshackle.core.services import Services from unshackle.core.tunnel import CHECK_ROUTE as TUNNEL_CHECK_ROUTE @@ -94,6 +97,24 @@ async def cors_middleware( return response +@web.middleware +async def no_5xx_middleware( + request: web.Request, handler: Callable[[web.Request], Awaitable[web.StreamResponse]] +) -> web.StreamResponse: + """Answer an uncaught exception or any unsent 5xx (a CDM subapplication's own) as a 422; see errors.py.""" + try: + response = await handler(request) + except Exception as e: + if isinstance(e, web.HTTPException) and e.status < 500: + raise + # the exception text can hold a caller's key, so the answer gets a generic message + log.exception(f"Unhandled error in {sanitize_log(request.path)}") + response = build_error_response(APIError(APIErrorCode.INTERNAL_ERROR, "Internal server error")) + if response.status >= 500 and not response.prepared: + response.set_status(APIError.default_http_status(APIErrorCode.INTERNAL_ERROR)) + return response + + log = logging.getLogger("api") # Route handler signature: takes the request, returns a response (streamed or complete). @@ -291,7 +312,7 @@ async def services(request: web.Request) -> web.Response: type: string nullable: true description: Full service documentation - '500': + '422': description: Server error content: application/json: @@ -698,7 +719,7 @@ async def list_titles(request: web.Request) -> web.Response: timestamp: type: string format: date-time - '500': + '422': description: Server error content: application/json: @@ -1229,7 +1250,7 @@ async def download_jobs(request: web.Request) -> web.Response: type: number '400': description: Invalid query parameters - '500': + '422': description: Server error """ # Extract query parameters @@ -1261,7 +1282,7 @@ async def download_job_detail(request: web.Request) -> web.Response: description: Download job details '404': description: Job not found - '500': + '422': description: Server error """ job_id = request.match_info["job_id"] @@ -1301,7 +1322,7 @@ async def download_job_events(request: web.Request) -> web.StreamResponse: type: string '404': description: Job not found - '500': + '422': description: Server error """ job_id = request.match_info["job_id"] @@ -1330,7 +1351,7 @@ async def cancel_download_job(request: web.Request) -> web.Response: description: The server cannot cancel the job '404': description: Job not found - '500': + '422': description: Server error """ job_id = request.match_info["job_id"] @@ -1355,7 +1376,7 @@ async def clear_finished_download_jobs(request: web.Request) -> web.Response: removed: type: integer description: Number of jobs removed - '500': + '422': description: Server error """ return await clear_finished_download_jobs_handler(request) @@ -1392,7 +1413,7 @@ async def retry_download_job(request: web.Request) -> web.Response: description: Job not found '409': description: Job is not in a terminal state - '500': + '422': description: Server error """ job_id = request.match_info["job_id"] @@ -1429,7 +1450,7 @@ async def prioritize_download_job(request: web.Request) -> web.Response: description: Job not found '409': description: Job is not queued - '500': + '422': description: Server error """ job_id = request.match_info["job_id"] @@ -1517,7 +1538,7 @@ async def profiles(request: web.Request) -> web.Response: type: array items: type: string - '500': + '422': description: Server error """ return await profiles_handler(request) @@ -1585,7 +1606,7 @@ async def server_config(request: web.Request) -> web.Response: items: type: string description: Available service tags (allowlist-filtered) - '500': + '422': description: Server error """ return await server_config_handler(request) @@ -1656,7 +1677,7 @@ async def download_history(request: web.Request) -> web.Response: type: integer '400': description: Invalid query parameters - '500': + '422': description: Server error """ query_params = {"limit": request.query.get("limit"), "service": request.query.get("service")} @@ -1681,7 +1702,7 @@ async def delete_history(request: web.Request) -> web.Response: description: History entry removed '404': description: History entry not found - '500': + '422': description: Server error """ return await delete_history_handler(request.match_info["job_id"], request) @@ -1706,7 +1727,7 @@ async def maintenance_clear_cache(request: web.Request) -> web.Response: type: boolean freed_bytes: type: integer - '500': + '422': description: Server error """ return await clear_cache_handler(request) @@ -1731,7 +1752,7 @@ async def maintenance_clear_temp(request: web.Request) -> web.Response: type: boolean freed_bytes: type: integer - '500': + '422': description: Server error """ return await clear_temp_handler(request) @@ -1783,7 +1804,7 @@ async def maintenance_refresh_services(request: web.Request) -> web.Response: description: Services whose new code failed to import items: type: string - '500': + '422': description: Server error """ return await refresh_services_handler(request) @@ -1818,7 +1839,7 @@ async def env_check(request: web.Request) -> web.Response: nullable: true required: type: boolean - '500': + '422': description: Server error """ return await env_check_handler(request) @@ -2048,7 +2069,7 @@ async def tunnel_check(request: web.Request) -> web.Response: description: The exit IP and country through the tunnel, and the lookup time in ms '404': description: No open tunnel with this id for this API key - '502': + '424': description: No IP lookup got through the tunnel, or the requests left from the server itself """ return await tunnel_check_handler(request) @@ -2653,6 +2674,44 @@ async def dashboard_status(request: web.Request) -> web.Response: return await dashboard_status_handler(request) +@api_handler +async def dashboard_drain_start(request: web.Request) -> web.Response: + """ + Dashboard: start a drain before a restart. + --- + summary: Dashboard drain start + description: > + Refuses every new remote session, download job and job retry with `SERVER_DRAINING` (409) + until the server restarts or the operator cancels the drain. Open remote sessions continue, and + each one gets a warning in its service log. Poll `/api/dashboard/status` until `sessions` + and the active jobs reach 0, then restart. A second call changes nothing. + tags: [Dashboard] + responses: + '200': + description: The status body, with `draining` true + '401': + description: Dashboard key missing or invalid + """ + return await dashboard_drain_start_handler(request) + + +@api_handler +async def dashboard_drain_stop(request: web.Request) -> web.Response: + """ + Dashboard: cancel a drain. + --- + summary: Dashboard drain cancel + description: The server accepts new remote sessions and download jobs again. + tags: [Dashboard] + responses: + '200': + description: The status body, with `draining` false + '401': + description: Dashboard key missing or invalid + """ + return await dashboard_drain_stop_handler(request) + + @api_handler async def dashboard_sessions(request: web.Request) -> web.Response: """ @@ -3335,6 +3394,8 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [ DASHBOARD_ROUTES: list[tuple[str, str, Handler, bool]] = [ ("GET", DASHBOARD_PREFIX + "status", dashboard_status, True), + ("POST", DASHBOARD_PREFIX + "drain", dashboard_drain_start, True), + ("DELETE", DASHBOARD_PREFIX + "drain", dashboard_drain_stop, True), ("GET", DASHBOARD_PREFIX + "sessions", dashboard_sessions, True), ("GET", DASHBOARD_PREFIX + "jobs", dashboard_jobs, True), ("GET", DASHBOARD_PREFIX + "logs", dashboard_logs, True), diff --git a/unshackle/core/api/session_store.py b/unshackle/core/api/session_store.py index abfa865a..55c74313 100644 --- a/unshackle/core/api/session_store.py +++ b/unshackle/core/api/session_store.py @@ -24,6 +24,11 @@ from unshackle.core.tracks import Track log = logging.getLogger("api.session") +DRAIN_WARNING = ( + "This server restarts for an update when the open downloads finish. " + "This download continues. Start new downloads after the restart." +) + @dataclass class SessionEntry: @@ -129,6 +134,7 @@ class SessionStore: self._sessions: Dict[str, SessionEntry] = {} self._lock = asyncio.Lock() self._cleanup_task: Optional[asyncio.Task] = None + self.draining = False @property def ttl(self) -> int: @@ -322,6 +328,21 @@ class SessionStore: """Snapshot of every live session.""" return list(self._sessions.values()) + def start_drain(self) -> None: + """Refuse new sessions and jobs until the restart, and warn each open session through its service log.""" + if self.draining: + return + self.draining = True + log.warning(f"Draining: new sessions and jobs are refused, {len(self._sessions)} session(s) open") + for entry in self._sessions.values(): + if entry.log_buffer is not None: + entry.log_buffer.append(logging.WARNING, DRAIN_WARNING) + + def stop_drain(self) -> None: + if self.draining: + self.draining = False + log.warning("Drain cancelled: new sessions and jobs are accepted again") + @property def session_count(self) -> int: """Number of active sessions.""" @@ -331,6 +352,19 @@ class SessionStore: session_store: Optional[SessionStore] = None +def refuse_while_draining() -> None: + """Raise SERVER_DRAINING for a new session or job once the operator has started a drain.""" + from unshackle.core.api.errors import APIError, APIErrorCode + + if get_session_store().draining: + raise APIError( + APIErrorCode.SERVER_DRAINING, + "This server restarts for an update and does not start new sessions or downloads now. " + "Try again in a few minutes.", + retryable=True, + ) + + def get_session_store() -> SessionStore: """Get or make the global remote session store singleton.""" global session_store