diff --git a/examples/text/README.md b/examples/text/README.md index 7720e9f..f48eff0 100644 --- a/examples/text/README.md +++ b/examples/text/README.md @@ -37,3 +37,12 @@ Verify the runner registration with go-livepeer: ``` curl http://localhost:8935/discovery | jq ``` + +Or stream through the SDK client, which self-funds metered calls when a +remote signer is given (pays the 402 challenge, then keeps paying on a +cadence for as long as the stream runs): + +```sh +uv run client.py http://localhost:8935/apps/story-runner/app/sse +uv run client.py https://orch:8935/apps/story-runner/app/sse --signer-url http://localhost:7936 +``` diff --git a/examples/text/client.py b/examples/text/client.py new file mode 100644 index 0000000..8e65820 --- /dev/null +++ b/examples/text/client.py @@ -0,0 +1,48 @@ +"""SDK client for the text stream demo: a self-funding single-shot SSE call. + +Offchain (no payments): + + uv run client.py http://localhost:8935/apps/story-runner/app/sse + +On-chain: pass a remote signer. call_runner pays the 402 challenge and, +for metered pricing (hour/seconds or 720p), keeps funding the call on a +cadence while the stream runs — no payment code needed here. + + uv run client.py https://orch:8935/apps/story-runner/app/sse \ + --signer-url http://localhost:7936 +""" +from __future__ import annotations + +import argparse +import asyncio + +from livepeer_gateway.live_runner import call_runner + + +async def main() -> None: + p = argparse.ArgumentParser(description="Stream a story from a single-shot runner.") + p.add_argument("runner_url", help="Runner app endpoint, e.g. .../apps/story-runner/app/sse") + p.add_argument("--signer-url", default=None, help="Remote signer URL. Omit for offchain.") + p.add_argument( + "--payment-unit", + default=None, + help="hour|seconds|720p|720p-pixel-seconds|fixed. Default: metered (live).", + ) + args = p.parse_args() + + stream = await call_runner( + args.runner_url, + method="GET", + signer_url=args.signer_url, + payment_unit=args.payment_unit, + stream=True, + ) + async with stream: + async for line in stream.aiter_lines(): + print(line) + if stream.released: + print("(session released by orchestrator)") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 043d88a..db4057b 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -371,6 +371,50 @@ 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. + + Certificate verification is disabled, matching every other HTTP helper in + the SDK. Error responses raise the same typed errors as request_json + (SignerRefreshRequired for 480, SkipPaymentCycle for 482, LivepeerHTTPError + otherwise) so callers can branch on status codes. + """ + 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..dd88536 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 @@ -28,8 +29,13 @@ import aiohttp from .channel_reader import ChannelReader -from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import open_stream, post_json, request_json +from .errors import ( + LivepeerGatewayError, + LivepeerHTTPError, + SignerRefreshRequired, + SkipPaymentCycle, +) +from .http import open_stream, post_empty, post_json, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -49,6 +55,16 @@ "720p-pixel-seconds": "lv2v", "fixed": "fixed", } +# Metered payment types need ongoing funding while a call is in flight; the +# orchestrator debits the prepaid balance on -livePaymentInterval (5s default) +# and cancels the request on the first under-funded tick. +_METERED_PAYMENT_TYPES = frozenset({"live", "lv2v"}) +# Fallback client payment cadence when the challenge does not advertise +# payment_interval_ms (go-livepeer master does not, yet). +_DEFAULT_PAYMENT_INTERVAL_S = 3.0 +# Pay at a fraction of the server debit tick so a payment always lands +# before the next debit. +_PAYMENT_CADENCE_FRACTION = 0.6 # golang format duration, eg "10s" _DURATION_RE = re.compile(r"^\s*(?P[0-9]+(?:\.[0-9]+)?)(?Pns|us|\u00b5s|ms|s|m|h)\s*$") @@ -129,6 +145,9 @@ class LiveRunnerCallResult: repr=False, compare=False, ) + # Orchestrator debit cadence in seconds, from the payment challenge's + # payment_interval_ms. None when the orchestrator does not report it. + server_payment_interval: Optional[float] = None @dataclass @@ -145,8 +164,22 @@ class LiveRunnerCallStream: runner_url: str runner: Optional[LiveRunnerInstance] payment_session: Optional[LivePaymentSession] - _session: aiohttp.ClientSession = field(repr=False, compare=False) - _response: aiohttp.ClientResponse = field(repr=False, compare=False) + # Session backing this single-shot request, from the payment challenge's + # manifest_id. Empty offchain: a stream body cannot supply the JSON-path + # session_id fallback. + session_id: str = "" + # Orchestrator debit cadence in seconds, from the payment challenge's + # payment_interval_ms. None when the orchestrator does not report it. + server_payment_interval: Optional[float] = None + # True once the orchestrator reported the backing session gone (payment + # loop hit a 404). The stream itself ends when the orchestrator cancels + # the proxied request. + released: bool = False + _session: aiohttp.ClientSession = field(repr=False, compare=False, kw_only=True) + _response: aiohttp.ClientResponse = field(repr=False, compare=False, kw_only=True) + _payment_task: Optional[asyncio.Task[None]] = field( + default=None, repr=False, compare=False, kw_only=True + ) @property def content_type(self) -> str: @@ -160,7 +193,18 @@ async def aiter_lines(self) -> AsyncIterator[str]: async for line in self._response.content: yield line.decode(errors="replace").rstrip("\n") + def _mark_released(self) -> None: + self.released = True + async def aclose(self) -> None: + # Stop funding before tearing down the transport so no payment is + # minted for a stream we are abandoning. + task = self._payment_task + self._payment_task = None + if task is not None: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task self._response.release() await self._session.close() @@ -296,10 +340,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) @@ -796,16 +840,60 @@ 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( + status=resp.status, + headers=resp.headers, + runner_url=runner_url, + runner=runner, + payment_session=None if payment_type == "fixed" else payment_session, + session_id=session_id, + server_payment_interval=( + challenge.payment_interval_s if challenge is not None else None + ), + _session=session, + _response=resp, ) - - data = await request_json( - runner_url, - method=method, - payload=request_payload, - **request_kwargs, - ) + # Metered pricing: the stream outlives this call, so it owns + # the payment loop; aclose() stops both. + if payment_session is not None and payment_type in _METERED_PAYMENT_TYPES: + call_stream._payment_task = asyncio.create_task( + _run_call_payments( + payment_session, + interval_s=_payment_cadence_s( + challenge.payment_interval_s if challenge is not None else None + ), + payment_url=_single_shot_payment_url(runner_url, session_id), + on_released=call_stream._mark_released, + ) + ) + return call_stream + + # Metered pricing: the orchestrator debits the prepaid balance + # while the request runs, so keep funding it for as long as we + # are waiting on the response. + pay_task: Optional[asyncio.Task[None]] = None + if payment_session is not None and payment_type in _METERED_PAYMENT_TYPES: + pay_task = asyncio.create_task( + _run_call_payments( + payment_session, + interval_s=_payment_cadence_s( + challenge.payment_interval_s if challenge is not None else None + ), + payment_url=_single_shot_payment_url(runner_url, session_id), + ) + ) + try: + data = await request_json( + runner_url, + method=method, + payload=request_payload, + **request_kwargs, + ) + finally: + if pay_task is not None: + pay_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await pay_task if not isinstance(data, dict): raise LivepeerGatewayError( f"Live runner call expected JSON object, got {type(data).__name__}" @@ -819,6 +907,9 @@ async def call_runner( or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "") ), payment_session=None if payment_type == "fixed" else payment_session, + server_payment_interval=( + challenge.payment_interval_s if challenge is not None else None + ), ) except LivepeerHTTPError as e: if e.status_code != 402: @@ -836,6 +927,9 @@ class _RunnerPaymentChallenge: payment_params: str orchestrator_url: str manifest_id: str + # Orchestrator debit cadence in seconds (payment_interval_ms); None when + # the orchestrator does not report it. + payment_interval_s: Optional[float] = None def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentChallenge: @@ -856,13 +950,91 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC if not isinstance(manifest_id, str) or not manifest_id: raise LivepeerGatewayError("Live runner payment challenge missing manifest_id") + interval_ms = data.get("payment_interval_ms") + payment_interval_s: Optional[float] = None + if isinstance(interval_ms, (int, float)) and not isinstance(interval_ms, bool) and interval_ms > 0: + payment_interval_s = float(interval_ms) / 1000.0 + return _RunnerPaymentChallenge( payment_params=payment_params, orchestrator_url=orchestrator_url, manifest_id=manifest_id, + payment_interval_s=payment_interval_s, ) +def _single_shot_payment_url(runner_url: str, session_id: str) -> str: + """Derive the session-scoped payment endpoint for a single-shot call. + + Single-shot runner URLs are orchestrator proxy routes of the form + ``…/apps/{runner_id}/app[/…]``; the matching payment endpoint is + ``…/apps/{runner_id}/session/{session_id}/payment``, which 404s once the + session is released and 409s for fixed pricing. Returns "" when the URL + does not match that shape, in which case payments fall back to the + orchestrator's generic ``/payment`` endpoint. + """ + if not session_id: + return "" + try: + parsed = urlparse(_normalize_http_base(runner_url)) + except LivepeerGatewayError: + return "" + match = re.match(r"^(?P.*/apps/[^/]+)/app(?:/.*)?$", parsed.path) + if match is None: + return "" + path = f"{match.group('base')}/session/{quote(session_id, safe='')}/payment" + return urlunparse((parsed.scheme, parsed.netloc, path, "", "", "")) + + +def _payment_cadence_s(server_interval_s: Optional[float]) -> float: + if server_interval_s is not None and server_interval_s > 0: + return server_interval_s * _PAYMENT_CADENCE_FRACTION + return _DEFAULT_PAYMENT_INTERVAL_S + + +async def _run_call_payments( + payment_session: LivePaymentSession, + *, + interval_s: float, + payment_url: str = "", + on_released: Optional[Callable[[], None]] = None, +) -> None: + """Fund an in-flight metered single-shot call until cancelled. + + The 402 challenge payment already carried the signer's preroll, so the + first follow-up payment waits one interval. Runs until cancelled by the + caller or until the orchestrator reports a terminal state: 404 (session + released), 409 (fixed-price, nothing to fund), or 403 (session/payment + mismatch). Transient errors are retried on the next tick — the + orchestrator cancels the call itself if funding actually stops. + """ + while True: + await asyncio.sleep(interval_s) + try: + await payment_session.send_payment(payment_url=payment_url or None) + except asyncio.CancelledError: + raise + except SkipPaymentCycle: + _LOG.debug("Live runner call payment: signer skipped this cycle") + except LivepeerHTTPError as e: + if e.status_code == 404: + _LOG.info("Live runner call payment: session released; stopping payments") + if on_released is not None: + on_released() + return + if e.status_code == 409: + _LOG.debug("Live runner call payment: fixed-price session; stopping payments") + return + if e.status_code == 403: + _LOG.error( + "Live runner call payment: session/payment mismatch; stopping payments: %s", e + ) + return + _LOG.warning("Live runner call payment failed; retrying next cycle: %s", e) + except Exception as e: + _LOG.warning("Live runner call payment failed; retrying next cycle: %s", e) + + async def _get_runner_payment( challenge: _RunnerPaymentChallenge, *, @@ -977,10 +1149,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 +1318,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..72aa598 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio import base64 import json import logging @@ -12,8 +11,6 @@ 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 @@ -240,44 +237,41 @@ 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 payment endpoint, such as the + session-scoped ``…/session/{session_id}/payment`` 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. + + 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" headers = { "Livepeer-Payment": payment.payment, - "Livepeer-Segment": payment.seg_creds, + "Livepeer-Segment": payment.seg_creds or "", } - 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 _payment_request(self) -> GetPaymentResponse: from .http import _http_origin, post_json