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

examples/echo/client.py

@@ -5,13 +5,12 @@
 import asyncio
 import sys
 import time
-from contextlib import nullcontext, suppress
+from contextlib import nullcontext
 from pathlib import Path
 
 import av
 
 from livepeer_gateway.errors import LivepeerGatewayError
-from livepeer_gateway.live_runner import stop_runner_session
 from livepeer_gateway.media_output import MediaOutput
 from livepeer_gateway.media_publish import MediaPublish
 from livepeer_gateway.http import post_json
@@ -32,6 +31,7 @@ def _parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Run the proxied echo Live Runner demo.")
     parser.add_argument("input")
     parser.add_argument("--discovery", default=DEFAULT_DISCOVERY)
+    parser.add_argument("--signer-url", default="", help="Remote signer URL; enables on-chain payment (self-funding session).")
     parser.add_argument("--output", default=DEFAULT_OUTPUT)
     parser.add_argument("--radius", type=int, default=75)
     parser.add_argument("--max-frames", type=int, default=0, help="Stop after this many input video frames (0 = full file).")
@@ -120,42 +120,42 @@ async def main() -> None:
     if not input_path.exists():
         raise SystemExit(f"input file does not exist: {input_path}")
 
-    session = None
-
     try:
-        session = await reserve_session(discovery_url=args.discovery, app=ECHO_APP_ID)
-        _log("runner_url:", session.runner.url if session.runner is not None else session.runner_url)
-        _log("session_id:", session.session_id)
-        _log("app_url:", session.app_url)
-
-        echo = await post_json(f"{session.app_url.rstrip('/')}/echo", {"radius": args.radius})
-        in_url = _channel_url(echo, "in")
-        out_url = _channel_url(echo, "out")
-        _log("in:", in_url)
-        _log("out:", out_url)
-
-        with nullcontext(sys.stdout.buffer) if output_stdout else output_path.open("wb") as fh:
-            def _write_chunk(chunk: bytes) -> None:
-                fh.write(chunk)
-                if output_stdout:
-                    fh.flush()
-
-            async with MediaOutput(out_url, on_bytes=_write_chunk):
-                await _publish_video(
-                    input_path,
-                    in_url,
-                    max_frames=max(0, args.max_frames),
-                    app_url=session.app_url,
-                    blur=args.blur,
-                )
-                _log("publish complete; waiting for output to drain...")
-            fh.flush()
+        # The session funds itself while open (auto_pay) and stops the runner
+        # session on exit; offchain (no --signer-url) the payment loop is a no-op.
+        async with await reserve_session(
+            discovery_url=args.discovery,
+            app=ECHO_APP_ID,
+            signer_url=args.signer_url or None,
+        ) as session:
+            _log("runner_url:", session.runner.url if session.runner is not None else session.runner_url)
+            _log("session_id:", session.session_id)
+            _log("app_url:", session.app_url)
+
+            echo = await post_json(f"{session.app_url.rstrip('/')}/echo", {"radius": args.radius})
+            in_url = _channel_url(echo, "in")
+            out_url = _channel_url(echo, "out")
+            _log("in:", in_url)
+            _log("out:", out_url)
+
+            with nullcontext(sys.stdout.buffer) if output_stdout else output_path.open("wb") as fh:
+                def _write_chunk(chunk: bytes) -> None:
+                    fh.write(chunk)
+                    if output_stdout:
+                        fh.flush()
+
+                async with MediaOutput(out_url, on_bytes=_write_chunk):
+                    await _publish_video(
+                        input_path,
+                        in_url,
+                        max_frames=max(0, args.max_frames),
+                        app_url=session.app_url,
+                        blur=args.blur,
+                    )
+                    _log("publish complete; waiting for output to drain...")
+                fh.flush()
     except LivepeerGatewayError as exc:
         raise SystemExit(f"ERROR: {exc}") from exc
-    finally:
-        if session is not None:
-            with suppress(Exception):
-                await stop_runner_session(session)
 
 
 if __name__ == "__main__":

pyproject.toml

@@ -19,6 +19,7 @@ generate-lp-rpc = "livepeer_gateway.codegen:main"
 [project.optional-dependencies]
 dev = [
   "grpcio-tools>=1.65.0",
+  "pytest>=8.0",
 ]
 
 examples = [

src/livepeer_gateway/__init__.py

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

src/livepeer_gateway/http.py

@@ -371,6 +371,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.