diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 043d88a..5efe595 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -371,6 +371,45 @@ async def get_json( return await request_json(url, headers=headers, timeout=timeout) +async def post_empty( + url: str, + *, + headers: Optional[dict[str, str]] = None, + timeout: float = 5.0, +) -> None: + """ + POST an empty body to `url` and discard the response. + """ + try: + client_timeout = aiohttp.ClientTimeout(total=timeout) + connector = aiohttp.TCPConnector(ssl=False) + async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: + async with session.post(url, data=b"", headers=headers) as resp: + if resp.status >= 400: + raw = await resp.text() + _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items())) + await resp.read() + except (SignerRefreshRequired, SkipPaymentCycle, LivepeerGatewayError): + raise + except ConnectionRefusedError as e: + raise LivepeerGatewayError( + f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})" + ) from e + except getattr(aiohttp, "ClientConnectorError", ()) as e: + os_error = getattr(e, "os_error", None) + if isinstance(os_error, ConnectionRefusedError): + raise LivepeerGatewayError( + f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})" + ) from e + raise LivepeerGatewayError( + f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" + ) from e + except (aiohttp.ClientError, asyncio.TimeoutError) as e: + raise LivepeerGatewayError( + f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" + ) from e + + def _parse_http_url(url: str, *, context: str = "URL") -> ParseResult: """ Normalize a URL for HTTP(S) endpoints. diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index fbe3010..971c2c0 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import contextlib import inspect import json import logging @@ -29,7 +30,7 @@ from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import open_stream, post_json, request_json +from .http import open_stream, post_empty, post_json, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -49,6 +50,9 @@ "720p-pixel-seconds": "lv2v", "fixed": "fixed", } +# Metered types are billed for as long as the work runs, so they need ongoing +# payments. Fixed pricing is settled by the upfront payment alone. +_METERED_PAYMENT_TYPES = frozenset({"live", "lv2v"}) # golang format duration, eg "10s" _DURATION_RE = re.compile(r"^\s*(?P[0-9]+(?:\.[0-9]+)?)(?Pns|us|\u00b5s|ms|s|m|h)\s*$") @@ -104,12 +108,57 @@ class LiveRunnerInstance: price_info: Optional[LiveRunnerPriceInfo] = None -@dataclass(frozen=True) +@dataclass class LiveRunnerSession: + """A reserved live runner session. + + A metered session is billed for as long as it is held, so on-chain + sessions fund themselves from the moment they are reserved until they are + closed. Use it as an async context manager (or call ``aclose()``) to + release the session's resources. + """ + session_id: str app_url: str runner_url: str runner: Optional[LiveRunnerInstance] = None + control_url: str = "" + # True once the orchestrator reported this session gone. + released: bool = False + _payment_task: Optional[asyncio.Task[None]] = field( + default=None, repr=False, compare=False + ) + + @property + def payment_url(self) -> str: + """This session's payment endpoint, or "" when none was reported.""" + return _payment_endpoint(self.control_url) + + def _start_payments(self, payment_session: LivePaymentSession) -> None: + if self._payment_task is not None: + return + self._payment_task = _start_funding( + payment_session, + self.payment_url, + lambda: setattr(self, "released", True), + ) + + async def stop_payments(self) -> None: + """Stop funding this session without releasing it. + + Useful to hand funding to something else, or to let a session lapse + deliberately. Closing the session stops payments too. + """ + self._payment_task = await _stop_funding(self._payment_task) + + async def aclose(self) -> None: + await self.stop_payments() + + async def __aenter__(self) -> LiveRunnerSession: + return self + + async def __aexit__(self, *exc: object) -> None: + await self.aclose() @dataclass(frozen=True) @@ -147,6 +196,11 @@ class LiveRunnerCallStream: payment_session: Optional[LivePaymentSession] _session: aiohttp.ClientSession = field(repr=False, compare=False) _response: aiohttp.ClientResponse = field(repr=False, compare=False) + # True once the orchestrator reported the backing session gone. + released: bool = False + _payment_task: Optional[asyncio.Task[None]] = field( + default=None, repr=False, compare=False + ) @property def content_type(self) -> str: @@ -160,7 +214,22 @@ async def aiter_lines(self) -> AsyncIterator[str]: async for line in self._response.content: yield line.decode(errors="replace").rstrip("\n") + def _start_payments( + self, + payment_session: LivePaymentSession, + payment_url: str = "", + ) -> None: + if self._payment_task is not None: + return + self._payment_task = _start_funding( + payment_session, + payment_url, + lambda: setattr(self, "released", True), + ) + async def aclose(self) -> None: + # Stop funding first: don't pay for a stream we are about to drop. + self._payment_task = await _stop_funding(self._payment_task) self._response.release() await self._session.close() @@ -296,10 +365,10 @@ async def close(self) -> None: _LOG.warning("Skipping live runner unregister without heartbeat secret") return try: - await _post_empty( + await post_empty( _join_endpoint(self.orchestrator_url, f"/runners/{quote(self.runner_id, safe='')}/unregister"), - {"Authorization": secret}, - self._timeout, + headers={"Authorization": secret}, + timeout=self._timeout, ) except Exception: _LOG.debug("Live runner unregister failed", exc_info=True) @@ -755,6 +824,8 @@ async def call_runner( payment_session: Optional[LivePaymentSession] = None payment_type = "" session_id = "" + needs_ongoing_funding = False + payment_url = "" request_headers: dict[str, str] = {} if signer_url: request_headers[_LIVE_RUNNER_PAYER_ADDRESS_HEADER] = payer_address @@ -782,6 +853,17 @@ async def call_runner( request_headers["Livepeer-Segment"] = payment.seg_creds or "" session_id = challenge.manifest_id + # Metered pricing bills for as long as the work runs. + needs_ongoing_funding = payment_type in _METERED_PAYMENT_TYPES + if needs_ongoing_funding: + payment_url = _payment_endpoint( + _session_control_url( + challenge.orchestrator_url, + runner.runner_id if runner is not None else "", + session_id, + ) + ) + try: request_kwargs: dict[str, Any] = {"timeout": timeout} if request_headers: @@ -796,16 +878,37 @@ async def call_runner( payload=request_payload, headers=request_headers or None, ) - return LiveRunnerCallStream( - resp.status, resp.headers, runner_url, runner, payment_session, session, resp, + call_stream = LiveRunnerCallStream( + resp.status, + resp.headers, + runner_url, + runner, + None if payment_type == "fixed" else payment_session, + session, + resp, ) + # The stream outlives this call, so it owns the funding. + if needs_ongoing_funding: + call_stream._start_payments( + cast(LivePaymentSession, payment_session), payment_url + ) + return call_stream - data = await request_json( - runner_url, - method=method, - payload=request_payload, - **request_kwargs, + # The request ends with this call, so the funding ends with it. + pay_task = ( + _start_funding(cast(LivePaymentSession, payment_session), payment_url) + if needs_ongoing_funding + else None ) + try: + data = await request_json( + runner_url, + method=method, + payload=request_payload, + **request_kwargs, + ) + finally: + await _stop_funding(pay_task) if not isinstance(data, dict): raise LivepeerGatewayError( f"Live runner call expected JSON object, got {type(data).__name__}" @@ -863,6 +966,56 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC ) +def _session_control_url(orchestrator_url: str, runner_id: str, session_id: str) -> str: + """Rebuild the control URL an orchestrator reports when reserving a session. + + Single-shot calls never receive one, since a payment challenge carries only + the orchestrator, the runner and the session id. Returns "" if a part is + missing. + """ + if not (orchestrator_url and runner_id and session_id): + return "" + try: + return _join_endpoint( + orchestrator_url, + f"/apps/{quote(runner_id, safe='')}/session/{quote(session_id, safe='')}", + ) + except LivepeerGatewayError: + return "" + + +def _payment_endpoint(control_url: str) -> str: + """The session-scoped payment endpoint under a control URL. + + Unlike the generic ``/payment``, it 404s once the session is gone. Returns + "" without a control URL, so callers fall back to the generic endpoint. + """ + return _join_endpoint(control_url, "payment") if control_url else "" + + +def _start_funding( + payment_session: LivePaymentSession, + payment_url: str, + on_released: Optional[Callable[[], None]] = None, +) -> asyncio.Task[None]: + """Run payments in the background for as long as the caller keeps the task.""" + + async def _fund() -> None: + released = await payment_session.run_payments(payment_url=payment_url or None) + if released and on_released is not None: + on_released() + + return asyncio.create_task(_fund()) + + +async def _stop_funding(task: Optional[asyncio.Task[None]]) -> None: + if task is not None and not task.done(): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + return None + + async def _get_runner_payment( challenge: _RunnerPaymentChallenge, *, @@ -934,25 +1087,6 @@ def _live_runner_price_info_from_json(value: object) -> Optional[LiveRunnerPrice ) -def _live_runner_session_from_json( - data: dict[str, Any], - *, - runner_url: str, - runner: Optional[LiveRunnerInstance], -) -> LiveRunnerSession: - session_id = data.get("session_id") - app_url = data.get("app_url") - if not isinstance(session_id, str) or not session_id.strip(): - raise LivepeerGatewayError("Live runner session reserve response missing session_id") - if not isinstance(app_url, str) or not app_url.strip(): - raise LivepeerGatewayError("Live runner session reserve response missing app_url") - return LiveRunnerSession( - session_id=session_id.strip(), - app_url=app_url.strip(), - runner_url=runner_url, - runner=runner, - ) - async def stop_runner_session( session: LiveRunnerSession | LiveRunnerSessionRequest, *, @@ -977,10 +1111,10 @@ async def stop_runner_session( url = _join_endpoint(control_url, "stop") if isinstance(token, str) and token.strip(): request_headers = {"Livepeer-Session-Token": token} - await _post_empty( + await post_empty( url, - request_headers, - timeout, + headers=request_headers, + timeout=timeout, ) @@ -1146,25 +1280,6 @@ def _is_trickle_channel_response(value: object) -> bool: ) and ("internal_url" not in value or isinstance(value.get("internal_url"), str)) -async def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: - try: - client_timeout = aiohttp.ClientTimeout(total=timeout) - connector = aiohttp.TCPConnector(ssl=False) - async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: - async with session.post(url, data=b"", headers=headers) as resp: - body = await resp.text() - if resp.status >= 400: - raise LivepeerGatewayError( - f"HTTP empty POST error: HTTP {resp.status}; body={body!r}" - ) - except LivepeerGatewayError: - raise - except getattr(aiohttp, "ClientConnectorError", ()) as e: - raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e - except (aiohttp.ClientError, asyncio.TimeoutError) as e: - raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e - - def _detect_gpu_pynvml() -> Optional[LiveRunnerGPU]: try: import pynvml # type: ignore[import-not-found] diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index fc94463..38e1278 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -12,13 +12,20 @@ from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen -import aiohttp - from . import lp_rpc_pb2 from .async_cache import async_lru_cache -from .errors import LivepeerGatewayError, PaymentError, SignerRefreshRequired +from .errors import ( + LivepeerGatewayError, + LivepeerHTTPError, + PaymentError, + SignerRefreshRequired, + SkipPaymentCycle, +) _LOG = logging.getLogger(__name__) +# Must stay under the signer's opening payment: 10s per-second, 60s pixel. +PAYMENT_INTERVAL_S = 3.0 + @dataclass(frozen=True) class GetPaymentResponse: payment: str @@ -240,44 +247,70 @@ async def get_payment(self) -> GetPaymentResponse: await self._refresh_payment_params(orchestrator_url) attempts += 1 - async def send_payment(self, orchestrator_url: Optional[str] = None) -> None: + async def send_payment( + self, + orchestrator_url: Optional[str] = None, + *, + payment_url: Optional[str] = None, + ) -> None: + """Generate a payment and POST it to the orchestrator. + + ``payment_url`` targets a specific endpoint, such as the session-scoped + one which 404s once the session is released. Without it the payment + goes to the orchestrator's generic ``/payment`` endpoint, which credits + the payer balance blindly and cannot report a dead session. + + Raises LivepeerHTTPError on error responses so callers can branch on + the status code, and SkipPaymentCycle when the signer gates the cycle. + """ if not self._signer_url: return - target = orchestrator_url or self._orchestrator_url - if not target: - raise PaymentError("orchestrator_url is required before sending payment") + from .http import _http_origin, post_empty - from .http import _extract_error_message_from_body, _http_origin + if payment_url: + url = payment_url + else: + target = orchestrator_url or self._orchestrator_url + if not target: + raise PaymentError("orchestrator_url is required before sending payment") + url = f"{_http_origin(target)}/payment" payment = await self.get_payment() - url = f"{_http_origin(target)}/payment" + if not payment.seg_creds: + # An empty segment header fails the orchestrator's sig check and + # comes back 403, which reads as a dead session, not a bad signer. + raise PaymentError("Signer returned a payment with no segCreds") headers = { "Livepeer-Payment": payment.payment, "Livepeer-Segment": payment.seg_creds, } - try: - timeout = aiohttp.ClientTimeout(total=5.0) - async with aiohttp.ClientSession(timeout=timeout) as session: - async with session.post(url, data=b"", headers=headers) as resp: - if resp.status >= 400: - body = await resp.text() - message = _extract_error_message_from_body(body) - body_part = f"; body={message!r}" if message else "" - raise PaymentError( - f"HTTP payment error: HTTP {resp.status} from endpoint (url={url}){body_part}" - ) - await resp.read() - except PaymentError: - raise - except getattr(aiohttp, "ClientConnectorError", ()) as e: - raise PaymentError( - f"HTTP payment error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" - ) from e - except (aiohttp.ClientError, asyncio.TimeoutError) as e: - raise PaymentError( - f"HTTP payment error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" - ) from e + await post_empty(url, headers=headers, timeout=5.0) + + async def run_payments(self, *, payment_url: Optional[str] = None) -> bool: + """Keep a metered session funded until cancelled or the session ends. + + Cancel the task to stop; the first payment waits one interval, since + the caller pays upfront. Pass the session-scoped ``payment_url`` to + learn when the session is gone, since the generic ``/payment`` credits + blindly. Returns True if the orchestrator reported it gone. + """ + while True: + await asyncio.sleep(PAYMENT_INTERVAL_S) + try: + await self.send_payment(payment_url=payment_url) + except SkipPaymentCycle as e: + _LOG.debug("Payment loop skipped cycle: %s", e) + except LivepeerHTTPError as e: + # A 4xx will not change on a retry (404 gone, 409 fixed price, + # 403 mismatch), so stop rather than mint tickets nobody will + # honour. 408 and 429 are the two that do ask to be retried. + if 400 <= e.status_code < 500 and e.status_code not in (408, 429): + _LOG.info("Payment loop stopping (HTTP %d): %s", e.status_code, e) + return e.status_code == 404 + _LOG.warning("Payment failed; retrying next cycle: %s", e) + except Exception as e: + _LOG.warning("Payment failed; retrying next cycle: %s", e) async def _payment_request(self) -> GetPaymentResponse: from .http import _http_origin, post_json diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index 5cb8e82..755f5da 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -299,13 +299,19 @@ async def reserve_session( raise LivepeerGatewayError("runner session response missing session_id") if not isinstance(app_url, str) or not app_url.strip(): raise LivepeerGatewayError("runner session response missing app_url") - return LiveRunnerSession( + session = LiveRunnerSession( session_id=session_id.strip(), app_url=app_url.strip(), runner_url=result.runner_url, runner=result.runner, + control_url=_string_value(result.data.get("control_url")), ) + # No payment session means fixed price or offchain: nothing to fund. + if result.payment_session is not None: + session._start_payments(result.payment_session) + return session + def _runner_candidates_from_discovery(entries: Sequence[dict[str, Any]]) -> list[LiveRunnerInstance]: candidates: list[LiveRunnerInstance] = []