diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py index 9b44713..144fe56 100644 --- a/src/livepeer_gateway/errors.py +++ b/src/livepeer_gateway/errors.py @@ -1,10 +1,99 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field +from typing import Literal + + +RunnerAttemptOutcome = Literal[ + "pending", # still in flight + "succeeded", + "payment_required", # runner answered 402; the next attempt pays + "refused", # runner answered 503: no capacity + "http", # runner answered with any other error status + "payment", # the signer failed, so the request never left + "unreachable", # the request never reached the runner + "timeout", # no answer in time; the runner may have the request + "disconnected", # the connection dropped after the request left + "invalid_response", + "cancelled", + "error", +] + + +@dataclass +class RunnerAttempt: + """One request ``call_runner`` made, or set out to make, to the runner.""" + + # The payment challenge this attempt answered; "" when it went out unpaid. + manifest_id: str = "" + # Signer auth id of the payment session this attempt opened. The signer + # charges when it signs, so this is set even if the request never left. + auth_id: str = "" + # True once the request left with Livepeer-Payment headers. + payment_sent: bool = False + # The runner's HTTP status, when known. + status_code: int | None = None + outcome: RunnerAttemptOutcome = "pending" + + +@dataclass +class RunnerCallRecord: + """Every attempt of one ``call_runner`` call, written as each step happens. + + The result, stream, or error of the call holds this same record. Pass one + to ``call_runner`` to read it even when the call is cancelled. + """ + + runner_url: str = "" + attempts: list[RunnerAttempt] = field(default_factory=list) + + @property + def payment_sent(self) -> bool: + """True once any attempt's request left with ``Livepeer-Payment`` headers. + + The call is then bound to this runner; another runner is safe only + while this is False. + """ + return any(attempt.payment_sent for attempt in self.attempts) + + @property + def auth_ids(self) -> tuple[str, ...]: + """Signer auth ids of every payment session the call opened, in order.""" + return tuple(dict.fromkeys(a.auth_id for a in self.attempts if a.auth_id)) + + @property + def outcome(self) -> RunnerAttemptOutcome: + """How the last attempt ended; ``pending`` before the first one.""" + return self.attempts[-1].outcome if self.attempts else "pending" class LivepeerGatewayError(RuntimeError): - """Base error for the library.""" + """Base error for the library. + + ``record`` is set on every error ``call_runner`` raises: the attempts it + made against the runner, with what was signed, sent, and how each ended. + ``payment_sent`` and ``auth_ids`` read from it, and are empty without one. + """ + + record: RunnerCallRecord | None = None + + @property + def payment_sent(self) -> bool: + """True once any attempt's request carried ``Livepeer-Payment`` headers. + + The call is then bound to this runner; another runner is safe only + while this is False. + """ + return self.record is not None and self.record.payment_sent + + @property + def auth_ids(self) -> tuple[str, ...]: + """Signer auth ids of every payment session the call opened. + + The signer charges when it signs, so these can be set while + ``payment_sent`` is still False. + """ + return self.record.auth_ids if self.record is not None else () class LivepeerHTTPError(LivepeerGatewayError): diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 699d042..c4bef5d 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -365,9 +365,11 @@ async def open_stream( f"HTTP stream error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" ) from e if resp.status >= 400: - raw = await resp.text() - resp.release() - await session.close() + try: + raw = await resp.text() + finally: + resp.release() + await session.close() _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items())) return session, resp diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 75faa28..5e16357 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -27,7 +27,14 @@ from aiohttp.helpers import parse_mimetype from .channel_reader import ChannelReader -from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired +from .errors import ( + LivepeerGatewayError, + LivepeerHTTPError, + RunnerAttempt, + RunnerAttemptOutcome, + RunnerCallRecord, + SignerRefreshRequired, +) from .http import _post_empty, _request_body, open_stream, post_json, request_json from .remote_signer import ( GetPaymentResponse, @@ -176,6 +183,23 @@ class LiveRunnerCallResult: # Non-JSON responses (an image, say) arrive unparsed in `content`; `data` stays empty. content: bytes | None = field(default=None, repr=False) content_type: str = "" + record: RunnerCallRecord = field( + default_factory=RunnerCallRecord, + repr=False, + compare=False, + ) + + @property + def payment_sent(self) -> bool: + return self.record.payment_sent + + @property + def auth_ids(self) -> tuple[str, ...]: + """Signer auth ids of every payment session the call opened. + + Fixed-price calls still list them, though their payment_session is not returned. + """ + return self.record.auth_ids @dataclass @@ -196,6 +220,15 @@ class LiveRunnerCallStream: _response: aiohttp.ClientResponse = field(repr=False, compare=False) # True once the orchestrator reported the backing session gone. released: bool = False + # The payment challenge's manifest_id, as LiveRunnerCallResult.session_id + # carries it for a unary call; "" when no challenge was answered. + # Correlation with the orchestrator, not a billing key. + session_id: str = "" + record: RunnerCallRecord = field( + default_factory=RunnerCallRecord, + repr=False, + compare=False, + ) _payment_task: asyncio.Task[None] | None = field( default=None, repr=False, compare=False ) @@ -204,6 +237,14 @@ class LiveRunnerCallStream: def content_type(self) -> str: return self._response.content_type + @property + def payment_sent(self) -> bool: + return self.record.payment_sent + + @property + def auth_ids(self) -> tuple[str, ...]: + return self.record.auth_ids + async def aiter_bytes(self) -> AsyncIterator[bytes]: async for chunk in self._response.content.iter_any(): yield chunk @@ -768,6 +809,7 @@ async def call_runner( payment_unit: str | None = ..., timeout: float = ..., max_payment_challenge_retries: int = ..., + record: RunnerCallRecord | None = ..., stream: Literal[False] = False, ) -> LiveRunnerCallResult: pass @@ -785,6 +827,7 @@ async def call_runner( payment_unit: str | None = ..., timeout: float = ..., max_payment_challenge_retries: int = ..., + record: RunnerCallRecord | None = ..., stream: Literal[True], ) -> LiveRunnerCallStream: pass @@ -801,22 +844,76 @@ async def call_runner( payment_unit: str | None = None, timeout: float = 5.0, max_payment_challenge_retries: int = 3, + record: RunnerCallRecord | None = None, stream: bool = False, ) -> LiveRunnerCallResult | LiveRunnerCallStream: """Call a runner once and return its result (or a live stream if ``stream=True``). With ``signer_url`` set, payment is automatic and **per call**: a 402 challenge is paid via the signer and retried (up to ``max_payment_challenge_retries``), one job, - one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. + one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors, and + ``LivepeerGatewayError`` on any other failure. + + Every attempt is written to ``record`` as it happens: what was signed, whether + the request left with payment, and how it ended. The result, stream, or error + holds that record, and its ``payment_sent`` says whether a gateway may still try + another runner. Pass ``record`` in to keep it readable if the call is cancelled. ``application/json`` and ``+json`` types parse into ``result.data``; anything else (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``. The request asks for no particular format, so the app picks what it returns. """ + if record is None: + record = RunnerCallRecord() + try: + return await _call_runner_attempts( + record, + runner_url, + runner=runner, + payload=payload, + method=method, + signer_url=signer_url, + signer_headers=signer_headers, + payment_unit=payment_unit, + timeout=timeout, + max_payment_challenge_retries=max_payment_challenge_retries, + stream=stream, + ) + except asyncio.CancelledError: + _end_attempt(record, "cancelled") + raise + except LivepeerGatewayError as e: + _end_attempt(record, _attempt_outcome(e)) + e.record = record + raise + except Exception as e: + _end_attempt(record, _attempt_outcome(e)) + error = LivepeerGatewayError( + f"Live runner call failed: {e.__class__.__name__}: {e} (url={record.runner_url})" + ) + error.record = record + raise error from e + + +async def _call_runner_attempts( + record: RunnerCallRecord, + runner_url: str, + *, + runner: LiveRunnerInstance | None, + payload: dict[str, Any] | None, + method: str, + signer_url: str | None, + signer_headers: dict[str, str] | None, + payment_unit: str | None, + timeout: float, + max_payment_challenge_retries: int, + stream: bool, +) -> LiveRunnerCallResult | LiveRunnerCallStream: runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "") if not runner_url: raise LivepeerGatewayError("Live runner call requires runner_url") + record.runner_url = runner_url request_payload = payload or {} payment_type = _runner_payment_type(runner, payment_unit) if signer_url else "" max_price: LiveRunnerPriceInfo | None = None @@ -827,8 +924,10 @@ async def call_runner( signer = await get_signer_info(signer_url, _freeze_headers(signer_headers)) payer_address = cast(str, signer.address) challenge: LivePaymentChallenge | None = None - attempts = (max(0, int(max_payment_challenge_retries)) + 1) * 2 - for attempt in range(attempts): + max_attempts = (max(0, int(max_payment_challenge_retries)) + 1) * 2 + for attempt_index in range(max_attempts): + attempt = RunnerAttempt() + record.attempts.append(attempt) payment_session: LivePaymentSession | None = None session_id = "" needs_ongoing_funding = False @@ -839,17 +938,21 @@ async def call_runner( request_headers[_LIVE_RUNNER_PAYER_ADDRESS_HEADER] = payer_address # Pending challenge means payment is needed. if challenge is not None: + attempt.manifest_id = challenge.manifest_id try: payment_session, payment = await _get_runner_payment( challenge, + attempt, payment_type=payment_type, app=runner.app if runner is not None else None, signer_url=signer_url or "", signer_headers=signer_headers, max_price=max_price, ) - except SignerRefreshRequired as e: - if attempt + 1 >= attempts: + except Exception as e: + # Any failure here is the signer's, never the runner's. + attempt.outcome = "payment" + if not isinstance(e, SignerRefreshRequired) or attempt_index + 1 >= max_attempts: raise # Could happen if embedded payment params expire; just retry in this case. _LOG.info( @@ -865,6 +968,8 @@ async def call_runner( # Metered pricing bills for as long as the work runs. needs_ongoing_funding = payment_type in _METERED_PAYMENT_TYPES + # Once tickets go out with the request, the call is bound to this runner. + attempt.payment_sent = "Livepeer-Payment" in request_headers try: request_kwargs: dict[str, Any] = {"timeout": timeout} if request_headers: @@ -879,6 +984,8 @@ async def call_runner( payload=request_payload, headers=request_headers or None, ) + attempt.status_code = resp.status + attempt.outcome = "succeeded" call_stream = LiveRunnerCallStream( resp.status, resp.headers, @@ -887,6 +994,8 @@ async def call_runner( None if payment_type == "fixed" else payment_session, session, resp, + session_id=session_id, + record=record, ) # The stream outlives this call, so it owns the funding. if needs_ongoing_funding: @@ -915,14 +1024,17 @@ async def call_runner( try: data = json.loads(body) except (UnicodeDecodeError, json.JSONDecodeError) as e: + attempt.outcome = "invalid_response" raise LivepeerGatewayError( f"HTTP JSON error: endpoint did not return valid JSON: {e} " f"(url={runner_url}, content_type={content_type})" ) from e if not isinstance(data, dict): + attempt.outcome = "invalid_response" raise LivepeerGatewayError( f"Live runner call expected JSON object, got {type(data).__name__}" ) + attempt.outcome = "succeeded" return LiveRunnerCallResult( data, runner_url=runner_url, @@ -934,18 +1046,58 @@ async def call_runner( payment_session=None if payment_type == "fixed" else payment_session, content=None if is_json else body, content_type=content_type, + record=record, ) except LivepeerHTTPError as e: + attempt.status_code = e.status_code if e.status_code != 402: + attempt.outcome = "refused" if e.status_code == 503 else "http" raise if not signer_url: + attempt.outcome = "payment_required" raise LivepeerGatewayError("Live runner paid call requires signer_url") from e - challenge = _parse_runner_payment_challenge(e) + try: + challenge = _parse_runner_payment_challenge(e) + except LivepeerGatewayError: + attempt.outcome = "invalid_response" + raise + attempt.outcome = "payment_required" continue raise LivepeerGatewayError("Live runner call exhausted payment challenge retries") +def _end_attempt(record: RunnerCallRecord, outcome: RunnerAttemptOutcome) -> None: + """Close the attempt still in flight, if any, with ``outcome``.""" + if record.attempts and record.attempts[-1].outcome == "pending": + record.attempts[-1].outcome = outcome + + +def _attempt_outcome(error: BaseException) -> RunnerAttemptOutcome: + """Classify a failed attempt by the network error behind it. + + Only errors that prove the request never left read as ``unreachable``. + aiohttp's total timeout raises a bare TimeoutError whether or not it had + connected, so that reads as ``timeout``: the runner may have the request. + """ + current: BaseException | None = error + while current is not None: + if isinstance( + current, + (aiohttp.ClientConnectorError, aiohttp.ConnectionTimeoutError, ConnectionRefusedError), + ): + return "unreachable" + if isinstance(current, TimeoutError): + return "timeout" + if isinstance( + current, + (aiohttp.ServerDisconnectedError, aiohttp.ClientPayloadError, aiohttp.ClientOSError, ConnectionError), + ): + return "disconnected" + current = current.__cause__ + return "error" + + def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> LivePaymentChallenge: try: data = json.loads(error.body) @@ -996,6 +1148,7 @@ async def _stop_funding(task: asyncio.Task[None] | None) -> None: async def _get_runner_payment( challenge: LivePaymentChallenge, + attempt: RunnerAttempt, *, payment_type: str, signer_url: str, @@ -1011,7 +1164,11 @@ async def _get_runner_payment( app=app, max_price=max_price.to_json() if max_price is not None else None, ) - payment = await session.get_payment() + try: + payment = await session.get_payment() + finally: + # The signer charges when it signs, so a signed session counts even on failure. + attempt.auth_id = session.auth_id if not payment.payment: raise LivepeerGatewayError("Live runner payment response missing payment") if not payment.seg_creds: diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index dd4554f..fed782f 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -6,6 +6,7 @@ import logging import re import ssl +from collections.abc import Mapping from dataclasses import dataclass, replace from functools import lru_cache from typing import Any, Optional @@ -26,6 +27,28 @@ # Must stay under the signer's opening payment: 10s per-second, 60s pixel. PAYMENT_INTERVAL_S = 3.0 + +def auth_id_from_signer_state(state: Mapping[str, Any] | None) -> str: + """Read the auth id sealed in a remote-signer payment state. + + go-livepeer returns that state as base64-encoded JSON under ``State``. + ``AuthID`` is the id its authorization webhook assigned. An unreadable + state, or one with no auth id yet, reads as empty. + """ + if not state: + return "" + raw = state.get("State") + if not isinstance(raw, str) or not raw: + return "" + try: + payload = json.loads(base64.standard_b64decode(raw)) + except (ValueError, UnicodeDecodeError): + return "" + if not isinstance(payload, dict): + return "" + auth_id = payload.get("AuthID") + return auth_id if isinstance(auth_id, str) else "" + @dataclass(frozen=True) class GetPaymentResponse: payment: str @@ -233,6 +256,11 @@ def __init__( self._max_refresh_retries = max(0, int(max_refresh_retries)) self._state: dict[str, Any] | None = None + @property + def auth_id(self) -> str: + """The signer's sealed auth id, empty until the first payment response.""" + return auth_id_from_signer_state(self._state) + async def get_payment(self) -> GetPaymentResponse: if not self._signer_url: return GetPaymentResponse(payment="", seg_creds=None) diff --git a/tests/test_live_payment_session.py b/tests/test_live_payment_session.py index 35afbb5..4e50b4b 100644 --- a/tests/test_live_payment_session.py +++ b/tests/test_live_payment_session.py @@ -1,5 +1,7 @@ from __future__ import annotations +import base64 +import json import types from unittest import mock @@ -14,6 +16,7 @@ LivePaymentChallenge, LivePaymentSession, PaymentSession, + auth_id_from_signer_state, get_signer_info, ) @@ -360,3 +363,41 @@ async def _post_json( assert first.address == "opaque-sender" assert first.sig == "opaque-signature" assert len(calls) == 1 + + async def test_auth_id_is_read_from_the_signed_state(self) -> None: + signed = base64.standard_b64encode( + json.dumps({"StateID": "state-1", "AuthID": "auth-1"}).encode() + ).decode() + + async def _post_json( + url: str, + payload: dict[str, object], + *, + headers: dict[str, str] | None = None, + timeout: float = 5.0, + ) -> dict[str, object]: + del url, payload, headers, timeout + return { + "payment": "payment", + "segCreds": "segment", + "state": {"State": signed, "Sig": "sig"}, + } + + with mock.patch("livepeer_gateway.http.post_json", side_effect=_post_json): + session = LivePaymentSession( + "https://signer.example.com", + type="fixed", + challenge=_challenge(), + ) + assert session.auth_id == "" + await session.get_payment() + + assert session.auth_id == "auth-1" + + +def test_auth_id_is_empty_when_the_signer_state_has_none() -> None: + assert auth_id_from_signer_state(None) == "" + assert auth_id_from_signer_state({"state": "lowercase"}) == "" + assert auth_id_from_signer_state({"State": "not-json"}) == "" + encoded = base64.standard_b64encode(json.dumps({"StateID": "state-1"}).encode()).decode() + assert auth_id_from_signer_state({"State": encoded}) == "" diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index 3cd2033..1ff7f3a 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -1,18 +1,23 @@ from __future__ import annotations import asyncio +import contextlib import json import os +import socket import sys +from collections.abc import AsyncIterator, Iterator from types import SimpleNamespace from unittest import mock import pytest +from aiohttp import web from livepeer_gateway import live_runner from livepeer_gateway.errors import ( LivepeerGatewayError, LivepeerHTTPError, + PaymentError, SignerRefreshRequired, ) from livepeer_gateway.live_runner import ( @@ -22,6 +27,7 @@ LiveRunnerPriceInfo, LiveRunnerRegistration, LiveRunnerSessionEvent, + RunnerCallRecord, register_runner, stop_runner_session, create_proxy, @@ -297,6 +303,8 @@ def __init__(self, signer_url: str, **kwargs: object) -> None: payment_sessions.append(self) sessions.append({"signer_url": signer_url, **kwargs}) + auth_id = "" + async def get_payment(self) -> object: return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64") @@ -392,6 +400,8 @@ async def test_paid_scope_runner_uses_lv2v_payment_type(self) -> None: ) class _PaymentSession: + auth_id = "" + def __init__(self, signer_url: str, **kwargs: object) -> None: sessions.append({"signer_url": signer_url, **kwargs}) @@ -465,6 +475,7 @@ async def test_paid_fixed_runner_retries_with_fresh_payment_and_no_renewal_sessi class _PaymentSession: def __init__(self, signer_url: str, **kwargs: object) -> None: sessions.append({"signer_url": signer_url, **kwargs}) + self.auth_id = f"auth-{len(sessions)}" async def get_payment(self) -> object: payment_number = len(sessions) @@ -528,6 +539,7 @@ def _request_body( assert sessions[0]["challenge"] == _payment_challenge("fixed-manifest") assert sessions[1]["challenge"] == _payment_challenge("fixed-manifest") assert result.payment_session is None + assert result.auth_ids == ("auth-1", "auth-2") async def test_paid_call_restarts_challenge_when_signer_requests_refresh( self, @@ -542,6 +554,8 @@ async def test_paid_call_restarts_challenge_when_signer_requests_refresh( runner_url = "https://service.example.com/apps/runner-1/session" class _PaymentSession: + auth_id = "" + def __init__(self, signer_url: str, **kwargs: object) -> None: payment_sessions.append(self) sessions.append({"signer_url": signer_url, **kwargs}) @@ -641,6 +655,8 @@ async def test_paid_call_repeated_refresh_requests_are_bounded(self) -> None: runner_url = "https://service.example.com/apps/runner-1/session" class _PaymentSession: + auth_id = "" + def __init__(self, signer_url: str, **kwargs: object) -> None: del signer_url, kwargs @@ -1833,3 +1849,326 @@ def test_torch_detects_current_device(self) -> None: gpu = live_runner._detect_gpu_torch() assert gpu == LiveRunnerGPU(id="1", name="Torch GPU", vram_mb=32) + + +# --------------------------------------------------------------------------- +# call_runner against a real runner. The signer is faked; the runner is a real +# aiohttp application on an ephemeral port. +# --------------------------------------------------------------------------- + + +@contextlib.asynccontextmanager +async def _serve(app: web.Application) -> AsyncIterator[str]: + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + port = site._server.sockets[0].getsockname()[1] # type: ignore[union-attr] + try: + yield f"http://127.0.0.1:{port}" + finally: + await runner.cleanup() + + +class _RecordedRequest: + def __init__(self, request: web.Request, body: bytes) -> None: + self.path = request.path + self.method = request.method + self.headers = dict(request.headers) + self.body = body + + +class _FakePaymentSession: + """Stands in for LivePaymentSession: returns fixed tickets, funds forever.""" + + instances: list[_FakePaymentSession] = [] + # What each session's get_payment raises, in order; None signs normally. + errors: list[Exception | None] = [] + + def __init__(self, signer_url: str, **kwargs: object) -> None: + self.signer_url = signer_url + self.kwargs = kwargs + self.auth_id = f"auth-{len(type(self).instances) + 1}" + type(self).instances.append(self) + + async def get_payment(self) -> object: + error = type(self).errors.pop(0) if type(self).errors else None + if error is not None: + raise error + return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64") + + async def run_payments(self) -> bool: + await asyncio.Event().wait() + return False + + +@contextlib.contextmanager +def _fake_signer(errors: list[Exception | None] | None = None) -> Iterator[None]: + _FakePaymentSession.instances = [] + _FakePaymentSession.errors = list(errors or []) + with ( + mock.patch.object(live_runner, "LivePaymentSession", _FakePaymentSession), + mock.patch.object( + live_runner, + "get_signer_info", + new_callable=mock.AsyncMock, + return_value=SimpleNamespace(address="opaque-payer", sig="opaque-signature"), + ), + ): + yield + + +def _runner_app( + seen: list[_RecordedRequest], + *, + statuses: list[int | str] | None = None, + stream: bool = False, +) -> web.Application: + """A runner that answers each request with the next status in ``statuses``. + + 402 carries a payment challenge; 200 answers JSON, or two SSE frames when + ``stream`` is set. ``"hang"`` never answers in time, and ``"truncated"`` + starts a 500 and drops the connection mid-body. + """ + pending = list(statuses or [200]) + + async def handler(request: web.Request) -> web.StreamResponse: + body = await request.read() + seen.append(_RecordedRequest(request, body)) + status = pending.pop(0) if len(pending) > 1 else pending[0] + if status == "hang": + await asyncio.sleep(1) + return web.Response(status=500) + if status == "truncated": + resp = web.StreamResponse(status=500) + resp.content_length = 1000 + await resp.prepare(request) + await resp.write(b"partial") + assert request.transport is not None + request.transport.close() + return resp + if status == 402: + return web.Response( + status=402, + text=_payment_challenge_body("manifest-1"), + content_type="application/json", + ) + if status != 200: + return web.Response(status=int(status), text="no") + if stream: + resp = web.StreamResponse(headers={"Content-Type": "text/event-stream"}) + await resp.prepare(request) + await resp.write(b"data: one\n\n") + await resp.write(b"data: two\n\n") + await resp.write_eof() + return resp + return web.json_response({"text": "hello", "n": len(seen)}) + + app = web.Application() + app.router.add_route("*", "/call", handler) + return app + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def _paid_call(base: str, **kwargs: object): + return call_runner( + f"{base}/call", + payload={"prompt": "hi"}, + signer_url="https://signer.example.com", + payment_unit="seconds", + **kwargs, + ) + + +class TestCallRunnerStreamSessionId: + async def test_stream_carries_manifest_id(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402, 200], stream=True)) as base: + with _fake_signer(): + async with await _paid_call(base, stream=True) as stream: + assert stream.session_id == "manifest-1" + assert stream.auth_ids == ("auth-1",) + assert stream.payment_session is _FakePaymentSession.instances[0] + lines = [line async for line in stream.aiter_lines() if line] + + assert lines == ["data: one", "data: two"] + assert len(seen) == 2 + + async def test_stream_without_signer_has_empty_session_id(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, stream=True)) as base: + async with await call_runner(f"{base}/call", payload={}, stream=True) as stream: + assert stream.session_id == "" + + +class TestCallRunnerPaymentSent: + @pytest.mark.parametrize("status", [500, 400]) + async def test_error_after_payment_carries_the_auth_id(self, status: int) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402, status])) as base: + with _fake_signer(): + with pytest.raises(LivepeerHTTPError) as info: + await _paid_call(base) + assert info.value.status_code == status + assert info.value.payment_sent is True + assert info.value.auth_ids == ("auth-1",) + assert len(seen) == 2 + + @pytest.mark.parametrize("status", [400, 503]) + async def test_error_before_payment_has_no_auth_id(self, status: int) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[status])) as base: + with _fake_signer(): + with pytest.raises(LivepeerHTTPError) as info: + await _paid_call(base) + assert info.value.status_code == status + assert info.value.payment_sent is False + assert info.value.auth_ids == () + assert len(seen) == 1 + + async def test_connection_refused_before_challenge(self) -> None: + with pytest.raises(LivepeerGatewayError, match="connection refused") as info: + await call_runner(f"http://127.0.0.1:{_free_port()}/call", payload={"prompt": "hi"}) + assert info.value.payment_sent is False + + async def test_stream_error_after_payment_says_so(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402, 500])) as base: + with _fake_signer(): + with pytest.raises(LivepeerHTTPError) as info: + await _paid_call(base, stream=True) + assert info.value.status_code == 500 + assert info.value.payment_sent is True + assert info.value.auth_ids == ("auth-1",) + + async def test_exhausted_retries_after_payment_says_so(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402])) as base: + with _fake_signer(): + with pytest.raises(LivepeerGatewayError, match="exhausted") as info: + await _paid_call(base, max_payment_challenge_retries=1) + assert info.value.payment_sent is True + assert info.value.auth_ids == ("auth-1", "auth-2", "auth-3") + + async def test_paid_call_without_signer_is_unpaid(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402])) as base: + with pytest.raises(LivepeerGatewayError, match="requires signer_url") as info: + await call_runner(f"{base}/call", payload={"prompt": "hi"}) + assert info.value.payment_sent is False + assert info.value.auth_ids == () + + +class TestCallRunnerRecord: + async def test_attempts_record_the_challenge_and_success(self) -> None: + seen: list[_RecordedRequest] = [] + record = RunnerCallRecord() + async with _serve(_runner_app(seen, statuses=[402, 200])) as base: + with _fake_signer(): + result = await _paid_call(base, record=record) + assert result.record is record + assert result.payment_sent is True + assert record.runner_url == f"{base}/call" + assert [a.outcome for a in record.attempts] == ["payment_required", "succeeded"] + assert [a.status_code for a in record.attempts] == [402, None] + assert [a.payment_sent for a in record.attempts] == [False, True] + assert [a.manifest_id for a in record.attempts] == ["", "manifest-1"] + assert [a.auth_id for a in record.attempts] == ["", "auth-1"] + + async def test_signer_failure_after_a_paid_attempt_still_says_paid(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402])) as base: + with _fake_signer([None, PaymentError("signer down")]): + with pytest.raises(PaymentError) as info: + await _paid_call(base) + record = info.value.record + assert record is not None + assert [a.outcome for a in record.attempts] == ["payment_required", "payment_required", "payment"] + assert info.value.payment_sent is True + assert info.value.auth_ids == ("auth-1", "auth-2") + + async def test_unpaid_retry_after_refresh_still_says_paid(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402, 402, 500])) as base: + with _fake_signer([None, SignerRefreshRequired("refresh")]): + with pytest.raises(LivepeerHTTPError) as info: + await _paid_call(base) + assert info.value.status_code == 500 + assert info.value.payment_sent is True + assert info.value.auth_ids == ("auth-1", "auth-2") + assert info.value.record is not None + assert info.value.record.attempts[-1].payment_sent is False + assert info.value.record.outcome == "http" + + async def test_refresh_on_the_last_attempt_keeps_earlier_auth_ids(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402])) as base: + with _fake_signer([None, None, SignerRefreshRequired("refresh")]): + with pytest.raises(SignerRefreshRequired) as info: + await _paid_call(base, max_payment_challenge_retries=1) + assert info.value.payment_sent is True + assert info.value.auth_ids == ("auth-1", "auth-2", "auth-3") + assert info.value.record is not None + assert info.value.record.outcome == "payment" + + async def test_truncated_stream_error_after_payment_is_wrapped(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[402, "truncated"], stream=True)) as base: + with _fake_signer(): + with pytest.raises(LivepeerGatewayError, match="Live runner call failed") as info: + await _paid_call(base, stream=True) + assert not isinstance(info.value, LivepeerHTTPError) + assert info.value.payment_sent is True + assert info.value.auth_ids == ("auth-1",) + assert info.value.record is not None + assert info.value.record.outcome == "disconnected" + + async def test_cancelled_call_leaves_the_record_readable(self) -> None: + seen: list[_RecordedRequest] = [] + record = RunnerCallRecord() + async with _serve(_runner_app(seen, statuses=[402, "hang"])) as base: + with _fake_signer(): + with pytest.raises(TimeoutError): + await asyncio.wait_for(_paid_call(base, record=record), 0.5) + assert record.payment_sent is True + assert record.auth_ids == ("auth-1",) + assert record.outcome == "cancelled" + + async def test_refused_connection_is_unreachable(self) -> None: + with pytest.raises(LivepeerGatewayError) as info: + await call_runner(f"http://127.0.0.1:{_free_port()}/call", payload={}) + assert info.value.record is not None + assert info.value.record.outcome == "unreachable" + assert info.value.payment_sent is False + + async def test_slow_runner_is_a_timeout_not_unreachable(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=["hang"])) as base: + with pytest.raises(LivepeerGatewayError) as info: + await call_runner(f"{base}/call", payload={}, timeout=0.2) + assert len(seen) == 1 + assert info.value.record is not None + assert info.value.record.outcome == "timeout" + + @pytest.mark.parametrize(("status", "outcome"), [(503, "refused"), (400, "http")]) + async def test_runner_error_status_outcomes(self, status: int, outcome: str) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, statuses=[status])) as base: + with pytest.raises(LivepeerHTTPError) as info: + await call_runner(f"{base}/call", payload={}) + assert info.value.record is not None + assert info.value.record.outcome == outcome + assert info.value.record.attempts[0].status_code == status + + async def test_stream_records_its_status(self) -> None: + seen: list[_RecordedRequest] = [] + async with _serve(_runner_app(seen, stream=True)) as base: + async with await call_runner(f"{base}/call", payload={}, stream=True) as stream: + assert stream.record.outcome == "succeeded" + assert stream.record.attempts[0].status_code == 200 + assert stream.payment_sent is False