From 63cecd6a49ff6a82e48d31d3cb1f5113d017ac1b Mon Sep 17 00:00:00 2001 From: seanhanca Date: Tue, 28 Jul 2026 17:49:46 -0700 Subject: [PATCH] fix(live-runner): send inPixels:1 for fixed-price runner payments A fixed-price single-shot live-runner generation advertises price_info.unit == "fixed" and the v0.9.0 orchestrator debits exactly one unit per generation (PixelsPerUnit == 1). The remote signer types these payments as "lv2v" and, absent an explicit unit count, recomputes pixels as a continuous 720p30 estimate (720*1280*30*60 = 1,658,880,000), inflating the per-request fixed fee ~1.66e9x and blowing past the max-100 ticket guard (observed e2e: HTTP 400 "numTickets 2721947758 exceeds maximum of 100"). Thread the runner's advertised pricing unit through _get_runner_payment and, for fixed-unit runners, send inPixels:1 to /generate-live-payment so the fee equals the per-request fixed price (numTickets ~1). Continuous runners (720p/hour) send no inPixels, so their payload is byte-identical and the signer keeps its automatic estimate. Gateway half of the paired fix. Requires go-livepeer PR #4006 (commit cd99507), which teaches the signer to honor req.InPixels on the lv2v path; both must ship together. Co-authored-by: Cursor --- src/livepeer_gateway/live_runner.py | 35 +++++++ src/livepeer_gateway/remote_signer.py | 9 ++ tests/test_live_runner_fixed_price.py | 139 ++++++++++++++++++++++++++ 3 files changed, 183 insertions(+) create mode 100644 tests/test_live_runner_fixed_price.py diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 04cb2a0..5f9a0f3 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -738,6 +738,40 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC ) +_FIXED_PRICE_UNIT = "fixed" + + +def _runner_price_unit(runner: Optional[LiveRunnerInstance]) -> str: + """Return the runner's advertised pricing unit, lowercased. + + The unit is read from the discovery/offering ``price_info`` carried on the + runner (``runner.raw["price_info"]["unit"]``). Returns ``""`` when the runner + or its price info is missing or malformed. + """ + if runner is None: + return "" + price_info = runner.raw.get("price_info") + if not isinstance(price_info, dict): + return "" + unit = price_info.get("unit") + return unit.strip().lower() if isinstance(unit, str) else "" + + +def _fixed_price_in_pixels(runner: Optional[LiveRunnerInstance]) -> Optional[int]: + """Unit count to bill for a runner's ``lv2v`` payment, or ``None``. + + A fixed-price single-shot runner advertises ``price_info.unit == "fixed"`` and + the v0.9.0 orchestrator debits exactly one unit per generation (with + ``PixelsPerUnit == 1``). Return ``1`` so the gateway sends ``inPixels:1`` and + the fee equals the per-request fixed price. Return ``None`` for continuous + runners (e.g. ``720p``/``hour``) so the signer keeps its automatic 720p30 + estimate and the request payload is unchanged. + """ + if _runner_price_unit(runner) == _FIXED_PRICE_UNIT: + return 1 + return None + + async def _get_runner_payment( challenge: _RunnerPaymentChallenge, *, @@ -754,6 +788,7 @@ async def _get_runner_payment( manifest_id=challenge.manifest_id, orchestrator_url=challenge.orchestrator_url, capabilities=byoc_capabilities_from_app(app), + in_pixels=_fixed_price_in_pixels(runner), ) payment = await session.get_payment() if not payment.payment: diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index fe3b916..f91176f 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -207,6 +207,7 @@ def __init__( manifest_id: str, orchestrator_url: Optional[str] = None, capabilities: Optional[lp_rpc_pb2.Capabilities] = None, + in_pixels: Optional[int] = None, max_refresh_retries: int = 3, ) -> None: self._signer_url = signer_url @@ -215,6 +216,12 @@ def __init__( self._payment_params = payment_params self._manifest_id = manifest_id self._capabilities = capabilities + # Explicit unit count for the signer's `inPixels`. When set (e.g. 1 for a + # fixed-price single-shot live-runner generation) it is forwarded to + # /generate-live-payment and takes precedence over the signer's automatic + # continuous 720p30 estimate on the lv2v path. Left None for continuous + # live-video runners so their payload stays byte-identical. + self._in_pixels = in_pixels self._max_refresh_retries = max(0, int(max_refresh_retries)) self._state: Optional[dict[str, Any]] = None self._orchestrator_url = orchestrator_url @@ -290,6 +297,8 @@ async def _payment_request(self) -> GetPaymentResponse: "type": self._type, "ManifestID": self._manifest_id, } + if self._in_pixels is not None: + payload["inPixels"] = self._in_pixels if self._capabilities is not None: payload["capabilities"] = base64.b64encode( self._capabilities.SerializeToString() diff --git a/tests/test_live_runner_fixed_price.py b/tests/test_live_runner_fixed_price.py new file mode 100644 index 0000000..4000347 --- /dev/null +++ b/tests/test_live_runner_fixed_price.py @@ -0,0 +1,139 @@ +"""Unit tests for fixed-price live-runner payment signalling. + +A fixed-price single-shot live-runner generation advertises +``price_info.unit == "fixed"`` and the v0.9.0 orchestrator debits exactly one +unit per generation (``PixelsPerUnit == 1``). The gateway must signal that unit +count to the remote signer as ``inPixels:1`` on the ``lv2v`` +``/generate-live-payment`` request; otherwise the signer falls back to the +continuous 720p30 estimate (720*1280*30*60 = 1,658,880,000 pixels) and inflates +the fee ~1.66e9x, blowing past the signer's max-100 ticket guard (observed e2e: +HTTP 400 "numTickets 2721947758 exceeds maximum of 100"). + +Pairs with go-livepeer PR #4006 (commit cd99507), which teaches the signer to +honor ``req.InPixels`` on the ``lv2v`` path. Both must ship together. + +Continuous runners (``720p``/``hour``) must NOT set ``inPixels`` so the payload +stays byte-identical and the signer keeps its automatic estimate. +""" +from __future__ import annotations + +import asyncio +from typing import Any, Optional +from unittest.mock import AsyncMock, patch + +from livepeer_gateway.live_runner import ( + LiveRunnerInstance, + _RunnerPaymentChallenge, + _fixed_price_in_pixels, + _get_runner_payment, + _runner_price_unit, +) + + +def _runner(unit: Optional[str]) -> LiveRunnerInstance: + """Build a discovered runner whose price_info advertises ``unit``. + + Passing ``unit=None`` omits ``price_info`` entirely to model a discovery + entry that carries no pricing unit. + """ + raw: dict[str, Any] = { + "url": "http://runner", + "app": "storyboard/fal-flux-schnell", + "runner_id": "r1", + "mode": "single-shot", + } + if unit is not None: + raw["price_info"] = { + "price_per_unit": 1284088677165, + "pixels_per_unit": 1, + "unit": unit, + } + return LiveRunnerInstance( + url="http://runner", + app="storyboard/fal-flux-schnell", + runner_id="r1", + mode="single-shot", + orchestrator_url="http://orch", + raw=raw, + ) + + +def _challenge() -> _RunnerPaymentChallenge: + return _RunnerPaymentChallenge( + payment_params="orch-payment-params-b64", + orchestrator_url="http://orch", + manifest_id="manifest-1", + ) + + +def _signer_response() -> dict[str, Any]: + return {"payment": "payment-b64", "segCreds": "seg-creds-b64", "state": {}} + + +def _capture_payment(runner: Optional[LiveRunnerInstance]) -> dict[str, Any]: + """Run ``_get_runner_payment`` and return the payload POSTed to the signer.""" + + async def go() -> dict[str, Any]: + post_json = AsyncMock(return_value=_signer_response()) + with patch("livepeer_gateway.http.post_json", post_json): + _, payment = await _get_runner_payment( + _challenge(), + signer_url="http://signer", + signer_headers=None, + runner=runner, + ) + assert payment.payment == "payment-b64" + assert payment.seg_creds == "seg-creds-b64" + post_json.assert_awaited_once() + # post_json(url, payload, headers=...) -> payload is positional arg 1. + return post_json.await_args.args[1] + + return asyncio.run(go()) + + +def test_fixed_unit_runner_sets_in_pixels_1() -> None: + payload = _capture_payment(_runner("fixed")) + assert payload["type"] == "lv2v" + assert payload["inPixels"] == 1 + + +def test_fixed_unit_is_case_insensitive() -> None: + payload = _capture_payment(_runner("Fixed")) + assert payload["inPixels"] == 1 + + +def test_continuous_720p_runner_omits_in_pixels() -> None: + payload = _capture_payment(_runner("720p")) + assert payload["type"] == "lv2v" + assert "inPixels" not in payload + + +def test_continuous_hour_runner_omits_in_pixels() -> None: + payload = _capture_payment(_runner("hour")) + assert "inPixels" not in payload + + +def test_missing_price_info_omits_in_pixels() -> None: + payload = _capture_payment(_runner(None)) + assert "inPixels" not in payload + + +def test_runner_none_omits_in_pixels() -> None: + payload = _capture_payment(None) + assert "inPixels" not in payload + + +def test_runner_price_unit_helper() -> None: + assert _runner_price_unit(_runner("fixed")) == "fixed" + assert _runner_price_unit(_runner(" Fixed ")) == "fixed" + assert _runner_price_unit(_runner("720p")) == "720p" + assert _runner_price_unit(_runner(None)) == "" + assert _runner_price_unit(None) == "" + + +def test_fixed_price_in_pixels_helper() -> None: + assert _fixed_price_in_pixels(_runner("fixed")) == 1 + assert _fixed_price_in_pixels(_runner("720p")) is None + assert _fixed_price_in_pixels(_runner("hour")) is None + assert _fixed_price_in_pixels(_runner(None)) is None + assert _fixed_price_in_pixels(None) is None