Report each signed payment session from call_runner

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",

src/livepeer_gateway/live_runner.py

@@ -87,6 +87,7 @@ class LiveRunnerSessionEvent:
 
 
 LiveRunnerSessionCallback = Callable[[LiveRunnerSessionEvent], None | Awaitable[None]]
+LivePaymentSessionCallback = Callable[[LivePaymentSession], None | Awaitable[None]]
 
 
 @dataclass(frozen=True)
@@ -762,6 +763,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 +781,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 +798,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 +807,13 @@ 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.
 
+    ``on_payment_session`` (sync or async) is called with each payment session the
+    signer signs, before the paid request goes out; if it raises, the request is not
+    sent. The signer charges when it signs, so a reported session can cost even if the
+    call then fails. A runner that answers 402 again after payment gets a new session;
+    metered calls keep paying on the one they have. ``session.manifest_id`` identifies
+    the orchestrator session each payment funds.
+
     ``application/json`` and ``+json`` types parse into ``result.data``; anything else
     (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``.
 
@@ -841,7 +852,10 @@ async def call_runner(
                     signer_url=signer_url or "",
                     signer_headers=signer_headers,
                     max_price=max_price,
+                    on_payment_session=on_payment_session,
                 )
+            except _RunnerPaymentCallbackError as e:
+                raise e.error
             except SignerRefreshRequired as e:
                 if attempt + 1 >= attempts:
                     raise
@@ -852,6 +866,7 @@ async def call_runner(
                 )
                 challenge = None
                 continue
+            _validate_runner_payment(payment)
             request_headers["Livepeer-Payment"] = payment.payment
             request_headers["Livepeer-Segment"] = payment.seg_creds or ""
             session_id = challenge.manifest_id
@@ -995,22 +1010,62 @@ async def _get_runner_payment(
     signer_url: str,
     signer_headers: dict[str, str] | None,
     max_price: LiveRunnerPriceInfo | None,
+    on_payment_session: LivePaymentSessionCallback | None,
     app: str | None = None,
 ) -> tuple[LivePaymentSession, GetPaymentResponse]:
+    session: LivePaymentSession
+
+    async def _report_payment() -> None:
+        if on_payment_session is None:
+            return
+        try:
+            result = on_payment_session(session)
+            if inspect.isawaitable(result):
+                await result
+        except Exception as e:
+            raise _RunnerPaymentCallbackError(e) from e
+
+    callback_args: dict[str, Any] = {}
+    if on_payment_session is not None:
+        callback_args["_on_payment"] = _report_payment
     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,
+        **callback_args,
     )
-    payment = await session.get_payment()
+    payment_request = session.get_payment()
+    if on_payment_session is None:
+        payment = await payment_request
+    else:
+        payment_task = asyncio.create_task(payment_request)
+        try:
+            payment = await asyncio.shield(payment_task)
+        except asyncio.CancelledError:
+            drain = asyncio.gather(payment_task, return_exceptions=True)
+            while not drain.done():
+                try:
+                    await asyncio.shield(drain)
+                except asyncio.CancelledError:
+                    continue
+            raise
+    return session, payment
+
+
+class _RunnerPaymentCallbackError(Exception):
+    def __init__(self, error: Exception) -> None:
+        super().__init__(str(error))
+        self.error = error
+
+
+def _validate_runner_payment(payment: GetPaymentResponse) -> None:
     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(

src/livepeer_gateway/remote_signer.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import logging
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, replace
 from typing import Any, Optional
 
@@ -121,6 +123,7 @@ def __init__(
         app: str | None = None,
         max_price: dict[str, Any] | None = None,
         max_refresh_retries: int = 3,
+        _on_payment: Callable[[], None | Awaitable[None]] | None = None,
     ) -> None:
         self._signer_url = signer_url
         self._signer_headers = _freeze_headers(signer_headers)
@@ -130,6 +133,12 @@ def __init__(
         self._max_price = dict(max_price) if max_price is not None else None
         self._max_refresh_retries = max(0, int(max_refresh_retries))
         self._state: dict[str, Any] | None = None
+        self._on_payment = _on_payment
+
+    @property
+    def manifest_id(self) -> str:
+        """The manifest ID from the 402 challenge this session pays for."""
+        return self._challenge.manifest_id
 
     async def get_payment(self) -> GetPaymentResponse:
         if not self._signer_url:
@@ -218,6 +227,12 @@ async def _payment_request(self) -> GetPaymentResponse:
             raise PaymentError(
                 f"GetPayment error: missing/invalid 'payment' in response (url={url})"
             )
+        on_payment = self._on_payment
+        self._on_payment = None
+        if on_payment is not None:
+            result = on_payment()
+            if inspect.isawaitable(result):
+                await result
 
         seg_creds = data.get("segCreds")
         if seg_creds is not None and not isinstance(seg_creds, str):

tests/test_live_payment_session.py

@@ -105,6 +105,7 @@ async def test_send_payment_preserves_typed_http_error(self) -> None:
 
     async def test_get_payment_sends_opaque_payment_params_and_state(self) -> None:
         calls: list[tuple[str, dict[str, object], dict[str, str] | None]] = []
+        reported: list[str] = []
 
         async def _post_json(
             url: str,
@@ -133,12 +134,14 @@ async def _post_json(
                     "currency": "wei",
                     "unit": "seconds",
                 },
+                _on_payment=lambda: reported.append("signed"),
             )
             first = await session.get_payment()
             second = await session.get_payment()
 
         assert first.payment == "payment"
         assert second.seg_creds == "segment"
+        assert reported == ["signed"]
         assert calls[0][0] == "https://signer.example.com/generate-live-payment"
         assert calls[0][1] == {
             "orchestrator": "opaque-payment-params",
@@ -291,3 +294,45 @@ async def _post_json(
         assert first.address == "opaque-sender"
         assert first.sig == "opaque-signature"
         assert len(calls) == 1
+
+    async def test_manifest_id_comes_from_the_challenge_and_survives_refresh(
+        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 payload, headers, timeout
+            if url == "https://signer.example.com/generate-live-payment":
+                payment_requests += 1
+                if payment_requests == 2:
+                    raise SignerRefreshRequired("refresh")
+                return {
+                    "payment": "payment",
+                    "segCreds": "segment",
+                    "state": {"state": "one"},
+                }
+            if url == "https://signer.example.com/sign-orchestrator-info":
+                return {"address": "opaque-sender", "signature": "opaque-signature"}
+            if url == "https://orch.example.com/refresh-payment":
+                return {"payment_params": "new-payment-params"}
+            raise AssertionError(f"unexpected POST {url}")
+
+        session = LivePaymentSession(
+            "https://signer.example.com",
+            type="fixed",
+            challenge=_challenge(),
+        )
+        assert session.manifest_id == "manifest-1"
+        with mock.patch("livepeer_gateway.http.post_json", side_effect=_post_json):
+            await session.get_payment()
+            await session.get_payment()
+
+        assert session.manifest_id == "manifest-1"
+        assert session._challenge.payment_params == "new-payment-params"