diff --git a/CHANGELOG.md b/CHANGELOG.md index a067f9f..bb0299c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- `call_runner(on_payment_session=...)` callback for each new payment session +- `LivePaymentSession.manifest_id` for payment attribution + ### Changed - `discover_orchestrators` is now asynchronous and must be awaited, including diff --git a/src/livepeer_gateway/__init__.py b/src/livepeer_gateway/__init__.py index 9e4c284..eb9274c 100644 --- a/src/livepeer_gateway/__init__.py +++ b/src/livepeer_gateway/__init__.py @@ -37,6 +37,7 @@ LiveRunnerSessionCallback, LiveRunnerSessionEvent, LiveRunnerProxy, + LivePaymentSessionCallback, call_runner, create_proxy, create_trickle_channels, @@ -78,6 +79,7 @@ "LiveRunnerSessionEvent", "LivePaymentChallenge", "LivePaymentSession", + "LivePaymentSessionCallback", "LiveRunnerProxy", "LivepeerGatewayError", "LivepeerHTTPError", diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 8d80b3b..d94bae7 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -33,6 +33,7 @@ GetPaymentResponse, LivePaymentChallenge, LivePaymentSession, + _SignedPaymentError, _freeze_headers, get_signer_info, ) @@ -87,6 +88,12 @@ class LiveRunnerSessionEvent: LiveRunnerSessionCallback = Callable[[LiveRunnerSessionEvent], None | Awaitable[None]] +LivePaymentSessionCallback = Callable[[LivePaymentSession], None | Awaitable[None]] + + +@dataclass(frozen=True) +class _PaymentSessionCallbackFailure: + error: BaseException @dataclass(frozen=True) @@ -762,6 +769,7 @@ async def call_runner( payment_unit: str | None = ..., timeout: float = ..., max_payment_challenge_retries: int = ..., + on_payment_session: LivePaymentSessionCallback | None = ..., stream: Literal[False] = False, ) -> LiveRunnerCallResult: pass @@ -779,6 +787,7 @@ async def call_runner( payment_unit: str | None = ..., timeout: float = ..., max_payment_challenge_retries: int = ..., + on_payment_session: LivePaymentSessionCallback | None = ..., stream: Literal[True], ) -> LiveRunnerCallStream: pass @@ -795,6 +804,7 @@ async def call_runner( payment_unit: str | None = None, timeout: float = 5.0, max_payment_challenge_retries: int = 3, + on_payment_session: LivePaymentSessionCallback | None = None, stream: bool = False, ) -> LiveRunnerCallResult | LiveRunnerCallStream: """Call a runner once and return its result (or a live stream if ``stream=True``). @@ -803,6 +813,11 @@ async def call_runner( paid via the signer and retried (up to ``max_payment_challenge_retries``), one job, one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. + After a payment string is obtained, `on_payment_session` is called. + Metered renewals do not invoke it. The callback may be sync or + async. Its exceptions propagate and abort the call. Cancellation waits for + an in-progress signer request and callback; callback errors still propagate. + ``application/json`` and ``+json`` types parse into ``result.data``; anything else (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``. @@ -833,14 +848,18 @@ async def call_runner( request_headers[_LIVE_RUNNER_PAYER_ADDRESS_HEADER] = payer_address # Pending challenge means payment is needed. if challenge is not None: + payment_session = LivePaymentSession( + signer_url=signer_url or "", + signer_headers=signer_headers, + type=payment_type, + challenge=challenge, + app=runner.app if runner is not None else None, + max_price=max_price.to_json() if max_price is not None else None, + ) try: - payment_session, payment = await _get_runner_payment( - challenge, - payment_type=payment_type, - app=runner.app if runner is not None else None, - signer_url=signer_url or "", - signer_headers=signer_headers, - max_price=max_price, + payment_outcome = await _await_runner_payment( + payment_session, + on_payment_session=on_payment_session, ) except SignerRefreshRequired as e: if attempt + 1 >= attempts: @@ -852,8 +871,15 @@ async def call_runner( ) challenge = None continue + if isinstance(payment_outcome, _PaymentSessionCallbackFailure): + raise payment_outcome.error + payment = payment_outcome + if not payment.seg_creds: + raise LivepeerGatewayError( + "Live runner payment response missing segCreds" + ) request_headers["Livepeer-Payment"] = payment.payment - request_headers["Livepeer-Segment"] = payment.seg_creds or "" + request_headers["Livepeer-Segment"] = cast(str, payment.seg_creds) session_id = challenge.manifest_id # Metered pricing bills for as long as the work runs. @@ -940,6 +966,77 @@ async def call_runner( raise LivepeerGatewayError("Live runner call exhausted payment challenge retries") +async def _await_runner_payment( + payment_session: LivePaymentSession, + *, + on_payment_session: LivePaymentSessionCallback | None, +) -> GetPaymentResponse | _PaymentSessionCallbackFailure: + if on_payment_session is None: + return await payment_session.get_payment() + + async def _pay_and_report() -> GetPaymentResponse | _PaymentSessionCallbackFailure: + try: + payment_outcome = await payment_session.get_payment() + except _SignedPaymentError as error: + payment_outcome = error + callback_error = await _notify_payment_session( + on_payment_session, + payment_session, + ) + if callback_error is not None: + return _PaymentSessionCallbackFailure(callback_error) + if isinstance(payment_outcome, _SignedPaymentError): + raise payment_outcome + return payment_outcome + + payment_task = asyncio.create_task(_pay_and_report()) + waiter = asyncio.gather(payment_task, return_exceptions=True) + cancelled = False + while not waiter.done(): + try: + await asyncio.shield(waiter) + # Reporting must finish even if the caller cancels more than once. + except asyncio.CancelledError: # NOSONAR + cancelled = True + + return _runner_payment_result(payment_task, cancelled=cancelled) + + +def _runner_payment_result( + payment_task: asyncio.Task[ + GetPaymentResponse | _PaymentSessionCallbackFailure + ], + *, + cancelled: bool, +) -> GetPaymentResponse | _PaymentSessionCallbackFailure: + if payment_task.cancelled(): + raise asyncio.CancelledError + error = payment_task.exception() + if error is not None: + if cancelled: + raise asyncio.CancelledError + raise error + result = payment_task.result() + if isinstance(result, _PaymentSessionCallbackFailure): + return result + if cancelled: + raise asyncio.CancelledError + return result + + +async def _notify_payment_session( + callback: LivePaymentSessionCallback, + payment_session: LivePaymentSession, +) -> BaseException | None: + try: + result = callback(payment_session) + if inspect.isawaitable(result): + await result + except (Exception, asyncio.CancelledError) as error: + return error + return None + + def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> LivePaymentChallenge: try: data = json.loads(error.body) @@ -988,31 +1085,6 @@ async def _stop_funding(task: asyncio.Task[None] | None) -> None: await task -async def _get_runner_payment( - challenge: LivePaymentChallenge, - *, - payment_type: str, - signer_url: str, - signer_headers: dict[str, str] | None, - max_price: LiveRunnerPriceInfo | None, - app: str | None = None, -) -> tuple[LivePaymentSession, GetPaymentResponse]: - session = LivePaymentSession( - signer_url=signer_url, - signer_headers=signer_headers, - type=payment_type, - challenge=challenge, - app=app, - max_price=max_price.to_json() if max_price is not None else None, - ) - payment = await session.get_payment() - if not payment.payment: - raise LivepeerGatewayError("Live runner payment response missing payment") - if not payment.seg_creds: - raise LivepeerGatewayError("Live runner payment response missing segCreds") - return session, payment - - def _runner_payment_type( runner: LiveRunnerInstance | None, payment_unit: str | None = None, diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 1c027bf..94ecfd8 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -18,6 +18,11 @@ # Must stay under the signer's opening payment of 10 seconds. PAYMENT_INTERVAL_S = 3.0 + +class _SignedPaymentError(PaymentError): + """The signer returned a payment string but an unusable response.""" + + @dataclass(frozen=True) class GetPaymentResponse: payment: str @@ -131,6 +136,11 @@ def __init__( self._max_refresh_retries = max(0, int(max_refresh_retries)) self._state: dict[str, Any] | None = None + @property + def manifest_id(self) -> str: + """The manifest ID of the payment challenge for this session.""" + return self._challenge.manifest_id + async def get_payment(self) -> GetPaymentResponse: if not self._signer_url: return GetPaymentResponse(payment="", seg_creds=None) @@ -220,14 +230,17 @@ async def _payment_request(self) -> GetPaymentResponse: ) seg_creds = data.get("segCreds") + # A non-string segment is unusable and carries no cursor. A blank one + # (missing or "") can still advance the signer's state, so store that + # before the caller refuses to send it. if seg_creds is not None and not isinstance(seg_creds, str): - raise PaymentError( + raise _SignedPaymentError( f"GetPayment error: invalid 'segCreds' in response (url={url})" ) state = data.get("state") if not isinstance(state, dict): - raise PaymentError( + raise _SignedPaymentError( f"Remote signer response missing 'state' object (url={url})" ) diff --git a/tests/test_live_payment_session.py b/tests/test_live_payment_session.py index e96af45..bc74717 100644 --- a/tests/test_live_payment_session.py +++ b/tests/test_live_payment_session.py @@ -7,6 +7,7 @@ from livepeer_gateway.errors import ( LivepeerHTTPError, + PaymentError, SignerRefreshRequired, ) from livepeer_gateway.remote_signer import ( @@ -268,6 +269,61 @@ async def _post_json( assert calls[4][1]["orchestrator"] == "new-payment-params" assert session._challenge.payment_url == _PAYMENT_URL + assert session.manifest_id == "manifest-1" + + @pytest.mark.parametrize("seg_creds", [None, ""]) + async def test_blank_segment_renewal_sends_the_new_cursor( + self, seg_creds: str | None + ) -> None: + calls: list[dict[str, object]] = [] + responses = [ + { + "payment": "payment-1", + "segCreds": "segment-1", + "state": {"cursor": "previous"}, + }, + { + "payment": "payment-2", + "segCreds": seg_creds, + "state": {"cursor": "advanced"}, + }, + { + "payment": "payment-3", + "segCreds": "segment-3", + "state": {"cursor": "later"}, + }, + ] + + async def _post_json( + url: str, + payload: dict[str, object], + *, + headers: dict[str, str] | None = None, + timeout: float = 5.0, + ) -> dict[str, object]: + del url, headers, timeout + calls.append(payload) + return responses[len(calls) - 1] + + post_empty = mock.AsyncMock() + with ( + mock.patch("livepeer_gateway.http.post_json", side_effect=_post_json), + mock.patch("livepeer_gateway.http._post_empty", post_empty), + ): + session = LivePaymentSession( + "https://signer.example.com", + type="live", + challenge=_challenge(), + ) + await session.send_payment() + with pytest.raises(PaymentError, match="no segCreds"): + await session.send_payment() + await session.send_payment() + + assert post_empty.await_count == 2 + assert "state" not in calls[0] + assert calls[1]["state"] == {"cursor": "previous"} + assert calls[2]["state"] == {"cursor": "advanced"} 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 3d542fd..a9c5b1b 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -13,6 +13,7 @@ from livepeer_gateway.errors import ( LivepeerGatewayError, LivepeerHTTPError, + PaymentError, SignerRefreshRequired, ) from livepeer_gateway.live_runner import ( @@ -26,7 +27,7 @@ stop_runner_session, create_proxy, ) -from livepeer_gateway.remote_signer import LivePaymentChallenge +from livepeer_gateway.remote_signer import LivePaymentChallenge, LivePaymentSession class TestLiveRunnerHelpers: @@ -631,6 +632,322 @@ def _request_body( assert payment_attempts == 2 +class TestPaymentSessionCallback: + @pytest.fixture + def payment_call(self, request): + url = "https://runner.example.com/call" + stream = getattr(request, "param", False) + reply = ( + ( + SimpleNamespace(close=mock.AsyncMock()), + SimpleNamespace(status=200, headers={}, release=mock.Mock()), + ) + if stream + else _json_data({"ok": True}) + ) + + def challenge(manifest): + return LivepeerHTTPError(402, url, _payment_challenge_body(manifest)) + + transport = mock.AsyncMock(side_effect=[challenge("manifest-1"), reply]) + signer = mock.AsyncMock( + return_value={"payment": "payment", "segCreds": "segment", "state": {}} + ) + with ( + mock.patch.object( + live_runner, "open_stream" if stream else "_request_body", transport + ), + mock.patch.object( + live_runner, + "get_signer_info", + mock.AsyncMock(return_value=SimpleNamespace(address="payer")), + ), + mock.patch("livepeer_gateway.http.post_json", signer), + ): + yield SimpleNamespace( + kwargs=dict( + runner_url=url, + signer_url="https://signer.example.com", + payment_unit="fixed", + stream=stream, + ), + transport=transport, + signer=signer, + reply=reply, + challenge=challenge, + ) + + @pytest.mark.parametrize( + "payment_call,async_callback,payment_unit", + [(False, False, "fixed"), (True, True, "seconds")], + indirect=["payment_call"], + ) + async def test_reports_each_successful_session_before_dispatch( + self, payment_call, async_callback, payment_unit + ) -> None: + call = payment_call + call.kwargs["payment_unit"] = payment_unit + call.signer.side_effect = [SignerRefreshRequired("refresh")] + [ + call.signer.return_value + ] * 2 + call.transport.side_effect = [ + call.challenge("refused"), + call.challenge("manifest-1"), + call.challenge("manifest-2"), + call.reply, + ] + sessions: list[LivePaymentSession] = [] + + def collect(session): + assert ( + call.signer.await_count + == call.transport.await_count + == len(sessions) + 2 + ) + sessions.append(session) + + callback = mock.AsyncMock(side_effect=collect) if async_callback else collect + result = await call_runner(**call.kwargs, on_payment_session=callback) + if call.kwargs["stream"]: + await result.aclose() + assert [s.manifest_id for s in sessions] == ["manifest-1", "manifest-2"] + assert result.payment_session is ( + sessions[-1] if payment_unit == "seconds" else None + ) + assert call.transport.await_count == 4 + + @pytest.mark.parametrize( + "error", + [ + RuntimeError("failed"), + SignerRefreshRequired("failed"), + LivepeerHTTPError(402, "https://callback.example.com"), + ], + ) + async def test_callback_errors_abort_without_retry( + self, payment_call, error + ) -> None: + callback = mock.AsyncMock(side_effect=error) + with pytest.raises(type(error)) as raised: + await call_runner(**payment_call.kwargs, on_payment_session=callback) + assert raised.value is error + assert raised.value.__context__ is None + assert ( + payment_call.transport.await_count == payment_call.signer.await_count == 1 + ) + + @pytest.mark.parametrize( + "outcome", + [ + LivepeerHTTPError(403, "https://signer.example.com"), + {"segCreds": "segment", "state": {}}, + ], + ) + async def test_signer_failure_does_not_report(self, payment_call, outcome) -> None: + if isinstance(outcome, Exception): + payment_call.signer.side_effect = outcome + else: + payment_call.signer.return_value = outcome + callback = mock.Mock() + with pytest.raises(LivepeerGatewayError): + await call_runner(**payment_call.kwargs, on_payment_session=callback) + callback.assert_not_called() + assert payment_call.transport.await_count == 1 + + @pytest.mark.parametrize( + ("outcome", "error_type", "message"), + [ + ( + {"payment": "payment", "state": {}}, + LivepeerGatewayError, + "Live runner payment response missing segCreds", + ), + ( + {"payment": "payment", "segCreds": None, "state": {}}, + LivepeerGatewayError, + "Live runner payment response missing segCreds", + ), + ( + {"payment": "payment", "segCreds": "", "state": {}}, + LivepeerGatewayError, + "Live runner payment response missing segCreds", + ), + ( + {"payment": "payment", "segCreds": 123, "state": {}}, + PaymentError, + "invalid 'segCreds'", + ), + ( + {"payment": "payment", "segCreds": "segment"}, + PaymentError, + "missing 'state'", + ), + ], + ) + async def test_signed_unusable_response_reports_before_failing( + self, payment_call, outcome, error_type, message + ) -> None: + payment_call.signer.return_value = outcome + callback = mock.Mock() + with pytest.raises(error_type, match=message) as raised: + await call_runner(**payment_call.kwargs, on_payment_session=callback) + assert isinstance(raised.value, error_type) + if error_type is LivepeerGatewayError: + assert type(raised.value) is LivepeerGatewayError + callback.assert_called_once() + assert payment_call.transport.await_count == 1 + + @pytest.mark.parametrize("seg_creds", [None, ""]) + async def test_blank_segment_keeps_the_runner_error_without_a_callback( + self, payment_call, seg_creds + ) -> None: + payment_call.signer.return_value = { + "payment": "payment", + "segCreds": seg_creds, + "state": {"cursor": "advanced"}, + } + with pytest.raises( + LivepeerGatewayError, + match="Live runner payment response missing segCreds", + ) as raised: + await call_runner(**payment_call.kwargs) + assert type(raised.value) is LivepeerGatewayError + assert payment_call.transport.await_count == 1 + + async def test_paid_request_failure_preserves_notification( + self, payment_call + ) -> None: + call = payment_call + error = LivepeerHTTPError(500, "https://runner.example.com") + call.transport.side_effect = [call.challenge("manifest-1"), error] + sessions: list[LivePaymentSession] = [] + with pytest.raises(LivepeerHTTPError) as raised: + await call_runner(**call.kwargs, on_payment_session=sessions.append) + assert raised.value is error + assert [s.manifest_id for s in sessions] == ["manifest-1"] + + @pytest.mark.parametrize("phase", ["signing", "callback"]) + async def test_cancellation_waits_for_reporting(self, payment_call, phase) -> None: + call = payment_call + entered, release = asyncio.Event(), asyncio.Event() + + async def block(*_args, **_kwargs): + entered.set() + await release.wait() + return call.signer.return_value + + callback = mock.AsyncMock(side_effect=block if phase == "callback" else None) + if phase == "signing": + call.signer.side_effect = block + task = asyncio.create_task( + call_runner(**call.kwargs, on_payment_session=callback) + ) + try: + await asyncio.wait_for(entered.wait(), timeout=1) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert callback.await_count == 1 + assert call.transport.await_count == 1 + finally: + release.set() + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def test_second_cancellation_still_waits_for_reporting( + self, payment_call + ) -> None: + entered, release = asyncio.Event(), asyncio.Event() + + async def collect(_session): + entered.set() + await release.wait() + + callback = mock.AsyncMock(side_effect=collect) + task = asyncio.create_task( + call_runner(**payment_call.kwargs, on_payment_session=callback) + ) + try: + await asyncio.wait_for(entered.wait(), timeout=1) + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + callback.assert_awaited_once() + finally: + release.set() + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def test_cancelled_signed_unusable_response_is_reported( + self, payment_call + ) -> None: + entered, release = asyncio.Event(), asyncio.Event() + payment_call.signer.return_value = { + "payment": "payment", + "segCreds": "segment", + } + + async def sign(*_args, **_kwargs): + entered.set() + await release.wait() + return payment_call.signer.return_value + + payment_call.signer.side_effect = sign + callback = mock.Mock() + task = asyncio.create_task( + call_runner(**payment_call.kwargs, on_payment_session=callback) + ) + try: + await asyncio.wait_for(entered.wait(), timeout=1) + task.cancel() + await asyncio.sleep(0) + release.set() + with pytest.raises(asyncio.CancelledError): + await task + callback.assert_called_once() + finally: + release.set() + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def test_callback_failure_wins_over_cancellation(self, payment_call) -> None: + entered, release = asyncio.Event(), asyncio.Event() + error = RuntimeError("failed") + + async def fail(_session): + entered.set() + await release.wait() + raise error + + task = asyncio.create_task( + call_runner( + **payment_call.kwargs, + on_payment_session=fail, + ) + ) + try: + await asyncio.wait_for(entered.wait(), timeout=1) + task.cancel() + await asyncio.sleep(0) + release.set() + with pytest.raises(RuntimeError) as raised: + await task + assert raised.value is error + assert raised.value.__context__ is None + finally: + release.set() + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + def _payment_challenge_body(manifest_id: str) -> str: return json.dumps( {