From 3b60852f3a2906f6cd234d03ee787aaa3dc84780 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Mon, 15 Jun 2026 20:40:41 +0200 Subject: [PATCH 1/3] feat(live-runner): surface payment session and add run_session_payments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reserve_session now returns the LivePaymentSession built during the reserve payment challenge (previously discarded). Add run_session_payments(session), a timer-driven loop that calls LivePaymentSession.send_payment on an interval to keep a long-lived session funded — the orchestrator meters open sessions by wall-clock time and releases them when the balance runs dry. No-op offchain. Reusable across any held-open transport (trickle today, websockets next). Co-Authored-By: Claude Opus 4.8 --- src/livepeer_gateway/__init__.py | 2 ++ src/livepeer_gateway/live_runner.py | 31 +++++++++++++++++++++++++++++ src/livepeer_gateway/selection.py | 1 + 3 files changed, 34 insertions(+) diff --git a/src/livepeer_gateway/__init__.py b/src/livepeer_gateway/__init__.py index 3baf9f5..3ab982e 100644 --- a/src/livepeer_gateway/__init__.py +++ b/src/livepeer_gateway/__init__.py @@ -54,6 +54,7 @@ create_trickle_channels, register_runner, remove_trickle_channels, + run_session_payments, stop_runner_session, ) from .discovery import discover_orchestrators, discover_runners @@ -138,6 +139,7 @@ "orchestrator_selector", "runner_selector", "reserve_session", + "run_session_payments", "StartJobRequest", "call_runner", "create_trickle_channels", diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 5a38e2e..373c751 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -89,6 +89,13 @@ class LiveRunnerSession: app_url: str runner_url: str runner: Optional[LiveRunnerInstance] = None + # Present when the session was reserved on-chain (signer_url given). Drive it + # with run_session_payments() to keep a long-lived session funded. + payment_session: Optional[LivePaymentSession] = field( + default=None, + repr=False, + compare=False, + ) @dataclass(frozen=True) @@ -1051,3 +1058,27 @@ 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. + + The orchestrator meters an open session by wall-clock time and releases it + when the balance runs dry, so any held-open transport (trickle, websocket) + must keep paying. This sends one payment every ``interval`` seconds until + cancelled. + + No-op for offchain sessions (no ``payment_session``). ``interval`` must stay + at or below the orchestrator's payment interval so the balance stays ahead; + tune it to the deployment. + """ + payment_session = session.payment_session + if payment_session is None: + return + while True: + await asyncio.sleep(interval) + await payment_session.send_payment() diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index dd87d4c..c432192 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -296,6 +296,7 @@ async def reserve_session( app_url=app_url.strip(), runner_url=result.runner_url, runner=result.runner, + payment_session=result.payment_session, ) From 37ee769012d5bcab9af4823ef203ddbafd656293 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 17 Jun 2026 16:11:57 +0200 Subject: [PATCH 2/3] feat(live-runner): make session own its payment loop lifecycle Add a session-owned payment lifecycle to LiveRunnerSession so streamed (trickle/websocket) sessions stay funded without the caller hand-managing a background task. Mirrors the lv2v Lv2vJob.start_payment_sender/close shape, but uses the transport-agnostic interval driver (run_session_payments) since the general live-runner path has no single output stream to meter and must also cover websocket. - LiveRunnerSession (still frozen) gains start_payments(), aclose(), and async context manager support; _payment_task stored via object.__setattr__. start_payments is idempotent, a no-op offchain, and warns instead of raising when called without a running loop. - run_session_payments now pays immediately before the first sleep (a cold start can leave a long gap after the reservation payment) and logs+retries per-cycle failures instead of dying. - reserve_session threads payment_interval through to the session. Callers can now do `async with await reserve_session(...) as session:` and get automatic start/stop, or use start_payments()/aclose() manually. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/livepeer_gateway/live_runner.py | 73 +++++++++++++- src/livepeer_gateway/selection.py | 2 + tests/test_live_runner_payments.py | 142 ++++++++++++++++++++++++++++ 3 files changed, 213 insertions(+), 4 deletions(-) create mode 100644 tests/test_live_runner_payments.py diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 373c751..2f3f672 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -90,12 +90,70 @@ class LiveRunnerSession: runner_url: str runner: Optional[LiveRunnerInstance] = None # Present when the session was reserved on-chain (signer_url given). Drive it - # with run_session_payments() to keep a long-lived session funded. + # with start_payments() / the async context manager to keep a long-lived + # session funded; 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 payment interval so the balance leads. + 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, + ) + + 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 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. + """ + awaitables: list[Awaitable[Any]] = [] + task = getattr(self, "_payment_task", None) + if task is not None and not task.done(): + task.cancel() + awaitables.append(task) + 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) @@ -1069,8 +1127,12 @@ async def run_session_payments( The orchestrator meters an open session by wall-clock time and releases it when the balance runs dry, so any held-open transport (trickle, websocket) - must keep paying. This sends one payment every ``interval`` seconds until - cancelled. + must keep paying. This 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. Per-cycle failures + are logged and retried rather than killing the loop. No-op for offchain sessions (no ``payment_session``). ``interval`` must stay at or below the orchestrator's payment interval so the balance stays ahead; @@ -1080,5 +1142,8 @@ async def run_session_payments( if payment_session is None: return while True: + try: + await payment_session.send_payment() + 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) - await payment_session.send_payment() diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index c432192..6ec9de0 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -274,6 +274,7 @@ async def reserve_session( app: Optional[FilterValue] = None, gpu: Optional[FilterValue] = None, timeout: float = 5.0, + payment_interval: float = 3.0, ) -> LiveRunnerSession: cursor = await runner_selector( signer_url=signer_url, @@ -296,6 +297,7 @@ async def reserve_session( app_url=app_url.strip(), runner_url=result.runner_url, runner=result.runner, + payment_interval=payment_interval, payment_session=result.payment_session, ) diff --git a/tests/test_live_runner_payments.py b/tests/test_live_runner_payments.py new file mode 100644 index 0000000..78ff7fb --- /dev/null +++ b/tests/test_live_runner_payments.py @@ -0,0 +1,142 @@ +"""Unit tests for live-runner session payment lifecycle. + +Covers the session-owned payment loop added on top of run_session_payments: +- run_session_payments pays immediately, then on interval, and is a no-op offchain +- LiveRunnerSession.start_payments is idempotent, offchain-safe, loop-aware +- aclose / async-context-manager cancel the loop and stop the session +""" +from __future__ import annotations + +import asyncio +from unittest.mock import AsyncMock, patch + +import pytest + +from livepeer_gateway import live_runner +from livepeer_gateway.live_runner import LiveRunnerSession, run_session_payments + + +class _FakePaymentSession: + """Duck-typed stand-in for LivePaymentSession (only send_payment is used).""" + + def __init__(self) -> None: + self.calls = 0 + self.paid = asyncio.Event() + + async def send_payment(self, orchestrator_url: str | None = None) -> None: + self.calls += 1 + self.paid.set() + + +def _session(payment_session=None, interval: float = 10.0) -> LiveRunnerSession: + return LiveRunnerSession( + session_id="sess-1", + app_url="http://app", + runner_url="http://runner", + payment_session=payment_session, + payment_interval=interval, + ) + + +def test_run_session_payments_noop_offchain() -> None: + async def go() -> None: + # Returns immediately when there is no payment_session. + await asyncio.wait_for(run_session_payments(_session(None), interval=0.01), timeout=1.0) + + asyncio.run(go()) + + +def test_run_session_payments_pays_immediately() -> None: + async def go() -> None: + ps = _FakePaymentSession() + # Long interval: only the immediate first payment should land before cancel. + task = asyncio.create_task(run_session_payments(_session(ps), interval=10.0)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert ps.calls >= 1 + + asyncio.run(go()) + + +def test_run_session_payments_survives_payment_error() -> None: + async def go() -> None: + ps = _FakePaymentSession() + original = ps.send_payment + attempts = {"n": 0} + + async def flaky(orchestrator_url: str | None = None) -> None: + attempts["n"] += 1 + if attempts["n"] == 1: + raise RuntimeError("transient signer error") + await original(orchestrator_url) + + ps.send_payment = flaky # type: ignore[assignment] + task = asyncio.create_task(run_session_payments(_session(ps), interval=0.01)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) # set on the 2nd, successful cycle + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert attempts["n"] >= 2 + + asyncio.run(go()) + + +def test_start_payments_noop_offchain() -> None: + async def go() -> None: + sess = _session(None) + assert sess.start_payments() is None + assert sess._payment_task is None + + asyncio.run(go()) + + +def test_start_payments_without_running_loop_returns_none() -> None: + # No running loop: logs a warning and skips rather than raising. + sess = _session(_FakePaymentSession()) + assert sess.start_payments() is None + assert sess._payment_task is None + + +def test_start_payments_is_idempotent() -> None: + async def go() -> None: + sess = _session(_FakePaymentSession()) + t1 = sess.start_payments() + t2 = sess.start_payments() + assert t1 is not None + assert t1 is t2 + t1.cancel() + with pytest.raises(asyncio.CancelledError): + await t1 + + asyncio.run(go()) + + +def test_aclose_cancels_loop_and_stops_session() -> None: + async def go() -> None: + sess = _session(_FakePaymentSession()) + sess.start_payments() + task = sess._payment_task + assert task is not None + with patch.object(live_runner, "stop_runner_session", new=AsyncMock()) as stop: + await sess.aclose() + stop.assert_awaited_once() + assert task.done() + + asyncio.run(go()) + + +def test_async_context_manager_starts_and_stops() -> None: + async def go() -> None: + ps = _FakePaymentSession() + sess = _session(ps) + with patch.object(live_runner, "stop_runner_session", new=AsyncMock()) as stop: + async with sess as entered: + assert entered is sess + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) + assert sess._payment_task is not None + stop.assert_awaited_once() + assert sess._payment_task.done() + + asyncio.run(go()) From 81ee7bee7932afe94c882a560e5a8bdba1f2cb04 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 17 Jun 2026 17:10:15 +0200 Subject: [PATCH 3/3] refactor(live-runner): handle skip-payment gate and anchor interval to server tick Refinements from reviewing the go-livepeer orchestrator design (ai_http.go ReserveLiveRunnerSession): after the reservation 402, the orchestrator holds the session as a prepaid balance debited by a server-side ticker (-livePaymentInterval, 5s default) and silently releases it when underfunded. The client just keeps crediting out-of-band, which is what run_session_payments already does. Two corrections: - Treat HTTP 482 / SkipPaymentCycle as a healthy "balance current" gate (debug, keep looping) instead of logging it as a payment failure. The orchestrator uses it to prevent overpayment; only genuine errors warn. - Document that payment_interval must stay at or below the orchestrator's livePaymentInterval (5s), which is why the 3s default carries margin. Add a test asserting a skip cycle does not kill the loop. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/livepeer_gateway/live_runner.py | 29 +++++++++++++++++++---------- tests/test_live_runner_payments.py | 25 +++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 10 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 2f3f672..bfb45a6 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -15,7 +15,7 @@ import aiohttp from .channel_reader import ChannelReader -from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired +from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired, SkipPaymentCycle from .http import post_json, request_json from .remote_signer import ( GetPaymentResponse, @@ -98,7 +98,8 @@ class LiveRunnerSession: compare=False, ) # Seconds between session payments while the session is held open. Must stay - # at or below the orchestrator's payment interval so the balance leads. + # 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( @@ -1125,18 +1126,23 @@ async def run_session_payments( ) -> None: """Keep a reserved live-runner session funded for its whole lifetime. - The orchestrator meters an open session by wall-clock time and releases it - when the balance runs dry, so any held-open transport (trickle, websocket) - must keep paying. This sends an initial payment immediately, then one every - ``interval`` seconds until cancelled. + 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. Per-cycle failures - are logged and retried rather than killing the loop. + 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. No-op for offchain sessions (no ``payment_session``). ``interval`` must stay - at or below the orchestrator's payment interval so the balance stays ahead; - tune it to the deployment. + 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: @@ -1144,6 +1150,9 @@ async def run_session_payments( while True: try: await payment_session.send_payment() + 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 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) diff --git a/tests/test_live_runner_payments.py b/tests/test_live_runner_payments.py index 78ff7fb..19f56fe 100644 --- a/tests/test_live_runner_payments.py +++ b/tests/test_live_runner_payments.py @@ -13,6 +13,7 @@ import pytest from livepeer_gateway import live_runner +from livepeer_gateway.errors import SkipPaymentCycle from livepeer_gateway.live_runner import LiveRunnerSession, run_session_payments @@ -83,6 +84,30 @@ async def flaky(orchestrator_url: str | None = None) -> None: asyncio.run(go()) +def test_run_session_payments_treats_skip_cycle_as_paid_up() -> None: + async def go() -> None: + ps = _FakePaymentSession() + original = ps.send_payment + attempts = {"n": 0} + + async def skip_then_pay(orchestrator_url: str | None = None) -> None: + attempts["n"] += 1 + if attempts["n"] == 1: + raise SkipPaymentCycle("HTTP 482 (skip payment cycle)") # orchestrator: balance current + await original(orchestrator_url) + + ps.send_payment = skip_then_pay # type: ignore[assignment] + task = asyncio.create_task(run_session_payments(_session(ps), interval=0.01)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) # set on the 2nd cycle, after the skip + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + # The skip did not kill the loop; it kept going and paid on the next cycle. + assert attempts["n"] >= 2 + + asyncio.run(go()) + + def test_start_payments_noop_offchain() -> None: async def go() -> None: sess = _session(None)