diff --git a/src/livepeer_gateway/byoc.py b/src/livepeer_gateway/byoc.py index 272d668..3c6ca64 100644 --- a/src/livepeer_gateway/byoc.py +++ b/src/livepeer_gateway/byoc.py @@ -144,6 +144,29 @@ def audio_url(self) -> Optional[str]: # Header building # --------------------------------------------------------------------------- +def _read_http_error_body(e: HTTPError, limit: int = 200) -> str: + """ + Read an HTTPError body without letting the read itself raise. + + When a server rejects a request early (the signer answering 401 before + consuming a multi-KB POST body) and closes the connection, the client's + buffered response can be truncated: `e.read()` then raises + http.client.IncompleteRead *inside* the except-handler, the status code + never reaches the error message, and callers see only + "payment failed: IncompleteRead(84 bytes read, 109 more expected)" + (live incident 2026-08-21 — a dead key misreported as a transient GPU + outage). Salvage whatever bytes arrived (IncompleteRead.partial) so the + caller can always report "HTTP : ". + """ + try: + return e.read().decode("utf-8", errors="replace")[:limit] + except Exception as read_err: + partial = getattr(read_err, "partial", b"") + if partial: + return partial.decode("utf-8", errors="replace")[:limit] + return f"" + + def _create_byoc_payment( *, orch_origin: str, @@ -209,7 +232,15 @@ def _create_byoc_payment( with urlopen(payment_req, timeout=timeout) as resp: payment_data = json.loads(resp.read()) except HTTPError as e: - body = e.read().decode("utf-8", errors="replace")[:200] + body = _read_http_error_body(e) + # 401 = the signer rejected this API key outright (invalid/revoked). + # Name it explicitly so downstream error classifiers can tell it + # apart from a transient outage. 403 (out of credits) keeps the + # generic shape below — classifiers already match on that. + if e.code == 401: + raise LivepeerGatewayError( + f"signer rejected key: HTTP 401: {body}" + ) from e raise LivepeerGatewayError(f"BYOC payment generation failed: HTTP {e.code}: {body}") from e result = {} @@ -262,7 +293,11 @@ def _sign_byoc_job( with urlopen(req, timeout=30.0) as resp: return json.loads(resp.read()) except HTTPError as e: - body = e.read().decode("utf-8", errors="replace")[:200] + body = _read_http_error_body(e) + if e.code == 401: + raise LivepeerGatewayError( + f"sign-byoc-job: signer rejected key: HTTP 401: {body}" + ) from e raise LivepeerGatewayError(f"sign-byoc-job failed: HTTP {e.code}: {body}") from e @@ -420,11 +455,7 @@ def submit_byoc_job( ) except HTTPError as e: - err_body = "" - try: - err_body = e.read().decode("utf-8", errors="replace")[:500] - except Exception: - pass + err_body = _read_http_error_body(e, limit=500) reason = f"HTTP {e.code}: {err_body}" _LOG.warning("BYOC job %s: orchestrator %s rejected: %s", job_id, orch_origin, reason) @@ -701,11 +732,7 @@ def submit_training_job( ) except HTTPError as e: - err_body = "" - try: - err_body = e.read().decode("utf-8", errors="replace")[:500] - except Exception: - pass + err_body = _read_http_error_body(e, limit=500) reason = f"HTTP {e.code}: {err_body}" _LOG.warning("Training job %s: orchestrator %s rejected: %s", job_id, orch_origin, reason) @@ -836,11 +863,7 @@ def refresh_training_payment( last_err = e # HTTP 4xx (other than 408/429) are not transient — fail fast if isinstance(e, HTTPError) and e.code not in (408, 429, 502, 503, 504): - err_body = "" - try: - err_body = e.read().decode("utf-8", errors="replace")[:200] - except Exception: - pass + err_body = _read_http_error_body(e) raise LivepeerGatewayError( f"Training refresh permanent failure for {job_id}: " f"HTTP {e.code}: {err_body}" diff --git a/tests/test_http_error_body.py b/tests/test_http_error_body.py new file mode 100644 index 0000000..e5391c2 --- /dev/null +++ b/tests/test_http_error_body.py @@ -0,0 +1,121 @@ +""" +Signer/orch HTTPError bodies must never mask the status code. + +Live incident (2026-08-21, cjob_0a9056941b2a): the signer rejected a dead +Daydream key at /generate-live-payment with 401 and a 193-byte JSON body. +Because the signer answers before consuming the multi-KB POST body and then +closes the connection, the client's buffered response can be truncated — +`e.read()` inside the `except HTTPError` handler raised +`http.client.IncompleteRead(84 bytes read, 109 more expected)` (84+109=193), +which escaped the handler and surfaced to users as +"payment failed: IncompleteRead(...)" — with no trace of the 401. Downstream +that classified as "GPU network briefly busy — retry", the exact opposite of +a permanent per-key auth failure. + +These tests pin the two guarantees of the fix: + 1. `_read_http_error_body` never raises and salvages partial bytes. + 2. The signer paths report the status code first ("signer rejected key: + HTTP 401: ...") even when the body read dies mid-flight. +""" + +import io +import json +from http.client import IncompleteRead +from urllib.error import HTTPError + +import pytest + +from livepeer_gateway.byoc import _read_http_error_body, _sign_byoc_job +from livepeer_gateway.errors import LivepeerGatewayError + +SIGNER_401_BODY = ( + b'{"success":false,"error":"Authentication failed","code":"AUTH/FAILED",' + b'"status":401,"details":{"cause":"Invalid access token"}}' +) + + +def _http_error(code: int, fp) -> HTTPError: + return HTTPError("https://signer.example/generate-live-payment", code, "x", {}, fp) + + +class _TruncatingBody(io.RawIOBase): + """A body whose read dies mid-flight, like a connection reset.""" + + def __init__(self, partial: bytes): + self._partial = partial + + def read(self, *a): # noqa: ANN002 - match file-like signature + raise IncompleteRead(self._partial, expected=109) + + +class _ExplodingBody(io.RawIOBase): + def read(self, *a): # noqa: ANN002 + raise ConnectionResetError("peer reset") + + +class TestReadHttpErrorBody: + def test_reads_a_healthy_body(self): + e = _http_error(403, io.BytesIO(b'{"error":"nope"}')) + assert _read_http_error_body(e) == '{"error":"nope"}' + + def test_salvages_incomplete_read_partial(self): + # The real failure: 84 of 193 bytes arrive before the reset. The + # salvaged prefix still names the failure ("Authentication failed"). + e = _http_error(401, _TruncatingBody(SIGNER_401_BODY[:84])) + body = _read_http_error_body(e) + assert "Authentication failed" in body + + def test_never_raises_even_with_no_salvageable_bytes(self): + e = _http_error(401, _ExplodingBody()) + body = _read_http_error_body(e) + assert "ConnectionResetError" in body + + def test_truncates_to_limit(self): + e = _http_error(500, io.BytesIO(b"x" * 1000)) + assert len(_read_http_error_body(e, limit=200)) == 200 + + +class TestSignByocJobSignerRejection: + """End-to-end through a real except-handler: the status must survive.""" + + def _run(self, monkeypatch, error: HTTPError) -> LivepeerGatewayError: + def fake_urlopen(req, timeout=None, context=None): + raise error + + monkeypatch.setattr("livepeer_gateway.byoc.urlopen", fake_urlopen) + with pytest.raises(LivepeerGatewayError) as exc_info: + _sign_byoc_job( + signer_url="https://signer.example", + signer_headers=None, + job_id="job-1", + capability="flux-schnell", + request_json="{}", + parameters_json="", + timeout_seconds=30, + ) + return exc_info.value + + def test_401_with_truncated_body_reports_signer_rejected_key(self, monkeypatch): + err = self._run( + monkeypatch, _http_error(401, _TruncatingBody(SIGNER_401_BODY[:84])) + ) + msg = str(err) + assert "signer rejected key" in msg + assert "HTTP 401" in msg + assert "IncompleteRead" not in msg.split("HTTP 401")[0] # status leads + + def test_403_keeps_the_existing_message_shape(self, monkeypatch): + # Downstream classifiers match "failed: HTTP 403" for the + # out-of-credits case — that shape must not change. + err = self._run( + monkeypatch, + _http_error(403, io.BytesIO(b'{"error":{"message":"signer auth rejected request with status 403"}}')), + ) + msg = str(err) + assert "HTTP 403" in msg + assert "signer auth rejected" in msg + + def test_500_reports_status_and_body(self, monkeypatch): + err = self._run(monkeypatch, _http_error(500, io.BytesIO(b"boom"))) + msg = str(err) + assert "HTTP 500" in msg and "boom" in msg