feat(live-runner): session-owned payment lifecycle for streamed sessions

src/livepeer_gateway/__init__.py

@@ -56,6 +56,7 @@
     create_trickle_channels,
     register_runner,
     remove_trickle_channels,
+    run_session_payments,
     stop_runner_session,
 )
 from .discovery import discover_orchestrators, discover_runners
@@ -141,6 +142,7 @@
     "orchestrator_selector",
     "runner_selector",
     "reserve_session",
+    "run_session_payments",
     "StartJobRequest",
     "call_runner",
     "create_proxy",

src/livepeer_gateway/http.py

@@ -331,6 +331,50 @@ 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.
+
+    Certificate verification is disabled, matching every other HTTP helper in
+    the SDK. Error responses raise the same typed errors as request_json
+    (SignerRefreshRequired for 480, SkipPaymentCycle for 482, LivepeerHTTPError
+    otherwise) so callers can branch on status codes.
+    """
+    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

@@ -12,11 +12,9 @@
 from typing import Any, Awaitable, Callable, Literal, NotRequired, Optional, Protocol, TypedDict, cast
 from urllib.parse import quote, urlparse, urlunparse
 
-import aiohttp
-
 from .channel_reader import ChannelReader
-from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired
-from .http import post_json, request_json
+from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired, SkipPaymentCycle
+from .http import post_empty, post_json, request_json
 from .remote_signer import (
     GetPaymentResponse,
     LivePaymentSession,
@@ -97,6 +95,114 @@ class LiveRunnerSession:
     app_url: str
     runner_url: str
     runner: Optional[LiveRunnerInstance] = None
+    # Session control base URL from the reserve response. Payments go to
+    # {control_url}/payment, which is scoped to this session and 404s once the
+    # orchestrator releases it. Empty for orchestrators that predate it.
+    control_url: str = ""
+    # Present when the session was reserved on-chain (signer_url given).
+    # reserve_session() starts the payment loop automatically (auto_pay=True);
+    # drive it manually with start_payments() / stop_payments() otherwise.
+    # run_session_payments() is the underlying loop.
+    payment_session: Optional[LivePaymentSession] = field(
+        default=None,
+        repr=False,
+        compare=False,
+    )
+    # Seconds between session payments while the session is held open. Must stay
+    # at or below the orchestrator's livePaymentInterval (5s default) so credits
+    # lead its server-side debit ticker. Default 3s keeps margin under that.
+    payment_interval: float = 3.0
+    # Background payment task, owned by this session once start_payments() runs.
+    _payment_task: Optional[asyncio.Task] = field(
+        default=None,
+        repr=False,
+        compare=False,
+    )
+    # Set once the orchestrator reports the session gone (payment 404).
+    _released: asyncio.Event = field(
+        default_factory=asyncio.Event,
+        repr=False,
+        compare=False,
+    )
+
+    @property
+    def released(self) -> bool:
+        """True once the orchestrator has released this session."""
+        return self._released.is_set()
+
+    async def wait_released(self) -> None:
+        """Block until the orchestrator releases this session.
+
+        Only fires when the payment loop observes the release (payment 404);
+        offchain sessions and sessions without a running payment loop never set
+        it.
+        """
+        await self._released.wait()
+
+    def _mark_released(self) -> None:
+        self._released.set()
+
+    def start_payments(self) -> Optional[asyncio.Task]:
+        """Start the background payment loop that keeps this session funded.
+
+        Idempotent and safe to call repeatedly. No-op offchain (no
+        ``payment_session``). Requires a running event loop; if called from sync
+        code it logs a warning and returns ``None``. Returns the task, or
+        ``None`` if payments could not be started.
+        """
+        if getattr(self, "_payment_task", None) is not None:
+            return self._payment_task
+        if self.payment_session is None:
+            return None
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            _LOG.warning(
+                "No running event loop; session payments not started. "
+                "Call session.start_payments() from async code to enable."
+            )
+            return None
+        task = loop.create_task(run_session_payments(self, interval=self.payment_interval))
+        object.__setattr__(self, "_payment_task", task)
+        return task
+
+    async def stop_payments(self) -> None:
+        """Cancel the payment loop without stopping the session.
+
+        Symmetrical complement to start_payments(); the session stays reserved
+        (and the orchestrator keeps debiting it), so use this only to hand off
+        or drain funding. No-op when no loop is running.
+        """
+        task = getattr(self, "_payment_task", None)
+        object.__setattr__(self, "_payment_task", None)
+        if task is not None and not task.done():
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+
+    async def aclose(self) -> None:
+        """Cancel the payment loop (if running) and stop the runner session.
+
+        Best-effort: both steps run even if one fails, and the first
+        non-cancellation error is re-raised after cleanup. The stop call is
+        skipped when the orchestrator already released the session.
+        """
+        awaitables: list[Awaitable[Any]] = [self.stop_payments()]
+        if not self.released:
+            awaitables.append(stop_runner_session(self))
+        results = await asyncio.gather(*awaitables, return_exceptions=True)
+        for result in results:
+            if isinstance(result, BaseException) and not isinstance(result, asyncio.CancelledError):
+                raise result
+
+    async def __aenter__(self) -> "LiveRunnerSession":
+        self.start_payments()
+        return self
+
+    async def __aexit__(self, *exc_info: object) -> None:
+        await self.aclose()
 
 
 @dataclass(frozen=True)
@@ -116,6 +222,9 @@ class LiveRunnerCallResult:
         repr=False,
         compare=False,
     )
+    # Orchestrator debit cadence in seconds, from the payment challenge's
+    # payment_interval_ms. None when the orchestrator does not report it.
+    server_payment_interval: Optional[float] = None
 
 
 @dataclass(frozen=True)
@@ -243,10 +352,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)
@@ -713,6 +822,9 @@ async def call_runner(
                     or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "")
                 ),
                 payment_session=None if payment_type == "fixed" else payment_session,
+                server_payment_interval=(
+                    challenge.payment_interval_s if challenge is not None else None
+                ),
             )
         except LivepeerHTTPError as e:
             if e.status_code != 402:
@@ -730,6 +842,9 @@ class _RunnerPaymentChallenge:
     payment_params: str
     orchestrator_url: str
     manifest_id: str
+    # Orchestrator debit cadence in seconds (payment_interval_ms); None when
+    # the orchestrator does not report it.
+    payment_interval_s: Optional[float] = None
 
 
 def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentChallenge:
@@ -750,10 +865,16 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC
     if not isinstance(manifest_id, str) or not manifest_id:
         raise LivepeerGatewayError("Live runner payment challenge missing manifest_id")
 
+    interval_ms = data.get("payment_interval_ms")
+    payment_interval_s: Optional[float] = None
+    if isinstance(interval_ms, (int, float)) and not isinstance(interval_ms, bool) and interval_ms > 0:
+        payment_interval_s = float(interval_ms) / 1000.0
+
     return _RunnerPaymentChallenge(
         payment_params=payment_params,
         orchestrator_url=orchestrator_url,
         manifest_id=manifest_id,
+        payment_interval_s=payment_interval_s,
     )
 
 
@@ -840,11 +961,13 @@ def _live_runner_session_from_json(
         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")
+    control_url = data.get("control_url")
     return LiveRunnerSession(
         session_id=session_id.strip(),
         app_url=app_url.strip(),
         runner_url=runner_url,
         runner=runner,
+        control_url=control_url.strip() if isinstance(control_url, str) else "",
     )
 
 async def stop_runner_session(
@@ -871,10 +994,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,
     )
 
 
@@ -1040,25 +1163,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]
@@ -1210,3 +1314,75 @@ def _decode_maybe_bytes(value: object) -> str:
     if isinstance(value, bytes):
         return value.decode("utf-8", errors="replace")
     return str(value or "")
+
+
+async def run_session_payments(
+    session: LiveRunnerSession,
+    *,
+    interval: float = 3.0,
+) -> None:
+    """Keep a reserved live-runner session funded for its whole lifetime.
+
+    After the reservation payment, go-livepeer holds the session as a prepaid
+    balance: a server-side ticker (``-livePaymentInterval``, default 5s) debits it
+    and releases the session once it runs dry. The client must keep crediting it
+    out-of-band, so this pushes payments on a cadence below that server tick. It
+    sends an initial payment immediately, then one every ``interval`` seconds until
+    cancelled.
+
+    The immediate first payment matters: a cold start can leave a long gap after
+    the reservation payment, so top up before the first sleep. The orchestrator
+    can answer a payment with a skip signal (HTTP 482, ``SkipPaymentCycle``) when
+    the balance is still sufficient; that is a normal "paid up" response, not a
+    failure. Other per-cycle failures are logged and retried rather than killing
+    the loop.
+
+    When the session carries a ``control_url``, payments go to the session-scoped
+    ``{control_url}/payment`` endpoint, which knows whether the session still
+    exists. A 404 there means the orchestrator released the session: the loop
+    stops and marks the session released (``session.released`` /
+    ``session.wait_released()``). A 409 means the session is fixed-price and
+    needs no follow-up payments. Without a ``control_url`` (older orchestrators)
+    the generic ``/payment`` endpoint is used, which credits blindly and cannot
+    detect a released session.
+
+    No-op for offchain sessions (no ``payment_session``). ``interval`` must stay
+    at or below the orchestrator's ``livePaymentInterval`` (5s default) so the
+    balance stays ahead; tune it to the deployment.
+    """
+    payment_session = session.payment_session
+    if payment_session is None:
+        return
+    payment_url = _join_endpoint(session.control_url, "payment") if session.control_url else None
+    while True:
+        try:
+            await payment_session.send_payment(payment_url=payment_url)
+        except SkipPaymentCycle as exc:
+            # Orchestrator says the balance is current; this is a healthy gate, not an error.
+            _LOG.debug("Live runner session payment skipped (balance current): %s", exc)
+        except LivepeerHTTPError as exc:
+            if exc.status_code == 404:
+                _LOG.warning(
+                    "Live runner session %s released by orchestrator; stopping payments",
+                    session.session_id,
+                )
+                session._mark_released()
+                return
+            if exc.status_code == 409:
+                _LOG.info(
+                    "Live runner session %s is fixed-price; no follow-up payments needed",
+                    session.session_id,
+                )
+                return
+            if exc.status_code == 403:
+                _LOG.error(
+                    "Live runner session %s payment rejected (mismatched session/payment); "
+                    "stopping payments: %s",
+                    session.session_id,
+                    exc,
+                )
+                return
+            _LOG.warning("Live runner session payment failed: %s", exc)
+        except Exception as exc:  # noqa: BLE001 - keep the loop alive across transient failures
+            _LOG.warning("Live runner session payment failed: %s", exc)
+        await asyncio.sleep(interval)