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..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, @@ -89,6 +89,72 @@ 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 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 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, + ) + + 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) @@ -1051,3 +1117,42 @@ 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. + + 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 + 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/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index dd87d4c..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,8 @@ 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..19f56fe --- /dev/null +++ b/tests/test_live_runner_payments.py @@ -0,0 +1,167 @@ +"""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.errors import SkipPaymentCycle +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_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) + 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())