feat(live-runner): keep metered sessions funded (single-shot, streams, reserved sessions)

src/livepeer_gateway/http.py

@@ -371,6 +371,45 @@ 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.
+    """
+    try:
+        client_timeout = aiohttp.ClientTimeout(total=timeout)
+        connector = aiohttp.TCPConnector(ssl=False)
+        async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session:
+            async with session.post(url, data=b"", headers=headers) as resp:
+                if resp.status >= 400:
+                    raw = await resp.text()
+                    _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items()))
+                await resp.read()
+    except (SignerRefreshRequired, SkipPaymentCycle, LivepeerGatewayError):
+        raise
+    except ConnectionRefusedError as e:
+        raise LivepeerGatewayError(
+            f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})"
+        ) from e
+    except getattr(aiohttp, "ClientConnectorError", ()) as e:
+        os_error = getattr(e, "os_error", None)
+        if isinstance(os_error, ConnectionRefusedError):
+            raise LivepeerGatewayError(
+                f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})"
+            ) from e
+        raise LivepeerGatewayError(
+            f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})"
+        ) from e
+    except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+        raise LivepeerGatewayError(
+            f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})"
+        ) from e
+
+
 def _parse_http_url(url: str, *, context: str = "URL") -> ParseResult:
     """
     Normalize a URL for HTTP(S) endpoints.

src/livepeer_gateway/live_runner.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import inspect
 import json
 import logging
@@ -29,7 +30,7 @@
 
 from .channel_reader import ChannelReader
 from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired
-from .http import open_stream, post_json, request_json
+from .http import open_stream, post_empty, post_json, request_json
 from .remote_signer import (
     GetPaymentResponse,
     LivePaymentSession,
@@ -49,6 +50,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<value>[0-9]+(?:\.[0-9]+)?)(?P<unit>ns|us|\u00b5s|ms|s|m|h)\s*$")
@@ -104,12 +108,57 @@ class LiveRunnerInstance:
     price_info: Optional[LiveRunnerPriceInfo] = 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: Optional[LiveRunnerInstance] = None
+    control_url: str = ""
+    # True once the orchestrator reported this session gone.
+    released: bool = False
+    _payment_task: Optional[asyncio.Task[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 _payment_endpoint(self.control_url)
+
+    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),
+        )
+
+    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)
@@ -147,6 +196,11 @@ class LiveRunnerCallStream:
     payment_session: Optional[LivePaymentSession]
     _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
+    _payment_task: Optional[asyncio.Task[None]] = field(
+        default=None, repr=False, compare=False
+    )
 
     @property
     def content_type(self) -> str:
@@ -160,7 +214,22 @@ 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: 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()
 
@@ -296,10 +365,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)
@@ -755,6 +824,8 @@ async def call_runner(
         payment_session: Optional[LivePaymentSession] = None
         payment_type = ""
         session_id = ""
+        needs_ongoing_funding = False
+        payment_url = ""
         request_headers: dict[str, str] = {}
         if signer_url:
             request_headers[_LIVE_RUNNER_PAYER_ADDRESS_HEADER] = payer_address
@@ -782,6 +853,17 @@ async def call_runner(
             request_headers["Livepeer-Segment"] = payment.seg_creds or ""
             session_id = challenge.manifest_id
 
+            # 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}
             if request_headers:
@@ -796,16 +878,37 @@ 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(
+                    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.
+                if needs_ongoing_funding:
+                    call_stream._start_payments(
+                        cast(LivePaymentSession, payment_session), payment_url
+                    )
+                return call_stream
 
-            data = await request_json(
-                runner_url,
-                method=method,
-                payload=request_payload,
-                **request_kwargs,
+            # The request ends with this call, so the funding ends with it.
+            pay_task = (
+                _start_funding(cast(LivePaymentSession, payment_session), payment_url)
+                if needs_ongoing_funding
+                else None
             )
+            try:
+                data = await request_json(
+                    runner_url,
+                    method=method,
+                    payload=request_payload,
+                    **request_kwargs,
+                )
+            finally:
+                await _stop_funding(pay_task)
             if not isinstance(data, dict):
                 raise LivepeerGatewayError(
                     f"Live runner call expected JSON object, got {type(data).__name__}"
@@ -863,6 +966,56 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC
     )
 
 
+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. 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='')}",
+        )
+    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,
+    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,
     *,
@@ -934,25 +1087,6 @@ def _live_runner_price_info_from_json(value: object) -> Optional[LiveRunnerPrice
     )
 
 
-def _live_runner_session_from_json(
-    data: dict[str, Any],
-    *,
-    runner_url: str,
-    runner: Optional[LiveRunnerInstance],
-) -> 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,
     *,
@@ -977,10 +1111,10 @@ async def stop_runner_session(
         url = _join_endpoint(control_url, "stop")
         if isinstance(token, str) and token.strip():
             request_headers = {"Livepeer-Session-Token": token}
-    await _post_empty(
+    await post_empty(
         url,
-        request_headers,
-        timeout,
+        headers=request_headers,
+        timeout=timeout,
     )
 
 
@@ -1146,25 +1280,6 @@ def _is_trickle_channel_response(value: object) -> bool:
     ) and ("internal_url" not in value or isinstance(value.get("internal_url"), str))
 
 
-async def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None:
-    try:
-        client_timeout = aiohttp.ClientTimeout(total=timeout)
-        connector = aiohttp.TCPConnector(ssl=False)
-        async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session:
-            async with session.post(url, data=b"", headers=headers) as resp:
-                body = await resp.text()
-                if resp.status >= 400:
-                    raise LivepeerGatewayError(
-                        f"HTTP empty POST error: HTTP {resp.status}; body={body!r}"
-                    )
-    except LivepeerGatewayError:
-        raise
-    except getattr(aiohttp, "ClientConnectorError", ()) as e:
-        raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e
-    except (aiohttp.ClientError, asyncio.TimeoutError) as e:
-        raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e
-
-
 def _detect_gpu_pynvml() -> Optional[LiveRunnerGPU]:
     try:
         import pynvml  # type: ignore[import-not-found]