From 38b8516efd808b374b2217c6b40f356189b4ff6c Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 12:40:47 +0200 Subject: [PATCH 01/18] feat(payments): give LivePaymentSession an interval payment loop Metered live-runner pricing debits a prepaid balance on the orchestrator's -livePaymentInterval and drops the session on the first tick it cannot cover, so every metered caller needs the same loop. Put it on the payment session itself, where both the single-shot and the reserved-session paths already hold one: - post_empty moves into http.py (Josh's suggestion on #31) so the payment POST disables cert verification like the rest of the SDK and raises typed status errors; send_payment routes through it and takes a payment_url for the session-scoped endpoint, which 404s once the session is gone instead of blind-crediting a balance - run_payments pays on an interval, returns True when the orchestrator reports the session released, and stops on the other terminal rejections (409 fixed price, 403 mismatch) rather than minting tickets no one will honor Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/http.py | 15 ++++ src/livepeer_gateway/remote_signer.py | 107 ++++++++++++++++++-------- 2 files changed, 90 insertions(+), 32 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 01c4b5d..4f2a827 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -403,6 +403,21 @@ 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.""" + await _request_body( + url, + method="POST", + headers=headers, + timeout=timeout, + ) + + def _parse_http_url(url: str, *, context: str = "URL") -> ParseResult: """ Normalize a URL for HTTP(S) endpoints. diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 0f43c6c..eff4ab7 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import base64 import json import logging @@ -11,13 +12,22 @@ 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__) +# Client payment cadence. The orchestrator debits metered sessions every +# -livePaymentInterval (5s by default) and drops the session on the first tick +# it cannot cover, so pay comfortably ahead of it. +PAYMENT_INTERVAL_S = 3.0 + @dataclass(frozen=True) class GetPaymentResponse: payment: str @@ -239,44 +249,77 @@ async def get_payment(self) -> GetPaymentResponse: await self._refresh_payment_params(orchestrator_url) attempts += 1 - async def send_payment(self, orchestrator_url: str | None = 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" 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, 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, + interval_s: float = PAYMENT_INTERVAL_S, + ) -> bool: + """Keep a metered session funded until cancelled or the session ends. + + Returns True when the orchestrator reports the session gone, so the + owner can surface it as released; returns False for the other terminal + rejections. Cancel the task to stop funding. + + The caller pays upfront before starting this loop, so the first + follow-up waits one interval. A payment covers the time since the + previous one, so transient failures are retried rather than fatal: + the next payment settles the arrears. + """ + while True: + await asyncio.sleep(interval_s) + try: + await self.send_payment(payment_url=payment_url) + except asyncio.CancelledError: + raise + except SkipPaymentCycle as e: + _LOG.debug("Payment loop skipped cycle: %s", e) + except LivepeerHTTPError as e: + # 404 session released, 409 fixed price, 403 session/payment + # mismatch: all terminal, and paying on would mint tickets the + # orchestrator will never honor. + 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 From dcad2df31d007eeb786776ebe0be37d399a5e0d5 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 12:42:50 +0200 Subject: [PATCH 02/18] feat(live-runner): fund metered single-shot calls while they run A metered single-shot call is billed for as long as it runs, but only the upfront challenge payment was ever sent, so anything outliving the signer's preroll (~10s for seconds pricing) was cancelled mid-flight by the orchestrator. Hold the funding for exactly as long as the work: - a plain call funds itself while the request is in flight and stops the moment the response lands - a streamed call hands the funding to LiveRunnerCallStream, since the stream outlives the call; aclose() stops it, and a released session surfaces as stream.released - payments go to the session-scoped endpoint, built from the challenge and the discovered runner id, so a dead session is reported instead of silently credited Fixed pricing is settled upfront and never starts a loop. Offchain calls have no payment session, so they never start one either. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/live_runner.py | 151 +++++++++++++++++++++------- 1 file changed, 115 insertions(+), 36 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 2d06836..532d70e 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 @@ -26,7 +27,7 @@ from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import _request_body, open_stream, post_json, request_json +from .http import _request_body, open_stream, post_empty, post_json, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -46,6 +47,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*$") @@ -130,6 +134,9 @@ class LiveRunnerCallResult: # Non-JSON responses (an image, say) arrive unparsed in `content`; `data` stays empty. content: Optional[bytes] = field(default=None, repr=False) content_type: str = "" + # Session-scoped payment endpoint for this call, when one could be built. + # Empty means payments fall back to the orchestrator's generic /payment. + payment_url: str = "" @dataclass @@ -146,8 +153,14 @@ class LiveRunnerCallStream: runner_url: str runner: LiveRunnerInstance | None payment_session: LivePaymentSession | None - _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. 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: @@ -162,6 +175,9 @@ async def aiter_lines(self) -> AsyncIterator[str]: yield line.decode(errors="replace").rstrip("\n") async def aclose(self) -> None: + # Stop funding first: no point minting a payment for a stream we are + # about to drop. + self._payment_task = await _stop_funding(self._payment_task) self._response.release() await self._session.close() @@ -297,10 +313,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) @@ -792,6 +808,19 @@ async def call_runner( request_headers["Livepeer-Segment"] = payment.seg_creds or "" session_id = challenge.manifest_id + # Metered pricing keeps billing for as long as the work runs, so + # whoever holds the session open has to keep paying for it. + needs_funding = payment_session is not None and payment_type in _METERED_PAYMENT_TYPES + payment_url = ( + _session_payment_url( + challenge.orchestrator_url if challenge is not None else "", + runner.runner_id if runner is not None else "", + session_id, + ) + if needs_funding + else "" + ) + try: request_kwargs: dict[str, Any] = {"timeout": timeout} if request_headers: @@ -806,16 +835,41 @@ 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=session, + _response=resp, ) - - body, content_type = await _request_body( - runner_url, - method=method, - payload=request_payload, - **request_kwargs, + # The stream outlives this call, so it owns the funding; + # aclose() stops both. + if needs_funding: + call_stream._payment_task = _start_funding( + cast(LivePaymentSession, payment_session), + payment_url, + lambda: setattr(call_stream, "released", True), + ) + return call_stream + + # A metered call is billed while we wait on it, so fund it for + # exactly as long as the request is in flight. + pay_task = ( + _start_funding(cast(LivePaymentSession, payment_session), payment_url) + if needs_funding + else None ) + try: + body, content_type = await _request_body( + runner_url, + method=method, + payload=request_payload, + **request_kwargs, + ) + finally: + await _stop_funding(pay_task) # Non-JSON bodies (an image, ndjson) are handed back unparsed in `content`. is_json = _is_json_content_type(content_type) data: dict[str, Any] = {} @@ -842,6 +896,7 @@ 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, + payment_url=payment_url, ) except LivepeerHTTPError as e: if e.status_code != 402: @@ -886,6 +941,49 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC ) +def _session_payment_url(orchestrator_url: str, runner_id: str, session_id: str) -> str: + """Build the session-scoped payment endpoint for a live runner session. + + Unlike the orchestrator's generic ``/payment``, which credits the payer + balance and returns 200 whether or not the session exists, this endpoint + 404s once the session is released, which is the only signal a payment loop + gets that it is funding nothing. Returns "" when a part is missing, in + which case payments fall back to the generic endpoint. + """ + 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='')}/payment", + ) + except LivepeerGatewayError: + return "" + + +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, *, @@ -1000,10 +1098,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, ) @@ -1174,25 +1272,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 (TimeoutError, aiohttp.ClientError) as e: - raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e - - def _detect_gpu_pynvml() -> LiveRunnerGPU | None: try: import pynvml # type: ignore[import-not-found] From 7cfcef772d7722eed77c3dcdf4397853d5ae6970 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 12:44:18 +0200 Subject: [PATCH 03/18] feat(live-runner): keep reserved sessions funded for as long as they are held reserve_session paid the reservation challenge and then dropped the payment session on the floor, so a metered session was released by the orchestrator seconds later no matter what the client did with it. The session now owns its funding from the moment it is reserved: - payments run against the session-scoped endpoint until the session is closed; a 404 means the orchestrator let it go, surfaced as session.released - stop_payments() drops funding without releasing the session, for draining or handing funding elsewhere; aclose() and the async context manager do the same on the way out - drop _live_runner_session_from_json, which had no callers and built sessions that could never fund themselves Fixed-price and offchain reservations carry no payment session, so they are unaffected. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/live_runner.py | 65 ++++++++++++++++++++--------- src/livepeer_gateway/selection.py | 8 +++- 2 files changed, 52 insertions(+), 21 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 532d70e..6e9b51d 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -106,12 +106,56 @@ class LiveRunnerInstance: price_info: LiveRunnerPriceInfo | None = 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: LiveRunnerInstance | None = None + # True once the orchestrator reported this session gone, either because it + # was stopped elsewhere or because it ran out of funds. + released: bool = False + _payment_task: Optional[asyncio.Task[None]] = field( + default=None, repr=False, compare=False + ) + + 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 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) @@ -1055,25 +1099,6 @@ def _live_runner_price_info_from_json(value: object) -> LiveRunnerPriceInfo | No ) -def _live_runner_session_from_json( - data: dict[str, Any], - *, - runner_url: str, - runner: LiveRunnerInstance | None, -) -> 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, *, diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index 968dedf..b385b79 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -300,12 +300,18 @@ 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, ) + # A metered session is billed for as long as it is held, so it funds + # itself from here until it is closed. Fixed-price and offchain + # reservations have no payment session and need nothing further. + if result.payment_session is not None: + session._start_payments(result.payment_session, result.payment_url) + return session def _runner_candidates_from_discovery(entries: Sequence[dict[str, Any]]) -> list[LiveRunnerInstance]: From 6cb47718bb391cc68546e0e3ab9f41a58baa5bd6 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 13:14:17 +0200 Subject: [PATCH 04/18] refactor(payments): resolve the payment cadence per call Binding PAYMENT_INTERVAL_S as a default argument freezes it at import, so overriding the module constant has no effect. Resolve it inside run_payments instead, which keeps the cadence adjustable in one place. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/remote_signer.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index eff4ab7..2f99d76 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -289,7 +289,7 @@ async def run_payments( self, *, payment_url: Optional[str] = None, - interval_s: float = PAYMENT_INTERVAL_S, + interval_s: Optional[float] = None, ) -> bool: """Keep a metered session funded until cancelled or the session ends. @@ -302,6 +302,9 @@ async def run_payments( previous one, so transient failures are retried rather than fatal: the next payment settles the arrears. """ + # Resolved per call rather than as a default argument so the cadence + # stays overridable at the module level. + interval_s = PAYMENT_INTERVAL_S if interval_s is None else interval_s while True: await asyncio.sleep(interval_s) try: From 97af81c4550a472b0064754d027c5378b3ab0952 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 13:25:30 +0200 Subject: [PATCH 05/18] docs(payments): note where the payment cadence should come from livepeer/go-livepeer#4001 adds payment_interval_ms to payment challenges. Until it lands the client guesses a cadence against the orchestrator's default debit interval, so record what replaces the guess. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/remote_signer.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 2f99d76..69e39c8 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -26,6 +26,11 @@ # Client payment cadence. The orchestrator debits metered sessions every # -livePaymentInterval (5s by default) and drops the session on the first tick # it cannot cover, so pay comfortably ahead of it. +# +# TODO: drive this from the orchestrator instead of guessing. Once +# livepeer/go-livepeer#4001 lands, payment challenges carry +# payment_interval_ms; read it there and pass it as run_payments(interval_s=), +# keeping this as the fallback for orchestrators that do not report one. PAYMENT_INTERVAL_S = 3.0 @dataclass(frozen=True) From aa1d8fdc58e07d6f3a5cd26a2a04b77dbc1e243a Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 13:34:13 +0200 Subject: [PATCH 06/18] feat(live-runner): take the session payment endpoint from control_url The reservation response reports the session's control URL, which the orchestrator builds as ServiceURI/apps/{runner}/session/{session}; its /payment path is the endpoint the funding loop wants. Use it directly for reserved sessions rather than the endpoint call_runner derives, so the client follows the orchestrator's own routing instead of reconstructing it. Single-shot calls keep deriving it, since a payment challenge carries no control URL. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/live_runner.py | 10 ++++++++++ src/livepeer_gateway/selection.py | 9 ++++++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 6e9b51d..f497fd2 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -120,6 +120,9 @@ class LiveRunnerSession: app_url: str runner_url: str runner: LiveRunnerInstance | None = None + # Base URL for this session's control endpoints, as reported by the + # orchestrator when the session was reserved. + control_url: str = "" # True once the orchestrator reported this session gone, either because it # was stopped elsewhere or because it ran out of funds. released: bool = False @@ -127,6 +130,13 @@ class LiveRunnerSession: default=None, repr=False, compare=False ) + @property + def payment_url(self) -> str: + """This session's payment endpoint, or "" if it reported no control URL.""" + if not self.control_url: + return "" + return _join_endpoint(self.control_url, "payment") + def _start_payments( self, payment_session: LivePaymentSession, diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index b385b79..29296c0 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -300,17 +300,24 @@ 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") + control_url = result.data.get("control_url") session = LiveRunnerSession( session_id=session_id.strip(), app_url=app_url.strip(), runner_url=result.runner_url, runner=result.runner, + control_url=control_url.strip() if isinstance(control_url, str) else "", ) # A metered session is billed for as long as it is held, so it funds # itself from here until it is closed. Fixed-price and offchain # reservations have no payment session and need nothing further. if result.payment_session is not None: - session._start_payments(result.payment_session, result.payment_url) + # The reservation response carries the session's control URL, so + # prefer it over the endpoint call_runner had to derive. + session._start_payments( + result.payment_session, + session.payment_url or result.payment_url, + ) return session From 478282f331e49c15e4cc7b3513573efacb8d2911 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 13:40:30 +0200 Subject: [PATCH 07/18] refactor(payments): drop the redundant CancelledError re-raise CancelledError has inherited from BaseException since Python 3.8, and this package requires 3.12, so the broadest handler in the loop cannot swallow it. Catching it only to re-raise added nothing. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/remote_signer.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 69e39c8..ad4112a 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -314,8 +314,6 @@ async def run_payments( await asyncio.sleep(interval_s) try: await self.send_payment(payment_url=payment_url) - except asyncio.CancelledError: - raise except SkipPaymentCycle as e: _LOG.debug("Payment loop skipped cycle: %s", e) except LivepeerHTTPError as e: From 9505698ccae85badd2c632892236a0fa97067e83 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 14:04:07 +0200 Subject: [PATCH 08/18] docs(payments): explain why run_payments exists and what it returns Say what drives the loop (the orchestrator debits on its own interval and drops the session on the first tick it cannot cover), what payment_url buys over the generic endpoint, and what the two return values mean. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/remote_signer.py | 24 ++++++++++++++++-------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index ad4112a..2fc6c23 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -298,14 +298,22 @@ async def run_payments( ) -> bool: """Keep a metered session funded until cancelled or the session ends. - Returns True when the orchestrator reports the session gone, so the - owner can surface it as released; returns False for the other terminal - rejections. Cancel the task to stop funding. - - The caller pays upfront before starting this loop, so the first - follow-up waits one interval. A payment covers the time since the - previous one, so transient failures are retried rather than fatal: - the next payment settles the arrears. + The orchestrator debits the session's prepaid balance on its own + interval and drops the session on the first tick it cannot cover, so + whoever holds a session open runs this for as long as they hold it. + Cancel the task to stop. The caller pays upfront, so the first payment + here waits one ``interval_s`` (default ``PAYMENT_INTERVAL_S``). + + ``payment_url`` should be the session-scoped endpoint, which 404s once + the session is gone; without it payments fall back to the + orchestrator's generic ``/payment``, which credits the balance whether + or not the session still exists. + + Returns True when the session is reported gone, so the owner can + surface it as released, and False on the other terminal rejections + (fixed price, or credentials naming a different session). Everything + else is retried: a payment covers the time since the last one, so the + next success settles the arrears. """ # Resolved per call rather than as a default argument so the cadence # stays overridable at the module level. From 1fca24e3a15cd615487f89a863fae915366025ac Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 14:11:59 +0200 Subject: [PATCH 09/18] docs(payments): say why the loop stops on any 4xx The old comment listed only the three known terminal codes, which read as if they were the only ones that stop the loop. Lead with the rule instead: a 4xx will not change on a retry, and 408 and 429 are the two that ask to be retried anyway. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/remote_signer.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 2fc6c23..b16737b 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -325,9 +325,9 @@ async def run_payments( except SkipPaymentCycle as e: _LOG.debug("Payment loop skipped cycle: %s", e) except LivepeerHTTPError as e: - # 404 session released, 409 fixed price, 403 session/payment - # mismatch: all terminal, and paying on would mint tickets the - # orchestrator will never honor. + # 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 From 8509b629e007d698eda30983219c2c267c48be92 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 14:13:22 +0200 Subject: [PATCH 10/18] docs(payments): trim the cadence comment to one line Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/remote_signer.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index b16737b..e582593 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -315,8 +315,7 @@ async def run_payments( else is retried: a payment covers the time since the last one, so the next success settles the arrears. """ - # Resolved per call rather than as a default argument so the cadence - # stays overridable at the module level. + # Not a default argument: those bind at import and freeze the constant. interval_s = PAYMENT_INTERVAL_S if interval_s is None else interval_s while True: await asyncio.sleep(interval_s) From e100211bf10b5456d39160a28bc164e772b8141b Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 15:52:35 +0200 Subject: [PATCH 11/18] refactor(payments): tighten the payment path after review Behaviour: - reserve_session no longer falls back to a payment endpoint derived from the challenge. An orchestrator reporting no control URL gets paid at its generic /payment, where before it would have been sent to a route it may not serve and answered with a 404 the loop reads as a released session. - send_payment rejects a payment the signer returned without segCreds instead of sending an empty segment header, which fails the orchestrator's sig check and comes back 403 - again read as a dead session rather than a bad signer. Shape: - split the endpoint derivation into _session_control_url and _payment_endpoint, so one place knows a payment endpoint sits under a session's control URL and the reported and derived URLs agree by construction - drop LiveRunnerCallResult.payment_url, which no longer has a consumer, and LiveRunnerSession derives its own from control_url - give LiveRunnerCallStream the same _start_payments as the session - drop run_payments' interval_s parameter; the cadence is the module constant until the orchestrator reports one - keep the stream's new fields after the existing ones so its constructor call stays positional - reuse selection.py's own _string_value for the control URL Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/live_runner.py | 123 +++++++++++++------------- src/livepeer_gateway/remote_signer.py | 37 +++----- src/livepeer_gateway/selection.py | 15 +--- 3 files changed, 78 insertions(+), 97 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index f497fd2..26513da 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -123,8 +123,7 @@ class LiveRunnerSession: # Base URL for this session's control endpoints, as reported by the # orchestrator when the session was reserved. control_url: str = "" - # True once the orchestrator reported this session gone, either because it - # was stopped elsewhere or because it ran out of funds. + # True once the orchestrator reported this session gone. released: bool = False _payment_task: Optional[asyncio.Task[None]] = field( default=None, repr=False, compare=False @@ -132,21 +131,15 @@ class LiveRunnerSession: @property def payment_url(self) -> str: - """This session's payment endpoint, or "" if it reported no control URL.""" - if not self.control_url: - return "" - return _join_endpoint(self.control_url, "payment") + """This session's payment endpoint, or "" when none was reported.""" + return _payment_endpoint(self.control_url) - def _start_payments( - self, - payment_session: LivePaymentSession, - payment_url: str = "", - ) -> None: + def _start_payments(self, payment_session: LivePaymentSession) -> None: if self._payment_task is not None: return self._payment_task = _start_funding( payment_session, - payment_url, + self.payment_url, lambda: setattr(self, "released", True), ) @@ -188,9 +181,6 @@ class LiveRunnerCallResult: # Non-JSON responses (an image, say) arrive unparsed in `content`; `data` stays empty. content: Optional[bytes] = field(default=None, repr=False) content_type: str = "" - # Session-scoped payment endpoint for this call, when one could be built. - # Empty means payments fall back to the orchestrator's generic /payment. - payment_url: str = "" @dataclass @@ -207,13 +197,12 @@ class LiveRunnerCallStream: runner_url: str runner: LiveRunnerInstance | None payment_session: LivePaymentSession | None - # True once the orchestrator reported the backing session gone. The stream - # itself ends when the orchestrator cancels the proxied request. + _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 - _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 + default=None, repr=False, compare=False ) @property @@ -228,9 +217,21 @@ 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: no point minting a payment for a stream we are - # about to drop. + # 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() @@ -833,6 +834,8 @@ async def call_runner( payment_session: LivePaymentSession | None = None payment_type = "" session_id = "" + needs_ongoing_funding = False + payment_url = "" # No preferred format: the app, or the upstream it fronts, picks. Only # control-plane calls ask for JSON. request_headers: dict[str, str] = {"Accept": "*/*"} @@ -862,18 +865,16 @@ async def call_runner( request_headers["Livepeer-Segment"] = payment.seg_creds or "" session_id = challenge.manifest_id - # Metered pricing keeps billing for as long as the work runs, so - # whoever holds the session open has to keep paying for it. - needs_funding = payment_session is not None and payment_type in _METERED_PAYMENT_TYPES - payment_url = ( - _session_payment_url( - challenge.orchestrator_url if challenge is not None else "", - runner.runner_id if runner is not None else "", - session_id, - ) - if needs_funding - else "" - ) + # 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} @@ -890,29 +891,25 @@ async def call_runner( headers=request_headers or None, ) 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=session, - _response=resp, + 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; - # aclose() stops both. - if needs_funding: - call_stream._payment_task = _start_funding( - cast(LivePaymentSession, payment_session), - payment_url, - lambda: setattr(call_stream, "released", True), + # 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 - # A metered call is billed while we wait on it, so fund it for - # exactly as long as the request is in flight. + # The request ends with this call, so the funding ends with it. pay_task = ( _start_funding(cast(LivePaymentSession, payment_session), payment_url) - if needs_funding + if needs_ongoing_funding else None ) try: @@ -950,7 +947,6 @@ 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, - payment_url=payment_url, ) except LivepeerHTTPError as e: if e.status_code != 402: @@ -995,26 +991,33 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC ) -def _session_payment_url(orchestrator_url: str, runner_id: str, session_id: str) -> str: - """Build the session-scoped payment endpoint for a live runner session. +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. - Unlike the orchestrator's generic ``/payment``, which credits the payer - balance and returns 200 whether or not the session exists, this endpoint - 404s once the session is released, which is the only signal a payment loop - gets that it is funding nothing. Returns "" when a part is missing, in - which case payments fall back to the generic endpoint. + 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='')}/payment", + 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, diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index e582593..a5b37ce 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -284,41 +284,26 @@ async def send_payment( url = f"{_http_origin(target)}/payment" payment = await self.get_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 or "", + "Livepeer-Segment": payment.seg_creds, } await post_empty(url, headers=headers, timeout=5.0) - async def run_payments( - self, - *, - payment_url: Optional[str] = None, - interval_s: Optional[float] = None, - ) -> bool: + async def run_payments(self, *, payment_url: Optional[str] = None) -> bool: """Keep a metered session funded until cancelled or the session ends. - The orchestrator debits the session's prepaid balance on its own - interval and drops the session on the first tick it cannot cover, so - whoever holds a session open runs this for as long as they hold it. - Cancel the task to stop. The caller pays upfront, so the first payment - here waits one ``interval_s`` (default ``PAYMENT_INTERVAL_S``). - - ``payment_url`` should be the session-scoped endpoint, which 404s once - the session is gone; without it payments fall back to the - orchestrator's generic ``/payment``, which credits the balance whether - or not the session still exists. - - Returns True when the session is reported gone, so the owner can - surface it as released, and False on the other terminal rejections - (fixed price, or credentials naming a different session). Everything - else is retried: a payment covers the time since the last one, so the - next success settles the arrears. + 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. """ - # Not a default argument: those bind at import and freeze the constant. - interval_s = PAYMENT_INTERVAL_S if interval_s is None else interval_s while True: - await asyncio.sleep(interval_s) + await asyncio.sleep(PAYMENT_INTERVAL_S) try: await self.send_payment(payment_url=payment_url) except SkipPaymentCycle as e: diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index 29296c0..cc992eb 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -300,24 +300,17 @@ 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") - control_url = result.data.get("control_url") session = LiveRunnerSession( session_id=session_id.strip(), app_url=app_url.strip(), runner_url=result.runner_url, runner=result.runner, - control_url=control_url.strip() if isinstance(control_url, str) else "", + control_url=_string_value(result.data.get("control_url")), ) - # A metered session is billed for as long as it is held, so it funds - # itself from here until it is closed. Fixed-price and offchain - # reservations have no payment session and need nothing further. + + # No payment session means fixed price or offchain: nothing to fund. if result.payment_session is not None: - # The reservation response carries the session's control URL, so - # prefer it over the endpoint call_runner had to derive. - session._start_payments( - result.payment_session, - session.payment_url or result.payment_url, - ) + session._start_payments(result.payment_session) return session From 15faf507ad78b9f040d7c4788f702e677c7f4b92 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 19:32:50 +0200 Subject: [PATCH 12/18] docs(payments): say what bounds the payment cadence The signer's opening payment bounds it, not the orchestrator's debit interval: both sides bill by elapsed time at the same rate. Drops the pointer to livepeer/go-livepeer#4001, closed for the same reason. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/remote_signer.py | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index a5b37ce..1670df9 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -23,14 +23,7 @@ ) _LOG = logging.getLogger(__name__) -# Client payment cadence. The orchestrator debits metered sessions every -# -livePaymentInterval (5s by default) and drops the session on the first tick -# it cannot cover, so pay comfortably ahead of it. -# -# TODO: drive this from the orchestrator instead of guessing. Once -# livepeer/go-livepeer#4001 lands, payment challenges carry -# payment_interval_ms; read it there and pass it as run_payments(interval_s=), -# keeping this as the fallback for orchestrators that do not report one. +# Must stay under the signer's opening payment: 10s per-second, 60s pixel. PAYMENT_INTERVAL_S = 3.0 @dataclass(frozen=True) From 9d13a2f738085e4f914047102d610618a13eadc1 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 14:41:10 -0700 Subject: [PATCH 13/18] fix(live-runner): align reservation and payment lifecycles Signed-off-by: Josh Allmann --- src/livepeer_gateway/http.py | 8 +- src/livepeer_gateway/live_runner.py | 105 +++++--- src/livepeer_gateway/remote_signer.py | 13 +- src/livepeer_gateway/selection.py | 63 ++++- tests/test_live_payment_session.py | 185 +++----------- tests/test_live_runner.py | 6 +- tests/test_live_runner_payments.py | 337 ++++++++++++++++++++++++++ tests/test_selection.py | 12 +- 8 files changed, 525 insertions(+), 204 deletions(-) create mode 100644 tests/test_live_runner_payments.py diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 4f2a827..b26a165 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -305,9 +305,9 @@ async def _request_body( async def request_json( url: str, *, - method: Optional[str] = None, - payload: Optional[dict[str, Any]] = None, - headers: Optional[dict[str, str]] = None, + method: str | None = None, + payload: dict[str, Any] | None = None, + headers: dict[str, str] | None = None, timeout: float = 5.0, ) -> Any: """ @@ -406,7 +406,7 @@ async def get_json( async def post_empty( url: str, *, - headers: Optional[dict[str, str]] = None, + headers: dict[str, str] | None = None, timeout: float = 5.0, ) -> None: """POST an empty body to ``url`` and discard the response.""" diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 26513da..0895ecd 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -20,7 +20,7 @@ overload, ) from collections.abc import AsyncIterator, Awaitable, Callable, Mapping -from urllib.parse import quote, urlparse, urlunparse +from urllib.parse import quote, unquote, urlparse, urlunparse import aiohttp from aiohttp.helpers import parse_mimetype @@ -125,13 +125,13 @@ class LiveRunnerSession: control_url: str = "" # True once the orchestrator reported this session gone. released: bool = False - _payment_task: Optional[asyncio.Task[None]] = field( + _payment_task: asyncio.Task[None] | 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 this session's required session-scoped payment endpoint.""" return _payment_endpoint(self.control_url) def _start_payments(self, payment_session: LivePaymentSession) -> None: @@ -149,10 +149,15 @@ async def stop_payments(self) -> None: 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) + await _stop_funding(self._payment_task) + self._payment_task = None async def aclose(self) -> None: + # Stop funding first so a slow or failed remote stop cannot mint another + # payment while this session is being closed. await self.stop_payments() + if not self.released: + await stop_runner_session(self) async def __aenter__(self) -> LiveRunnerSession: return self @@ -179,7 +184,7 @@ class LiveRunnerCallResult: compare=False, ) # Non-JSON responses (an image, say) arrive unparsed in `content`; `data` stays empty. - content: Optional[bytes] = field(default=None, repr=False) + content: bytes | None = field(default=None, repr=False) content_type: str = "" @@ -201,7 +206,7 @@ class LiveRunnerCallStream: _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( + _payment_task: asyncio.Task[None] | None = field( default=None, repr=False, compare=False ) @@ -220,7 +225,7 @@ async def aiter_lines(self) -> AsyncIterator[str]: def _start_payments( self, payment_session: LivePaymentSession, - payment_url: str = "", + payment_url: str, ) -> None: if self._payment_task is not None: return @@ -232,7 +237,8 @@ def _start_payments( 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) + await _stop_funding(self._payment_task) + self._payment_task = None self._response.release() await self._session.close() @@ -871,7 +877,11 @@ async def call_runner( payment_url = _payment_endpoint( _session_control_url( challenge.orchestrator_url, - runner.runner_id if runner is not None else "", + ( + runner.runner_id + if runner is not None + else _runner_id_from_url(runner_url) + ), session_id, ) ) @@ -995,50 +1005,63 @@ def _session_control_url(orchestrator_url: str, runner_id: str, session_id: 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. + the orchestrator, the runner and the session id. """ 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='')}", + raise LivepeerGatewayError( + "Live runner session payment requires orchestrator, runner_id, and session_id" ) - except LivepeerGatewayError: + return _join_endpoint( + orchestrator_url, + f"/apps/{quote(runner_id, safe='')}/session/{quote(session_id, safe='')}", + ) + + +def _runner_id_from_url(runner_url: str) -> str: + """Extract the runner id from a canonical ``/apps/{id}/...`` URL.""" + parts = [part for part in urlparse(runner_url).path.split("/") if part] + try: + apps_index = parts.index("apps") + runner_id = parts[apps_index + 1] + except (ValueError, IndexError): return "" + return unquote(runner_id) 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. + Unlike the generic ``/payment``, it 404s once the session is gone. """ - return _join_endpoint(control_url, "payment") if control_url else "" + if not control_url.strip(): + raise LivepeerGatewayError( + "Live runner session payment requires control_url; refusing generic /payment fallback" + ) + return _join_endpoint(control_url, "payment") def _start_funding( payment_session: LivePaymentSession, payment_url: str, - on_released: Optional[Callable[[], None]] = None, + on_released: Callable[[], None] | 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) + released = await payment_session.run_payments(payment_url=payment_url) 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(): +async def _stop_funding(task: asyncio.Task[None] | None) -> None: + if task is None: + return + if not task.done(): task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await task - return None + with contextlib.suppress(asyncio.CancelledError): + await task async def _get_runner_payment( @@ -1119,13 +1142,25 @@ async def stop_runner_session( ) -> None: request_headers: dict[str, str] = {} if isinstance(session, LiveRunnerSession): - runner_url = session.runner_url.strip() - session_id = session.session_id.strip() - if not runner_url: - raise LivepeerGatewayError("Live runner session stop requires runner_url") - if not session_id: - raise LivepeerGatewayError("Live runner session stop requires session_id") - url = _join_endpoint(runner_url, f"/{quote(session_id, safe='')}/stop") + # This helper is also the public cleanup path, so callers that do not + # use aclose() still stop local funding before the remote reservation. + await session.stop_payments() + if session.released: + return + + control_url = session.control_url.strip() + if control_url: + url = _join_endpoint(control_url, "stop") + else: + # Kept only for cleaning up a malformed legacy reservation. New + # reservations require control_url before they can be selected. + runner_url = session.runner_url.strip() + session_id = session.session_id.strip() + if not runner_url: + raise LivepeerGatewayError("Live runner session stop requires runner_url") + if not session_id: + raise LivepeerGatewayError("Live runner session stop requires session_id") + url = _join_endpoint(runner_url, f"/{quote(session_id, safe='')}/stop") else: headers = getattr(session, "headers", None) get = getattr(headers, "get", None) @@ -1141,6 +1176,8 @@ async def stop_runner_session( headers=request_headers, timeout=timeout, ) + if isinstance(session, LiveRunnerSession): + session.released = True def detect_process_gpu() -> LiveRunnerGPU | None: diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 1670df9..aeb8b7e 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -287,14 +287,19 @@ async def send_payment( } await post_empty(url, headers=headers, timeout=5.0) - async def run_payments(self, *, payment_url: Optional[str] = None) -> bool: + async def run_payments(self, *, payment_url: str) -> 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. + the caller pays upfront. ``payment_url`` must be the session-scoped + endpoint; the generic ``/payment`` endpoint is intentionally unsupported + because it cannot report that the session is gone. Returns True if the + orchestrator reported it gone. """ + if not payment_url.strip(): + raise PaymentError( + "session-scoped payment_url is required; refusing generic /payment fallback" + ) while True: await asyncio.sleep(PAYMENT_INTERVAL_S) try: diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index cc992eb..c96125e 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -25,6 +25,7 @@ LiveRunnerSession, _live_runner_price_info_from_json, call_runner, + stop_runner_session, ) from .orch_info import get_orch_info @@ -293,27 +294,77 @@ async def reserve_session( gpu=gpu, timeout=timeout, ) - result = await cursor.next() + while True: + result = await cursor.next() + try: + session = _reserved_session_from_result(result) + except LivepeerGatewayError as e: + await _cleanup_rejected_reservation(result, timeout=timeout) + cursor.rejections.append( + RunnerRejection(url=result.runner_url, reason=str(e)) + ) + _LOG.debug( + "reserve_session rejected candidate %s: %s", + result.runner_url, + e, + ) + continue + + # 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 _reserved_session_from_result(result: LiveRunnerCallResult) -> LiveRunnerSession: session_id = result.data.get("session_id") app_url = result.data.get("app_url") + control_url = _string_value(result.data.get("control_url")) if not isinstance(session_id, str) or not session_id.strip(): 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") + if not control_url: + raise LivepeerGatewayError("runner session response missing control_url") 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")), + control_url=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) + # Validate the reported URL before accepting the reservation or starting a + # background task. This also guarantees that funding is session-scoped. + _ = session.payment_url return session +async def _cleanup_rejected_reservation( + result: LiveRunnerCallResult, + *, + timeout: float, +) -> None: + session_id = _string_value(result.data.get("session_id")) + if not session_id: + return + # Do not trust a malformed control_url during cleanup. The runner URL plus + # session id names the same stop endpoint and is safe for this best effort. + session = LiveRunnerSession( + session_id=session_id, + app_url=_string_value(result.data.get("app_url")), + runner_url=result.runner_url, + runner=result.runner, + ) + try: + await stop_runner_session(session, timeout=timeout) + except Exception: + _LOG.debug( + "Failed to clean up rejected runner reservation %s", + result.runner_url, + exc_info=True, + ) + + def _runner_candidates_from_discovery(entries: Sequence[dict[str, Any]]) -> list[LiveRunnerInstance]: candidates: list[LiveRunnerInstance] = [] for entry in entries: diff --git a/tests/test_live_payment_session.py b/tests/test_live_payment_session.py index 094571f..19a8d7c 100644 --- a/tests/test_live_payment_session.py +++ b/tests/test_live_payment_session.py @@ -7,6 +7,7 @@ from livepeer_gateway import lp_rpc_pb2 from livepeer_gateway.errors import ( + LivepeerHTTPError, PaymentError, SignerRefreshRequired, ) @@ -97,38 +98,7 @@ async def test_none_signer_exits_early(self) -> None: assert payment.payment == "" assert payment.seg_creds is None - async def test_send_payment_uses_default_tls_verification(self) -> None: - class _Response: - status = 204 - headers: dict[str, str] = {} - - async def __aenter__(self) -> _Response: - return self - - async def __aexit__(self, *args: object) -> None: - return None - - async def read(self) -> bytes: - return b"" - - async def text(self) -> str: - raise AssertionError( - "successful payment responses must not be decoded as text" - ) - - class _Session: - def __init__(self, **kwargs: object) -> None: - self.kwargs = kwargs - - async def __aenter__(self) -> _Session: - return self - - async def __aexit__(self, *args: object) -> None: - return None - - def post(self, *args: object, **kwargs: object) -> _Response: - return _Response() - + async def test_send_payment_reuses_empty_post_helper(self) -> None: session = LivePaymentSession( "https://signer.example.com", type="lv2v", @@ -136,6 +106,7 @@ def post(self, *args: object, **kwargs: object) -> _Response: manifest_id="manifest-1", ) + post_empty = mock.AsyncMock() with ( mock.patch.object( session, @@ -144,49 +115,17 @@ def post(self, *args: object, **kwargs: object) -> _Response: return_value=types.SimpleNamespace(payment="p", seg_creds="s") ), ), - mock.patch( - "livepeer_gateway.remote_signer.aiohttp.TCPConnector" - ) as connector_mock, - mock.patch( - "livepeer_gateway.remote_signer.aiohttp.ClientSession", - side_effect=_Session, - ) as client_session_mock, + mock.patch("livepeer_gateway.http.post_empty", post_empty), ): await session.send_payment("https://orchestrator.example.com") - connector_mock.assert_not_called() - assert "connector" not in client_session_mock.call_args.kwargs + post_empty.assert_awaited_once_with( + "https://orchestrator.example.com/payment", + headers={"Livepeer-Payment": "p", "Livepeer-Segment": "s"}, + timeout=5.0, + ) async def test_send_payment_uses_constructor_orchestrator_url(self) -> None: - posts: list[tuple[object, dict[str, object]]] = [] - - class _Response: - status = 204 - headers: dict[str, str] = {} - - async def __aenter__(self) -> _Response: - return self - - async def __aexit__(self, *args: object) -> None: - return None - - async def read(self) -> bytes: - return b"" - - class _Session: - def __init__(self, **kwargs: object) -> None: - del kwargs - - async def __aenter__(self) -> _Session: - return self - - async def __aexit__(self, *args: object) -> None: - return None - - def post(self, url: object, **kwargs: object) -> _Response: - posts.append((url, kwargs)) - return _Response() - session = LivePaymentSession( "https://signer.example.com", type="lv2v", @@ -195,6 +134,7 @@ def post(self, url: object, **kwargs: object) -> _Response: orchestrator_url="https://orchestrator.example.com/base", ) + post_empty = mock.AsyncMock() with ( mock.patch.object( session, @@ -203,52 +143,13 @@ def post(self, url: object, **kwargs: object) -> _Response: return_value=types.SimpleNamespace(payment="p", seg_creds="s") ), ), - mock.patch( - "livepeer_gateway.remote_signer.aiohttp.ClientSession", - side_effect=_Session, - ), + mock.patch("livepeer_gateway.http.post_empty", post_empty), ): await session.send_payment() - assert posts[0][0] == "https://orchestrator.example.com/payment" - assert posts[0][1]["headers"] == { - "Livepeer-Payment": "p", - "Livepeer-Segment": "s", - } - - async def test_send_payment_accepts_binary_payment_result_response(self) -> None: - class _Response: - status = 200 - headers: dict[str, str] = {} - - async def __aenter__(self) -> _Response: - return self - - async def __aexit__(self, *args: object) -> None: - return None - - async def read(self) -> bytes: - return b"\x82\x01protobuf-payment-result" - - async def text(self) -> str: - raise AssertionError( - "successful binary payment response decoded as text" - ) - - class _Session: - def __init__(self, **kwargs: object) -> None: - del kwargs - - async def __aenter__(self) -> _Session: - return self - - async def __aexit__(self, *args: object) -> None: - return None - - def post(self, *args: object, **kwargs: object) -> _Response: - del args, kwargs - return _Response() + assert post_empty.await_args.args[0] == "https://orchestrator.example.com/payment" + async def test_send_payment_prefers_session_scoped_url(self) -> None: session = LivePaymentSession( "https://signer.example.com", type="lv2v", @@ -256,6 +157,7 @@ def post(self, *args: object, **kwargs: object) -> _Response: manifest_id="manifest-1", ) + post_empty = mock.AsyncMock() with ( mock.patch.object( session, @@ -264,44 +166,19 @@ def post(self, *args: object, **kwargs: object) -> _Response: return_value=types.SimpleNamespace(payment="p", seg_creds="s") ), ), - mock.patch( - "livepeer_gateway.remote_signer.aiohttp.ClientSession", - side_effect=_Session, - ), + mock.patch("livepeer_gateway.http.post_empty", post_empty), ): - await session.send_payment("https://orchestrator.example.com") - - async def test_send_payment_error_decodes_body_for_message(self) -> None: - class _Response: - status = 400 - headers: dict[str, str] = {} - - async def __aenter__(self) -> _Response: - return self - - async def __aexit__(self, *args: object) -> None: - return None - - async def read(self) -> bytes: - raise AssertionError("error payment responses should use text decoding") - - async def text(self) -> str: - return '{"error":{"message":"payment rejected"}}' - - class _Session: - def __init__(self, **kwargs: object) -> None: - del kwargs - - async def __aenter__(self) -> _Session: - return self - - async def __aexit__(self, *args: object) -> None: - return None + await session.send_payment( + "https://orchestrator.example.com", + payment_url="https://orchestrator.example.com/apps/r/session/s/payment", + ) - def post(self, *args: object, **kwargs: object) -> _Response: - del args, kwargs - return _Response() + assert ( + post_empty.await_args.args[0] + == "https://orchestrator.example.com/apps/r/session/s/payment" + ) + async def test_send_payment_preserves_typed_http_error(self) -> None: session = LivePaymentSession( "https://signer.example.com", type="lv2v", @@ -309,6 +186,12 @@ def post(self, *args: object, **kwargs: object) -> _Response: manifest_id="manifest-1", ) + error = LivepeerHTTPError( + 400, + "https://orchestrator.example.com/payment", + body='{"error":{"message":"payment rejected"}}', + message="payment rejected", + ) with ( mock.patch.object( session, @@ -318,14 +201,14 @@ def post(self, *args: object, **kwargs: object) -> _Response: ), ), mock.patch( - "livepeer_gateway.remote_signer.aiohttp.ClientSession", - side_effect=_Session, + "livepeer_gateway.http.post_empty", + new=mock.AsyncMock(side_effect=error), ), ): - with pytest.raises(PaymentError) as raised: + with pytest.raises(LivepeerHTTPError) as raised: await session.send_payment("https://orchestrator.example.com") - assert "payment rejected" in str(raised.value) + assert raised.value is error async def test_get_payment_sends_opaque_payment_params_and_state(self) -> None: calls: list[tuple[str, dict[str, object], dict[str, str] | None]] = [] diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index e85d1f2..2adff3f 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -134,7 +134,7 @@ def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: runner_url="https://service.example.com/apps/runner-1/session", ) - with mock.patch.object(live_runner, "_post_empty", side_effect=_post_empty): + with mock.patch.object(live_runner, "post_empty", side_effect=_post_empty): await stop_runner_session(session) assert stopped == [ @@ -158,7 +158,7 @@ def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: } ) - with mock.patch.object(live_runner, "_post_empty", side_effect=_post_empty): + with mock.patch.object(live_runner, "post_empty", side_effect=_post_empty): await stop_runner_session(request, timeout=12.0) assert stopped == [ @@ -1026,7 +1026,7 @@ def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: "heartbeat_secret": "heartbeat-token", }, ): - with mock.patch.object(live_runner, "_post_empty", side_effect=_post_empty): + with mock.patch.object(live_runner, "post_empty", side_effect=_post_empty): reg = await register_runner( "http://orch.example.com", secret="secret-token", diff --git a/tests/test_live_runner_payments.py b/tests/test_live_runner_payments.py new file mode 100644 index 0000000..4522f39 --- /dev/null +++ b/tests/test_live_runner_payments.py @@ -0,0 +1,337 @@ +from __future__ import annotations + +import asyncio +from unittest import mock + +import pytest + +from livepeer_gateway import live_runner, remote_signer, selection +from livepeer_gateway.errors import ( + LivepeerGatewayError, + LivepeerHTTPError, + NoRunnerAvailableError, + PaymentError, + SkipPaymentCycle, +) +from livepeer_gateway.live_runner import LiveRunnerCallResult, LiveRunnerSession +from livepeer_gateway.remote_signer import LivePaymentSession + + +_CONTROL_URL = "https://orch.example.com/apps/runner-1/session/session-1" +_PAYMENT_URL = f"{_CONTROL_URL}/payment" + + +def _http_error(status: int) -> LivepeerHTTPError: + return LivepeerHTTPError(status, _PAYMENT_URL) + + +def _live_payment_session() -> LivePaymentSession: + return LivePaymentSession( + "https://signer.example.com", + type="live", + payment_params="opaque", + manifest_id="session-1", + ) + + +class _FundingSession: + def __init__(self, *, released: bool = False) -> None: + self.released = released + self.urls: list[str] = [] + self.started = asyncio.Event() + self.cancelled = asyncio.Event() + + async def run_payments(self, *, payment_url: str) -> bool: + self.urls.append(payment_url) + self.started.set() + try: + await asyncio.Future() + except asyncio.CancelledError: + self.cancelled.set() + raise + return self.released + + +def _session(*, control_url: str = _CONTROL_URL) -> LiveRunnerSession: + return LiveRunnerSession( + session_id="session-1", + app_url=f"{_CONTROL_URL}/app", + runner_url="https://orch.example.com/apps/runner-1/session", + control_url=control_url, + ) + + +class TestPaymentLoop: + async def test_requires_session_scoped_url(self) -> None: + with pytest.raises(PaymentError, match="session-scoped payment_url"): + await _live_payment_session().run_payments(payment_url="") + + @pytest.mark.parametrize( + "status, released", [(403, False), (404, True), (409, False)] + ) + async def test_terminal_status_stops_loop( + self, status: int, released: bool + ) -> None: + payment_session = _live_payment_session() + with ( + mock.patch.object( + payment_session, + "send_payment", + new=mock.AsyncMock(side_effect=_http_error(status)), + ) as send_payment, + mock.patch.object(remote_signer, "PAYMENT_INTERVAL_S", 0), + ): + result = await asyncio.wait_for( + payment_session.run_payments(payment_url=_PAYMENT_URL), + timeout=1.0, + ) + + assert result is released + send_payment.assert_awaited_once_with(payment_url=_PAYMENT_URL) + + @pytest.mark.parametrize( + "first_error", + [RuntimeError("network"), _http_error(408), SkipPaymentCycle("paid up")], + ) + async def test_retryable_error_reaches_next_cycle( + self, first_error: Exception + ) -> None: + payment_session = _live_payment_session() + send_payment = mock.AsyncMock(side_effect=[first_error, _http_error(404)]) + with ( + mock.patch.object(payment_session, "send_payment", new=send_payment), + mock.patch.object(remote_signer, "PAYMENT_INTERVAL_S", 0), + ): + released = await asyncio.wait_for( + payment_session.run_payments(payment_url=_PAYMENT_URL), + timeout=1.0, + ) + + assert released + assert send_payment.await_count == 2 + + +class TestSessionPaymentLifecycle: + def test_payment_url_is_derived_from_control_url(self) -> None: + assert _session().payment_url == _PAYMENT_URL + + @pytest.mark.parametrize("control_url", ["", "ftp://orch/session/session-1"]) + def test_payment_url_rejects_missing_or_invalid_control_url( + self, control_url: str + ) -> None: + with pytest.raises(LivepeerGatewayError): + _session(control_url=control_url).payment_url + + async def test_start_payments_uses_only_session_scoped_endpoint(self) -> None: + payment_session = _FundingSession() + session = _session() + + session._start_payments(payment_session) # type: ignore[arg-type] + await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) + + assert payment_session.urls == [_PAYMENT_URL] + await session.stop_payments() + + async def test_start_payments_is_idempotent(self) -> None: + payment_session = _FundingSession() + session = _session() + + session._start_payments(payment_session) # type: ignore[arg-type] + task = session._payment_task + session._start_payments(payment_session) # type: ignore[arg-type] + + assert session._payment_task is task + await session.stop_payments() + + async def test_stop_payments_cancels_and_clears_task(self) -> None: + payment_session = _FundingSession() + session = _session() + session._start_payments(payment_session) # type: ignore[arg-type] + await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) + + await session.stop_payments() + + assert session._payment_task is None + assert payment_session.cancelled.is_set() + + async def test_aclose_stops_funding_then_remote_session(self) -> None: + payment_session = _FundingSession() + session = _session() + session._start_payments(payment_session) # type: ignore[arg-type] + await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) + + stop = mock.AsyncMock() + with mock.patch.object(live_runner, "stop_runner_session", stop): + await session.aclose() + + assert payment_session.cancelled.is_set() + stop.assert_awaited_once_with(session) + + async def test_aclose_skips_remote_stop_when_already_released(self) -> None: + session = _session() + session.released = True + stop = mock.AsyncMock() + + with mock.patch.object(live_runner, "stop_runner_session", stop): + await session.aclose() + + stop.assert_not_awaited() + + async def test_async_context_manager_closes_session(self) -> None: + session = _session() + stop = mock.AsyncMock() + + with mock.patch.object(live_runner, "stop_runner_session", stop): + async with session as entered: + assert entered is session + + stop.assert_awaited_once_with(session) + + +class TestStopRunnerSession: + async def test_stops_payment_task_and_uses_control_url(self) -> None: + payment_session = _FundingSession() + session = _session() + session._start_payments(payment_session) # type: ignore[arg-type] + await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) + post_empty = mock.AsyncMock() + + with mock.patch.object(live_runner, "post_empty", post_empty): + await live_runner.stop_runner_session(session, timeout=12.0) + + assert payment_session.cancelled.is_set() + assert session._payment_task is None + assert session.released + post_empty.assert_awaited_once_with( + f"{_CONTROL_URL}/stop", + headers={}, + timeout=12.0, + ) + + async def test_remote_stop_failure_still_stops_payment_task(self) -> None: + payment_session = _FundingSession() + session = _session() + session._start_payments(payment_session) # type: ignore[arg-type] + await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) + + with ( + mock.patch.object( + live_runner, + "post_empty", + new=mock.AsyncMock(side_effect=LivepeerGatewayError("stop failed")), + ), + pytest.raises(LivepeerGatewayError, match="stop failed"), + ): + await live_runner.stop_runner_session(session) + + assert payment_session.cancelled.is_set() + assert session._payment_task is None + assert not session.released + + async def test_already_released_session_only_stops_local_funding(self) -> None: + payment_session = _FundingSession() + session = _session() + session._start_payments(payment_session) # type: ignore[arg-type] + await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) + session.released = True + post_empty = mock.AsyncMock() + + with mock.patch.object(live_runner, "post_empty", post_empty): + await live_runner.stop_runner_session(session) + + assert payment_session.cancelled.is_set() + post_empty.assert_not_awaited() + + +class _Cursor: + def __init__(self, *results: LiveRunnerCallResult) -> None: + self.results = list(results) + self.rejections = [] + + async def next(self) -> LiveRunnerCallResult: + if self.results: + return self.results.pop(0) + raise NoRunnerAvailableError( + f"All runners failed ({len(self.rejections)} tried)", + rejections=list(self.rejections), + ) + + +def _reservation( + name: str, + *, + control_url: str | None, + payment_session: object | None = None, +) -> LiveRunnerCallResult: + data = { + "session_id": f"session-{name}", + "app_url": f"https://orch.example.com/session-{name}/app", + } + if control_url is not None: + data["control_url"] = control_url + return LiveRunnerCallResult( + data, + runner_url=f"https://orch.example.com/runner-{name}/session", + payment_session=payment_session, # type: ignore[arg-type] + ) + + +class TestReservationSelection: + async def test_paid_reservation_starts_scoped_funding(self) -> None: + payment_session = _FundingSession() + cursor = _Cursor( + _reservation( + "1", + control_url=_CONTROL_URL, + payment_session=payment_session, + ) + ) + + with mock.patch.object( + selection, "runner_selector", new=mock.AsyncMock(return_value=cursor) + ): + session = await selection.reserve_session() + + await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) + assert payment_session.urls == [_PAYMENT_URL] + await session.stop_payments() + + @pytest.mark.parametrize( + "bad_control_url", + [None, "ftp://orch.example.com/session/bad"], + ) + async def test_invalid_control_url_rejects_candidate_and_tries_next( + self, bad_control_url: str | None + ) -> None: + cursor = _Cursor( + _reservation("bad", control_url=bad_control_url), + _reservation("good", control_url=_CONTROL_URL), + ) + cleanup = mock.AsyncMock() + + with ( + mock.patch.object( + selection, + "runner_selector", + new=mock.AsyncMock(return_value=cursor), + ), + mock.patch.object(selection, "stop_runner_session", cleanup), + ): + session = await selection.reserve_session() + + assert session.session_id == "session-good" + assert len(cursor.rejections) == 1 + cleanup.assert_awaited_once() + + async def test_all_missing_control_urls_fail_selection(self) -> None: + cursor = _Cursor(_reservation("bad", control_url=None)) + with ( + mock.patch.object( + selection, + "runner_selector", + new=mock.AsyncMock(return_value=cursor), + ), + mock.patch.object(selection, "stop_runner_session", new=mock.AsyncMock()), + pytest.raises(NoRunnerAvailableError, match="missing control_url"), + ): + await selection.reserve_session() diff --git a/tests/test_selection.py b/tests/test_selection.py index b7d3140..e56d371 100644 --- a/tests/test_selection.py +++ b/tests/test_selection.py @@ -86,7 +86,11 @@ async def _call_runner( ) -> LiveRunnerCallResult: calls.append((runner.url, payload, method, timeout)) return LiveRunnerCallResult( - {"session_id": "session-1", "app_url": "https://orch-a/apps/a/app"}, + { + "session_id": "session-1", + "app_url": "https://orch-a/apps/a/app", + "control_url": "https://orch-a/apps/a/session/session-1", + }, runner_url=runner.url, runner=runner, session_id="session-1", @@ -336,7 +340,11 @@ async def _call_runner( ) -> LiveRunnerCallResult: del payload, method, timeout return LiveRunnerCallResult( - {"session_id": "session-1", "app_url": "https://orch-a/apps/a/app"}, + { + "session_id": "session-1", + "app_url": "https://orch-a/apps/a/app", + "control_url": "https://orch-a/apps/a/session/session-1", + }, runner_url=runner.url, runner=runner, session_id="session-1", From 95845f05e17a08551298b0d58697137f9d64f83c Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 15:41:32 -0700 Subject: [PATCH 14/18] refactor(live-runner): require session control URLs Signed-off-by: Josh Allmann --- src/livepeer_gateway/live_runner.py | 45 +++++++++++++++++++---------- src/livepeer_gateway/selection.py | 24 +++++---------- tests/test_live_runner.py | 1 + tests/test_live_runner_payments.py | 8 +++-- 4 files changed, 44 insertions(+), 34 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 0895ecd..639ee8c 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -119,16 +119,22 @@ class LiveRunnerSession: session_id: str app_url: str runner_url: str - runner: LiveRunnerInstance | None = None # Base URL for this session's control endpoints, as reported by the # orchestrator when the session was reserved. - control_url: str = "" + control_url: str + runner: LiveRunnerInstance | None = None # True once the orchestrator reported this session gone. released: bool = False _payment_task: asyncio.Task[None] | None = field( default=None, repr=False, compare=False ) + def __post_init__(self) -> None: + if not isinstance(self.control_url, str) or not self.control_url.strip(): + raise LivepeerGatewayError("Live runner session requires control_url") + self.control_url = self.control_url.strip() + _ = _payment_endpoint(self.control_url) + @property def payment_url(self) -> str: """Return this session's required session-scoped payment endpoint.""" @@ -1147,20 +1153,7 @@ async def stop_runner_session( await session.stop_payments() if session.released: return - - control_url = session.control_url.strip() - if control_url: - url = _join_endpoint(control_url, "stop") - else: - # Kept only for cleaning up a malformed legacy reservation. New - # reservations require control_url before they can be selected. - runner_url = session.runner_url.strip() - session_id = session.session_id.strip() - if not runner_url: - raise LivepeerGatewayError("Live runner session stop requires runner_url") - if not session_id: - raise LivepeerGatewayError("Live runner session stop requires session_id") - url = _join_endpoint(runner_url, f"/{quote(session_id, safe='')}/stop") + url = _join_endpoint(session.control_url, "stop") else: headers = getattr(session, "headers", None) get = getattr(headers, "get", None) @@ -1180,6 +1173,26 @@ async def stop_runner_session( session.released = True +async def _stop_runner_session_by_url( + runner_url: str, + session_id: str, + *, + timeout: float = 5.0, +) -> None: + """Best-effort cleanup path for a reservation with no valid control URL.""" + runner_url = runner_url.strip() + session_id = session_id.strip() + if not runner_url: + raise LivepeerGatewayError("Live runner session cleanup requires runner_url") + if not session_id: + raise LivepeerGatewayError("Live runner session cleanup requires session_id") + await post_empty( + _join_endpoint(runner_url, f"/{quote(session_id, safe='')}/stop"), + headers={}, + timeout=timeout, + ) + + def detect_process_gpu() -> LiveRunnerGPU | None: for detector in (_detect_gpu_pynvml, _detect_gpu_torch, _detect_gpu_nvidia_smi): try: diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index c96125e..ce2c630 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -24,8 +24,8 @@ LiveRunnerInstance, LiveRunnerSession, _live_runner_price_info_from_json, + _stop_runner_session_by_url, call_runner, - stop_runner_session, ) from .orch_info import get_orch_info @@ -326,17 +326,13 @@ def _reserved_session_from_result(result: LiveRunnerCallResult) -> LiveRunnerSes raise LivepeerGatewayError("runner session response missing app_url") if not control_url: raise LivepeerGatewayError("runner session response missing control_url") - session = LiveRunnerSession( + return LiveRunnerSession( session_id=session_id.strip(), app_url=app_url.strip(), runner_url=result.runner_url, - runner=result.runner, control_url=control_url, + runner=result.runner, ) - # Validate the reported URL before accepting the reservation or starting a - # background task. This also guarantees that funding is session-scoped. - _ = session.payment_url - return session async def _cleanup_rejected_reservation( @@ -347,16 +343,12 @@ async def _cleanup_rejected_reservation( session_id = _string_value(result.data.get("session_id")) if not session_id: return - # Do not trust a malformed control_url during cleanup. The runner URL plus - # session id names the same stop endpoint and is safe for this best effort. - session = LiveRunnerSession( - session_id=session_id, - app_url=_string_value(result.data.get("app_url")), - runner_url=result.runner_url, - runner=result.runner, - ) try: - await stop_runner_session(session, timeout=timeout) + await _stop_runner_session_by_url( + result.runner_url, + session_id, + timeout=timeout, + ) except Exception: _LOG.debug( "Failed to clean up rejected runner reservation %s", diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index 2adff3f..5f64826 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -132,6 +132,7 @@ def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: session_id="session-1", app_url="https://service.example.com/app", runner_url="https://service.example.com/apps/runner-1/session", + control_url="https://service.example.com/apps/runner-1/session/session-1", ) with mock.patch.object(live_runner, "post_empty", side_effect=_post_empty): diff --git a/tests/test_live_runner_payments.py b/tests/test_live_runner_payments.py index 4522f39..df0a6a5 100644 --- a/tests/test_live_runner_payments.py +++ b/tests/test_live_runner_payments.py @@ -315,7 +315,7 @@ async def test_invalid_control_url_rejects_candidate_and_tries_next( "runner_selector", new=mock.AsyncMock(return_value=cursor), ), - mock.patch.object(selection, "stop_runner_session", cleanup), + mock.patch.object(selection, "_stop_runner_session_by_url", cleanup), ): session = await selection.reserve_session() @@ -331,7 +331,11 @@ async def test_all_missing_control_urls_fail_selection(self) -> None: "runner_selector", new=mock.AsyncMock(return_value=cursor), ), - mock.patch.object(selection, "stop_runner_session", new=mock.AsyncMock()), + mock.patch.object( + selection, + "_stop_runner_session_by_url", + new=mock.AsyncMock(), + ), pytest.raises(NoRunnerAvailableError, match="missing control_url"), ): await selection.reserve_session() From 88aebc7f44cc140059e741ddd646679816f42031 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 18:36:55 -0700 Subject: [PATCH 15/18] refactor(live-runner): own server payment challenges --- src/livepeer_gateway/__init__.py | 3 +- src/livepeer_gateway/live_runner.py | 121 +++-------------------- src/livepeer_gateway/remote_signer.py | 87 ++++++----------- src/livepeer_gateway/selection.py | 52 ++-------- tests/test_live_payment_session.py | 133 ++++++-------------------- tests/test_live_runner.py | 69 ++++++++++--- tests/test_live_runner_payments.py | 83 ++++++---------- 7 files changed, 165 insertions(+), 383 deletions(-) diff --git a/src/livepeer_gateway/__init__.py b/src/livepeer_gateway/__init__.py index f474df5..bc23a23 100644 --- a/src/livepeer_gateway/__init__.py +++ b/src/livepeer_gateway/__init__.py @@ -61,7 +61,7 @@ ) from .discovery import discover_orchestrators, discover_runners from .orch_info import get_orch_info -from .remote_signer import LivePaymentSession, PaymentSession +from .remote_signer import LivePaymentChallenge, LivePaymentSession, PaymentSession from .scope import start_scope from .selection import ( RunnerSelectionCursor, @@ -107,6 +107,7 @@ "LiveRunnerSession", "LiveRunnerSessionCallback", "LiveRunnerSessionEvent", + "LivePaymentChallenge", "LivePaymentSession", "LiveRunnerProxy", "LivepeerGatewayError", diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 639ee8c..d2d7e90 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -20,7 +20,7 @@ overload, ) from collections.abc import AsyncIterator, Awaitable, Callable, Mapping -from urllib.parse import quote, unquote, urlparse, urlunparse +from urllib.parse import quote, urlparse, urlunparse import aiohttp from aiohttp.helpers import parse_mimetype @@ -30,6 +30,7 @@ from .http import _request_body, open_stream, post_empty, post_json, request_json from .remote_signer import ( GetPaymentResponse, + LivePaymentChallenge, LivePaymentSession, _freeze_headers, get_signer_info, @@ -133,19 +134,13 @@ def __post_init__(self) -> None: if not isinstance(self.control_url, str) or not self.control_url.strip(): raise LivepeerGatewayError("Live runner session requires control_url") self.control_url = self.control_url.strip() - _ = _payment_endpoint(self.control_url) - - @property - def payment_url(self) -> str: - """Return this session's required session-scoped payment endpoint.""" - return _payment_endpoint(self.control_url) + _ = _join_endpoint(self.control_url, "stop") 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), ) @@ -231,13 +226,11 @@ async def aiter_lines(self) -> AsyncIterator[str]: 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), ) @@ -840,14 +833,13 @@ async def call_runner( if signer_url: signer = await get_signer_info(signer_url, _freeze_headers(signer_headers)) payer_address = cast(str, signer.address) - challenge: _RunnerPaymentChallenge | None = None + challenge: LivePaymentChallenge | None = None attempts = (max(0, int(max_payment_challenge_retries)) + 1) * 2 for attempt in range(attempts): payment_session: LivePaymentSession | None = None payment_type = "" session_id = "" needs_ongoing_funding = False - payment_url = "" # No preferred format: the app, or the upstream it fronts, picks. Only # control-plane calls ask for JSON. request_headers: dict[str, str] = {"Accept": "*/*"} @@ -879,18 +871,6 @@ async def call_runner( # 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 _runner_id_from_url(runner_url) - ), - session_id, - ) - ) try: request_kwargs: dict[str, Any] = {"timeout": timeout} @@ -917,14 +897,12 @@ async def call_runner( ) # The stream outlives this call, so it owns the funding. if needs_ongoing_funding: - call_stream._start_payments( - cast(LivePaymentSession, payment_session), payment_url - ) + call_stream._start_payments(cast(LivePaymentSession, payment_session)) return call_stream # The request ends with this call, so the funding ends with it. pay_task = ( - _start_funding(cast(LivePaymentSession, payment_session), payment_url) + _start_funding(cast(LivePaymentSession, payment_session)) if needs_ongoing_funding else None ) @@ -975,14 +953,7 @@ async def call_runner( raise LivepeerGatewayError("Live runner call exhausted payment challenge retries") -@dataclass(frozen=True) -class _RunnerPaymentChallenge: - payment_params: str - orchestrator_url: str - manifest_id: str - - -def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentChallenge: +def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> LivePaymentChallenge: try: data = json.loads(error.body) except json.JSONDecodeError as e: @@ -991,70 +962,30 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC raise LivepeerGatewayError("Live runner payment challenge response must be a JSON object") payment_params = data.get("payment_params") - orchestrator_url = data.get("orchestrator") manifest_id = data.get("manifest_id") + payment_url = data.get("payment_url") if not isinstance(payment_params, str) or not payment_params: raise LivepeerGatewayError("Live runner payment challenge missing payment_params") - if not isinstance(orchestrator_url, str) or not orchestrator_url: - raise LivepeerGatewayError("Live runner payment challenge missing orchestrator") if not isinstance(manifest_id, str) or not manifest_id: raise LivepeerGatewayError("Live runner payment challenge missing manifest_id") + if not isinstance(payment_url, str) or not payment_url: + raise LivepeerGatewayError("Live runner payment challenge missing payment_url") - return _RunnerPaymentChallenge( + return LivePaymentChallenge( payment_params=payment_params, - orchestrator_url=orchestrator_url, manifest_id=manifest_id, + payment_url=payment_url, ) -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. - """ - if not (orchestrator_url and runner_id and session_id): - raise LivepeerGatewayError( - "Live runner session payment requires orchestrator, runner_id, and session_id" - ) - return _join_endpoint( - orchestrator_url, - f"/apps/{quote(runner_id, safe='')}/session/{quote(session_id, safe='')}", - ) - - -def _runner_id_from_url(runner_url: str) -> str: - """Extract the runner id from a canonical ``/apps/{id}/...`` URL.""" - parts = [part for part in urlparse(runner_url).path.split("/") if part] - try: - apps_index = parts.index("apps") - runner_id = parts[apps_index + 1] - except (ValueError, IndexError): - return "" - return unquote(runner_id) - - -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. - """ - if not control_url.strip(): - raise LivepeerGatewayError( - "Live runner session payment requires control_url; refusing generic /payment fallback" - ) - return _join_endpoint(control_url, "payment") - - def _start_funding( payment_session: LivePaymentSession, - payment_url: str, on_released: Callable[[], None] | 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) + released = await payment_session.run_payments() if released and on_released is not None: on_released() @@ -1071,7 +1002,7 @@ async def _stop_funding(task: asyncio.Task[None] | None) -> None: async def _get_runner_payment( - challenge: _RunnerPaymentChallenge, + challenge: LivePaymentChallenge, *, payment_type: str, signer_url: str, @@ -1081,9 +1012,7 @@ async def _get_runner_payment( signer_url=signer_url, signer_headers=signer_headers, type=payment_type, - payment_params=challenge.payment_params, - manifest_id=challenge.manifest_id, - orchestrator_url=challenge.orchestrator_url, + challenge=challenge, ) payment = await session.get_payment() if not payment.payment: @@ -1173,26 +1102,6 @@ async def stop_runner_session( session.released = True -async def _stop_runner_session_by_url( - runner_url: str, - session_id: str, - *, - timeout: float = 5.0, -) -> None: - """Best-effort cleanup path for a reservation with no valid control URL.""" - runner_url = runner_url.strip() - session_id = session_id.strip() - if not runner_url: - raise LivepeerGatewayError("Live runner session cleanup requires runner_url") - if not session_id: - raise LivepeerGatewayError("Live runner session cleanup requires session_id") - await post_empty( - _join_endpoint(runner_url, f"/{quote(session_id, safe='')}/stop"), - headers={}, - timeout=timeout, - ) - - def detect_process_gpu() -> LiveRunnerGPU | None: for detector in (_detect_gpu_pynvml, _detect_gpu_torch, _detect_gpu_nvidia_smi): try: diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index aeb8b7e..21a1e60 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -6,7 +6,7 @@ import logging import re import ssl -from dataclasses import dataclass +from dataclasses import dataclass, replace from functools import lru_cache from typing import Any, Optional from urllib.error import HTTPError, URLError @@ -32,6 +32,15 @@ class GetPaymentResponse: seg_creds: Optional[str] = None +@dataclass(frozen=True) +class LivePaymentChallenge: + """The complete payment contract returned by a live-runner 402.""" + + payment_params: str + manifest_id: str + payment_url: str + + @dataclass(frozen=True) class SignerMaterial: """ @@ -210,19 +219,15 @@ def __init__( *, signer_headers: dict[str, str] | None = None, type: str, - payment_params: str, - manifest_id: str, - orchestrator_url: str | None = None, + challenge: LivePaymentChallenge, max_refresh_retries: int = 3, ) -> None: self._signer_url = signer_url self._signer_headers = _freeze_headers(signer_headers) self._type = type - self._payment_params = payment_params - self._manifest_id = manifest_id + self._challenge = challenge self._max_refresh_retries = max(0, int(max_refresh_retries)) self._state: dict[str, Any] | None = None - self._orchestrator_url = orchestrator_url async def get_payment(self) -> GetPaymentResponse: if not self._signer_url: @@ -239,26 +244,11 @@ async def get_payment(self) -> GetPaymentResponse: ) from e if self._state is None: raise - orchestrator_url = e.orchestrator_url - if not orchestrator_url: - raise PaymentError( - "Signer refresh response missing Livepeer-Orchestrator-URL header" - ) from e - await self._refresh_payment_params(orchestrator_url) + await self._refresh_payment_params() attempts += 1 - 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. + async def send_payment(self) -> None: + """Generate a payment and POST it to the challenge's endpoint. Raises LivepeerHTTPError on error responses so callers can branch on the status code, and SkipPaymentCycle when the signer gates the cycle. @@ -266,15 +256,7 @@ async def send_payment( if not self._signer_url: return - from .http import _http_origin, post_empty - - 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" + from .http import post_empty payment = await self.get_payment() if not payment.seg_creds: @@ -285,25 +267,19 @@ async def send_payment( "Livepeer-Payment": payment.payment, "Livepeer-Segment": payment.seg_creds, } - await post_empty(url, headers=headers, timeout=5.0) + await post_empty(self._challenge.payment_url, headers=headers, timeout=5.0) - async def run_payments(self, *, payment_url: str) -> bool: + async def run_payments(self) -> 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. ``payment_url`` must be the session-scoped - endpoint; the generic ``/payment`` endpoint is intentionally unsupported - because it cannot report that the session is gone. Returns True if the - orchestrator reported it gone. + the caller pays upfront. Returns True if the orchestrator reports that + the challenge's session-scoped endpoint is gone. """ - if not payment_url.strip(): - raise PaymentError( - "session-scoped payment_url is required; refusing generic /payment fallback" - ) while True: await asyncio.sleep(PAYMENT_INTERVAL_S) try: - await self.send_payment(payment_url=payment_url) + await self.send_payment() except SkipPaymentCycle as e: _LOG.debug("Payment loop skipped cycle: %s", e) except LivepeerHTTPError as e: @@ -322,9 +298,9 @@ async def _payment_request(self) -> GetPaymentResponse: url = f"{_http_origin(self._signer_url)}/generate-live-payment" payload: dict[str, Any] = { - "orchestrator": self._payment_params, + "orchestrator": self._challenge.payment_params, "type": self._type, - "ManifestID": self._manifest_id, + "ManifestID": self._challenge.manifest_id, } if self._state is not None: payload["state"] = self._state @@ -352,19 +328,19 @@ async def _payment_request(self) -> GetPaymentResponse: self._state = state return GetPaymentResponse(payment=payment, seg_creds=seg_creds) - async def _refresh_payment_params(self, orchestrator_url: str) -> None: + async def _refresh_payment_params(self) -> None: from .http import _http_origin, post_json signer = await get_signer_info(self._signer_url or "", self._signer_headers) if not signer.address: raise PaymentError("Cannot refresh payment without signer address") - url = f"{_http_origin(orchestrator_url)}/refresh-payment" + url = f"{_http_origin(self._challenge.payment_url)}/refresh-payment" data = await post_json( url, { "sender": signer.address, - "manifest_id": self._manifest_id, + "manifest_id": self._challenge.manifest_id, }, ) payment_params = data.get("payment_params") @@ -372,12 +348,11 @@ async def _refresh_payment_params(self, orchestrator_url: str) -> None: raise PaymentError( f"RefreshPayment error: missing/invalid 'payment_params' in response (url={url})" ) - self._payment_params = payment_params - refreshed_orchestrator_url = data.get("orchestrator") - self._orchestrator_url = ( - refreshed_orchestrator_url - if isinstance(refreshed_orchestrator_url, str) and refreshed_orchestrator_url.strip() - else orchestrator_url + # Refresh rotates the embedded payment material. The initial scoped + # endpoint remains authoritative for the lifetime of this session. + self._challenge = replace( + self._challenge, + payment_params=payment_params, ) diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index ce2c630..9a5d47e 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -24,7 +24,6 @@ LiveRunnerInstance, LiveRunnerSession, _live_runner_price_info_from_json, - _stop_runner_session_by_url, call_runner, ) from .orch_info import get_orch_info @@ -294,29 +293,7 @@ async def reserve_session( gpu=gpu, timeout=timeout, ) - while True: - result = await cursor.next() - try: - session = _reserved_session_from_result(result) - except LivepeerGatewayError as e: - await _cleanup_rejected_reservation(result, timeout=timeout) - cursor.rejections.append( - RunnerRejection(url=result.runner_url, reason=str(e)) - ) - _LOG.debug( - "reserve_session rejected candidate %s: %s", - result.runner_url, - e, - ) - continue - - # 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 _reserved_session_from_result(result: LiveRunnerCallResult) -> LiveRunnerSession: + result = await cursor.next() session_id = result.data.get("session_id") app_url = result.data.get("app_url") control_url = _string_value(result.data.get("control_url")) @@ -326,7 +303,7 @@ def _reserved_session_from_result(result: LiveRunnerCallResult) -> LiveRunnerSes raise LivepeerGatewayError("runner session response missing app_url") if not control_url: raise LivepeerGatewayError("runner session response missing control_url") - return LiveRunnerSession( + session = LiveRunnerSession( session_id=session_id.strip(), app_url=app_url.strip(), runner_url=result.runner_url, @@ -334,27 +311,10 @@ def _reserved_session_from_result(result: LiveRunnerCallResult) -> LiveRunnerSes runner=result.runner, ) - -async def _cleanup_rejected_reservation( - result: LiveRunnerCallResult, - *, - timeout: float, -) -> None: - session_id = _string_value(result.data.get("session_id")) - if not session_id: - return - try: - await _stop_runner_session_by_url( - result.runner_url, - session_id, - timeout=timeout, - ) - except Exception: - _LOG.debug( - "Failed to clean up rejected runner reservation %s", - result.runner_url, - exc_info=True, - ) + # 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]: diff --git a/tests/test_live_payment_session.py b/tests/test_live_payment_session.py index 19a8d7c..372fe53 100644 --- a/tests/test_live_payment_session.py +++ b/tests/test_live_payment_session.py @@ -8,16 +8,27 @@ from livepeer_gateway import lp_rpc_pb2 from livepeer_gateway.errors import ( LivepeerHTTPError, - PaymentError, SignerRefreshRequired, ) from livepeer_gateway.remote_signer import ( + LivePaymentChallenge, LivePaymentSession, PaymentSession, get_signer_info, ) +_PAYMENT_URL = "https://orch.example.com/apps/runner/session/manifest-1/payment" + + +def _challenge(*, payment_params: str = "opaque") -> LivePaymentChallenge: + return LivePaymentChallenge( + payment_params=payment_params, + manifest_id="manifest-1", + payment_url=_PAYMENT_URL, + ) + + class TestPaymentSession: def test_get_payment_round_trips_state_without_cross_session_leak(self) -> None: calls: list[tuple[str, dict[str, object], dict[str, str] | None]] = [] @@ -88,12 +99,11 @@ async def test_none_signer_exits_early(self) -> None: session = LivePaymentSession( None, type="lv2v", - payment_params="opaque", - manifest_id="manifest-1", + challenge=_challenge(), ) payment = await session.get_payment() - await session.send_payment("https://orchestrator.example.com") + await session.send_payment() assert payment.payment == "" assert payment.seg_creds is None @@ -102,8 +112,7 @@ async def test_send_payment_reuses_empty_post_helper(self) -> None: session = LivePaymentSession( "https://signer.example.com", type="lv2v", - payment_params="opaque", - manifest_id="manifest-1", + challenge=_challenge(), ) post_empty = mock.AsyncMock() @@ -117,73 +126,19 @@ async def test_send_payment_reuses_empty_post_helper(self) -> None: ), mock.patch("livepeer_gateway.http.post_empty", post_empty), ): - await session.send_payment("https://orchestrator.example.com") + await session.send_payment() post_empty.assert_awaited_once_with( - "https://orchestrator.example.com/payment", + _PAYMENT_URL, headers={"Livepeer-Payment": "p", "Livepeer-Segment": "s"}, timeout=5.0, ) - async def test_send_payment_uses_constructor_orchestrator_url(self) -> None: - session = LivePaymentSession( - "https://signer.example.com", - type="lv2v", - payment_params="opaque", - manifest_id="manifest-1", - orchestrator_url="https://orchestrator.example.com/base", - ) - - post_empty = mock.AsyncMock() - with ( - mock.patch.object( - session, - "get_payment", - new=mock.AsyncMock( - return_value=types.SimpleNamespace(payment="p", seg_creds="s") - ), - ), - mock.patch("livepeer_gateway.http.post_empty", post_empty), - ): - await session.send_payment() - - assert post_empty.await_args.args[0] == "https://orchestrator.example.com/payment" - - async def test_send_payment_prefers_session_scoped_url(self) -> None: - session = LivePaymentSession( - "https://signer.example.com", - type="lv2v", - payment_params="opaque", - manifest_id="manifest-1", - ) - - post_empty = mock.AsyncMock() - with ( - mock.patch.object( - session, - "get_payment", - new=mock.AsyncMock( - return_value=types.SimpleNamespace(payment="p", seg_creds="s") - ), - ), - mock.patch("livepeer_gateway.http.post_empty", post_empty), - ): - await session.send_payment( - "https://orchestrator.example.com", - payment_url="https://orchestrator.example.com/apps/r/session/s/payment", - ) - - assert ( - post_empty.await_args.args[0] - == "https://orchestrator.example.com/apps/r/session/s/payment" - ) - async def test_send_payment_preserves_typed_http_error(self) -> None: session = LivePaymentSession( "https://signer.example.com", type="lv2v", - payment_params="opaque", - manifest_id="manifest-1", + challenge=_challenge(), ) error = LivepeerHTTPError( @@ -206,7 +161,7 @@ async def test_send_payment_preserves_typed_http_error(self) -> None: ), ): with pytest.raises(LivepeerHTTPError) as raised: - await session.send_payment("https://orchestrator.example.com") + await session.send_payment() assert raised.value is error @@ -233,8 +188,7 @@ async def _post_json( "https://signer.example.com", signer_headers={"Authorization": "token"}, type="lv2v", - payment_params="opaque-payment-params", - manifest_id="manifest-1", + challenge=_challenge(payment_params="opaque-payment-params"), ) first = await session.get_payment() second = await session.get_payment() @@ -271,8 +225,7 @@ async def _post_json( session = LivePaymentSession( "https://signer.example.com", type="lv2v", - payment_params="old-payment-params", - manifest_id="manifest-1", + challenge=_challenge(payment_params="old-payment-params"), ) with pytest.raises(SignerRefreshRequired): await session.get_payment() @@ -288,7 +241,7 @@ async def _post_json( ) ] - async def test_stateful_480_refreshes_payment_params_from_orchestrator_header( + async def test_stateful_480_refreshes_params_from_payment_url_origin( self, ) -> None: calls: list[tuple[str, dict[str, object]]] = [] @@ -313,10 +266,7 @@ async def _post_json( "state": {"state": "one"}, } if payment_requests == 2: - raise SignerRefreshRequired( - "refresh", - orchestrator_url="https://orch.example.com", - ) + raise SignerRefreshRequired("refresh") return { "payment": "payment-2", "segCreds": "segment-2", @@ -328,6 +278,8 @@ async def _post_json( return { "payment_params": "new-payment-params", "orchestrator": "https://orch.example.com", + "manifest_id": "manifest-1", + "payment_url": "https://orch.example.com/payment", } raise AssertionError(f"unexpected POST {url}") @@ -335,8 +287,7 @@ async def _post_json( session = LivePaymentSession( "https://signer.example.com", type="lv2v", - payment_params="old-payment-params", - manifest_id="manifest-1", + challenge=_challenge(payment_params="old-payment-params"), ) first_payment = await session.get_payment() payment = await session.get_payment() @@ -350,37 +301,7 @@ async def _post_json( ) assert calls[4][1]["orchestrator"] == "new-payment-params" - async def test_480_without_orchestrator_header_fails(self) -> None: - payment_requests = 0 - - async def _post_json( - url: str, - payload: dict[str, object], - *, - headers: dict[str, str] | None = None, - timeout: float = 5.0, - ) -> dict[str, object]: - nonlocal payment_requests - del url, payload, headers, timeout - payment_requests += 1 - if payment_requests == 1: - return { - "payment": "payment-1", - "segCreds": "segment-1", - "state": {"state": "one"}, - } - raise SignerRefreshRequired("refresh") - - with mock.patch("livepeer_gateway.http.post_json", side_effect=_post_json): - session = LivePaymentSession( - "https://signer.example.com", - type="lv2v", - payment_params="old-payment-params", - manifest_id="manifest-1", - ) - await session.get_payment() - with pytest.raises(PaymentError, match="missing Livepeer-Orchestrator-URL"): - await session.get_payment() + assert session._challenge.payment_url == _PAYMENT_URL async def test_get_signer_info_caches_result(self) -> None: calls: list[tuple[str, dict[str, object]]] = [] diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index 5f64826..1140c1d 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -26,6 +26,7 @@ stop_runner_session, create_proxy, ) +from livepeer_gateway.remote_signer import LivePaymentChallenge class TestLiveRunnerHelpers: @@ -43,6 +44,38 @@ def test_join_endpoint_preserves_base_path(self) -> None: == "https://orch.example.com:8935/base/runners/heartbeat" ) + def test_payment_challenge_uses_server_supplied_url(self) -> None: + body = json.dumps( + { + "payment_params": "opaque-payment-params", + "manifest_id": "manifest-1", + "payment_url": _payment_url("manifest-1"), + } + ) + + challenge = live_runner._parse_runner_payment_challenge( + LivepeerHTTPError(402, "https://runner.example.com", body) + ) + + assert challenge == _payment_challenge("manifest-1") + + @pytest.mark.parametrize("payment_url", [None, ""]) + def test_payment_challenge_requires_payment_url( + self, payment_url: str | None + ) -> None: + body = json.dumps( + { + "payment_params": "opaque-payment-params", + "manifest_id": "manifest-1", + "payment_url": payment_url, + } + ) + + with pytest.raises(LivepeerGatewayError, match="missing payment_url"): + live_runner._parse_runner_payment_challenge( + LivepeerHTTPError(402, "https://runner.example.com", body) + ) + def test_parse_go_duration(self) -> None: assert live_runner._parse_go_duration_s("500ms", default=5.0) == 0.5 assert live_runner._parse_go_duration_s("5s", default=1.0) == 5.0 @@ -289,9 +322,7 @@ def _request_body( "signer_url": "https://signer.example.com", "signer_headers": {"Authorization": "token"}, "type": "live", - "payment_params": "opaque-payment-params", - "manifest_id": "manifest-1", - "orchestrator_url": "https://orchestrator.example.com", + "challenge": _payment_challenge("manifest-1"), } ] assert result.payment_session is payment_sessions[0] @@ -360,9 +391,7 @@ def _request_body( "signer_url": "https://signer.example.com", "signer_headers": None, "type": "lv2v", - "payment_params": "opaque-payment-params", - "manifest_id": "manifest-scope", - "orchestrator_url": "https://orchestrator.example.com", + "challenge": _payment_challenge("manifest-scope"), } ] @@ -439,8 +468,8 @@ def _request_body( assert len(sessions) == 2 assert sessions[0]["type"] == "fixed" assert sessions[1]["type"] == "fixed" - assert sessions[0]["manifest_id"] == "fixed-manifest" - assert sessions[1]["manifest_id"] == "fixed-manifest" + assert sessions[0]["challenge"] == _payment_challenge("fixed-manifest") + assert sessions[1]["challenge"] == _payment_challenge("fixed-manifest") assert result.payment_session is None async def test_paid_call_restarts_challenge_when_signer_requests_refresh( @@ -533,17 +562,13 @@ def _request_body( "signer_url": "https://signer.example.com", "signer_headers": None, "type": "live", - "payment_params": "opaque-payment-params", - "manifest_id": "manifest-1", - "orchestrator_url": "https://orchestrator.example.com", + "challenge": _payment_challenge("manifest-1"), }, { "signer_url": "https://signer.example.com", "signer_headers": None, "type": "live", - "payment_params": "opaque-payment-params", - "manifest_id": "manifest-2", - "orchestrator_url": "https://orchestrator.example.com", + "challenge": _payment_challenge("manifest-2"), }, ] assert sig_mock.call_count == 1 @@ -615,10 +640,26 @@ def _payment_challenge_body(manifest_id: str) -> str: "payment_params": "opaque-payment-params", "orchestrator": "https://orchestrator.example.com", "manifest_id": manifest_id, + "payment_url": _payment_url(manifest_id), } ) +def _payment_challenge(manifest_id: str) -> LivePaymentChallenge: + return LivePaymentChallenge( + payment_params="opaque-payment-params", + manifest_id=manifest_id, + payment_url=_payment_url(manifest_id), + ) + + +def _payment_url(manifest_id: str) -> str: + return ( + "https://orchestrator.example.com/apps/runner-1/session/" + f"{manifest_id}/payment" + ) + + def _json_data(data: dict[str, object]) -> tuple[bytes, str]: return json.dumps(data).encode("utf-8"), "application/json" diff --git a/tests/test_live_runner_payments.py b/tests/test_live_runner_payments.py index df0a6a5..e9a3342 100644 --- a/tests/test_live_runner_payments.py +++ b/tests/test_live_runner_payments.py @@ -9,12 +9,10 @@ from livepeer_gateway.errors import ( LivepeerGatewayError, LivepeerHTTPError, - NoRunnerAvailableError, - PaymentError, SkipPaymentCycle, ) from livepeer_gateway.live_runner import LiveRunnerCallResult, LiveRunnerSession -from livepeer_gateway.remote_signer import LivePaymentSession +from livepeer_gateway.remote_signer import LivePaymentChallenge, LivePaymentSession _CONTROL_URL = "https://orch.example.com/apps/runner-1/session/session-1" @@ -29,20 +27,21 @@ def _live_payment_session() -> LivePaymentSession: return LivePaymentSession( "https://signer.example.com", type="live", - payment_params="opaque", - manifest_id="session-1", + challenge=LivePaymentChallenge( + payment_params="opaque", + manifest_id="session-1", + payment_url=_PAYMENT_URL, + ), ) class _FundingSession: def __init__(self, *, released: bool = False) -> None: self.released = released - self.urls: list[str] = [] self.started = asyncio.Event() self.cancelled = asyncio.Event() - async def run_payments(self, *, payment_url: str) -> bool: - self.urls.append(payment_url) + async def run_payments(self) -> bool: self.started.set() try: await asyncio.Future() @@ -62,10 +61,6 @@ def _session(*, control_url: str = _CONTROL_URL) -> LiveRunnerSession: class TestPaymentLoop: - async def test_requires_session_scoped_url(self) -> None: - with pytest.raises(PaymentError, match="session-scoped payment_url"): - await _live_payment_session().run_payments(payment_url="") - @pytest.mark.parametrize( "status, released", [(403, False), (404, True), (409, False)] ) @@ -82,12 +77,12 @@ async def test_terminal_status_stops_loop( mock.patch.object(remote_signer, "PAYMENT_INTERVAL_S", 0), ): result = await asyncio.wait_for( - payment_session.run_payments(payment_url=_PAYMENT_URL), + payment_session.run_payments(), timeout=1.0, ) assert result is released - send_payment.assert_awaited_once_with(payment_url=_PAYMENT_URL) + send_payment.assert_awaited_once_with() @pytest.mark.parametrize( "first_error", @@ -103,7 +98,7 @@ async def test_retryable_error_reaches_next_cycle( mock.patch.object(remote_signer, "PAYMENT_INTERVAL_S", 0), ): released = await asyncio.wait_for( - payment_session.run_payments(payment_url=_PAYMENT_URL), + payment_session.run_payments(), timeout=1.0, ) @@ -112,24 +107,20 @@ async def test_retryable_error_reaches_next_cycle( class TestSessionPaymentLifecycle: - def test_payment_url_is_derived_from_control_url(self) -> None: - assert _session().payment_url == _PAYMENT_URL - @pytest.mark.parametrize("control_url", ["", "ftp://orch/session/session-1"]) - def test_payment_url_rejects_missing_or_invalid_control_url( + def test_session_rejects_missing_or_invalid_control_url( self, control_url: str ) -> None: with pytest.raises(LivepeerGatewayError): - _session(control_url=control_url).payment_url + _session(control_url=control_url) - async def test_start_payments_uses_only_session_scoped_endpoint(self) -> None: + async def test_start_payments_starts_challenge_owned_session(self) -> None: payment_session = _FundingSession() session = _session() session._start_payments(payment_session) # type: ignore[arg-type] await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) - assert payment_session.urls == [_PAYMENT_URL] await session.stop_payments() async def test_start_payments_is_idempotent(self) -> None: @@ -251,10 +242,7 @@ def __init__(self, *results: LiveRunnerCallResult) -> None: async def next(self) -> LiveRunnerCallResult: if self.results: return self.results.pop(0) - raise NoRunnerAvailableError( - f"All runners failed ({len(self.rejections)} tried)", - rejections=list(self.rejections), - ) + raise AssertionError("unexpected extra runner selection") def _reservation( @@ -293,49 +281,36 @@ async def test_paid_reservation_starts_scoped_funding(self) -> None: session = await selection.reserve_session() await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) - assert payment_session.urls == [_PAYMENT_URL] await session.stop_payments() @pytest.mark.parametrize( "bad_control_url", [None, "ftp://orch.example.com/session/bad"], ) - async def test_invalid_control_url_rejects_candidate_and_tries_next( + async def test_invalid_control_url_fails_immediately( self, bad_control_url: str | None ) -> None: cursor = _Cursor( _reservation("bad", control_url=bad_control_url), _reservation("good", control_url=_CONTROL_URL), ) - cleanup = mock.AsyncMock() - with ( - mock.patch.object( - selection, - "runner_selector", - new=mock.AsyncMock(return_value=cursor), - ), - mock.patch.object(selection, "_stop_runner_session_by_url", cleanup), + with mock.patch.object( + selection, + "runner_selector", + new=mock.AsyncMock(return_value=cursor), ): - session = await selection.reserve_session() + with pytest.raises(LivepeerGatewayError): + await selection.reserve_session() - assert session.session_id == "session-good" - assert len(cursor.rejections) == 1 - cleanup.assert_awaited_once() + assert len(cursor.results) == 1 - async def test_all_missing_control_urls_fail_selection(self) -> None: + async def test_missing_control_url_is_contract_error(self) -> None: cursor = _Cursor(_reservation("bad", control_url=None)) - with ( - mock.patch.object( - selection, - "runner_selector", - new=mock.AsyncMock(return_value=cursor), - ), - mock.patch.object( - selection, - "_stop_runner_session_by_url", - new=mock.AsyncMock(), - ), - pytest.raises(NoRunnerAvailableError, match="missing control_url"), + with mock.patch.object( + selection, + "runner_selector", + new=mock.AsyncMock(return_value=cursor), ): - await selection.reserve_session() + with pytest.raises(LivepeerGatewayError, match="missing control_url"): + await selection.reserve_session() From e21f31f8db5ff12b92cf0737a21f377c183418cf Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 18:47:05 -0700 Subject: [PATCH 16/18] refactor(live-runner): centralize session close --- src/livepeer_gateway/live_runner.py | 6 +----- tests/test_live_runner_payments.py | 16 +--------------- 2 files changed, 2 insertions(+), 20 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index d2d7e90..c93b4da 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -154,11 +154,7 @@ async def stop_payments(self) -> None: self._payment_task = None async def aclose(self) -> None: - # Stop funding first so a slow or failed remote stop cannot mint another - # payment while this session is being closed. - await self.stop_payments() - if not self.released: - await stop_runner_session(self) + await stop_runner_session(self) async def __aenter__(self) -> LiveRunnerSession: return self diff --git a/tests/test_live_runner_payments.py b/tests/test_live_runner_payments.py index e9a3342..35d3987 100644 --- a/tests/test_live_runner_payments.py +++ b/tests/test_live_runner_payments.py @@ -145,29 +145,15 @@ async def test_stop_payments_cancels_and_clears_task(self) -> None: assert session._payment_task is None assert payment_session.cancelled.is_set() - async def test_aclose_stops_funding_then_remote_session(self) -> None: - payment_session = _FundingSession() + async def test_aclose_delegates_to_stop_runner_session(self) -> None: session = _session() - session._start_payments(payment_session) # type: ignore[arg-type] - await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) stop = mock.AsyncMock() with mock.patch.object(live_runner, "stop_runner_session", stop): await session.aclose() - assert payment_session.cancelled.is_set() stop.assert_awaited_once_with(session) - async def test_aclose_skips_remote_stop_when_already_released(self) -> None: - session = _session() - session.released = True - stop = mock.AsyncMock() - - with mock.patch.object(live_runner, "stop_runner_session", stop): - await session.aclose() - - stop.assert_not_awaited() - async def test_async_context_manager_closes_session(self) -> None: session = _session() stop = mock.AsyncMock() From cd205df3d46fb53cb8e0347372d34378efed12f7 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 18:50:45 -0700 Subject: [PATCH 17/18] refactor(http): mark empty post helper internal --- src/livepeer_gateway/http.py | 2 +- src/livepeer_gateway/live_runner.py | 6 +++--- src/livepeer_gateway/remote_signer.py | 4 ++-- tests/test_live_payment_session.py | 4 ++-- tests/test_live_runner.py | 6 +++--- tests/test_live_runner_payments.py | 6 +++--- 6 files changed, 14 insertions(+), 14 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index b26a165..699d042 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -403,7 +403,7 @@ async def get_json( return await request_json(url, headers=headers, timeout=timeout) -async def post_empty( +async def _post_empty( url: str, *, headers: dict[str, str] | None = None, diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index c93b4da..45193dc 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -27,7 +27,7 @@ from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import _request_body, open_stream, post_empty, post_json, request_json +from .http import _post_empty, _request_body, open_stream, post_json, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentChallenge, @@ -369,7 +369,7 @@ 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"), headers={"Authorization": secret}, timeout=self._timeout, @@ -1089,7 +1089,7 @@ 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, headers=request_headers, timeout=timeout, diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 21a1e60..6505816 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -256,7 +256,7 @@ async def send_payment(self) -> None: if not self._signer_url: return - from .http import post_empty + from .http import _post_empty payment = await self.get_payment() if not payment.seg_creds: @@ -267,7 +267,7 @@ async def send_payment(self) -> None: "Livepeer-Payment": payment.payment, "Livepeer-Segment": payment.seg_creds, } - await post_empty(self._challenge.payment_url, headers=headers, timeout=5.0) + await _post_empty(self._challenge.payment_url, headers=headers, timeout=5.0) async def run_payments(self) -> bool: """Keep a metered session funded until cancelled or the session ends. diff --git a/tests/test_live_payment_session.py b/tests/test_live_payment_session.py index 372fe53..65dda6a 100644 --- a/tests/test_live_payment_session.py +++ b/tests/test_live_payment_session.py @@ -124,7 +124,7 @@ async def test_send_payment_reuses_empty_post_helper(self) -> None: return_value=types.SimpleNamespace(payment="p", seg_creds="s") ), ), - mock.patch("livepeer_gateway.http.post_empty", post_empty), + mock.patch("livepeer_gateway.http._post_empty", post_empty), ): await session.send_payment() @@ -156,7 +156,7 @@ async def test_send_payment_preserves_typed_http_error(self) -> None: ), ), mock.patch( - "livepeer_gateway.http.post_empty", + "livepeer_gateway.http._post_empty", new=mock.AsyncMock(side_effect=error), ), ): diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index 1140c1d..e2ca113 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -168,7 +168,7 @@ def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: control_url="https://service.example.com/apps/runner-1/session/session-1", ) - with mock.patch.object(live_runner, "post_empty", side_effect=_post_empty): + with mock.patch.object(live_runner, "_post_empty", side_effect=_post_empty): await stop_runner_session(session) assert stopped == [ @@ -192,7 +192,7 @@ def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: } ) - with mock.patch.object(live_runner, "post_empty", side_effect=_post_empty): + with mock.patch.object(live_runner, "_post_empty", side_effect=_post_empty): await stop_runner_session(request, timeout=12.0) assert stopped == [ @@ -1068,7 +1068,7 @@ def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: "heartbeat_secret": "heartbeat-token", }, ): - with mock.patch.object(live_runner, "post_empty", side_effect=_post_empty): + with mock.patch.object(live_runner, "_post_empty", side_effect=_post_empty): reg = await register_runner( "http://orch.example.com", secret="secret-token", diff --git a/tests/test_live_runner_payments.py b/tests/test_live_runner_payments.py index 35d3987..c6fab96 100644 --- a/tests/test_live_runner_payments.py +++ b/tests/test_live_runner_payments.py @@ -173,7 +173,7 @@ async def test_stops_payment_task_and_uses_control_url(self) -> None: await asyncio.wait_for(payment_session.started.wait(), timeout=1.0) post_empty = mock.AsyncMock() - with mock.patch.object(live_runner, "post_empty", post_empty): + with mock.patch.object(live_runner, "_post_empty", post_empty): await live_runner.stop_runner_session(session, timeout=12.0) assert payment_session.cancelled.is_set() @@ -194,7 +194,7 @@ async def test_remote_stop_failure_still_stops_payment_task(self) -> None: with ( mock.patch.object( live_runner, - "post_empty", + "_post_empty", new=mock.AsyncMock(side_effect=LivepeerGatewayError("stop failed")), ), pytest.raises(LivepeerGatewayError, match="stop failed"), @@ -213,7 +213,7 @@ async def test_already_released_session_only_stops_local_funding(self) -> None: session.released = True post_empty = mock.AsyncMock() - with mock.patch.object(live_runner, "post_empty", post_empty): + with mock.patch.object(live_runner, "_post_empty", post_empty): await live_runner.stop_runner_session(session) assert payment_session.cancelled.is_set() From c49e5c65be1fc6c0221619d3ca5fb2ad6a4aa1bd Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 19:07:06 -0700 Subject: [PATCH 18/18] docs(live-runner): trim session comments --- src/livepeer_gateway/live_runner.py | 10 +--------- 1 file changed, 1 insertion(+), 9 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 45193dc..deb981d 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -109,19 +109,11 @@ class LiveRunnerInstance: @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. - """ + """A reserved live runner session.""" session_id: str app_url: str runner_url: str - # Base URL for this session's control endpoints, as reported by the - # orchestrator when the session was reserved. control_url: str runner: LiveRunnerInstance | None = None # True once the orchestrator reported this session gone.