From 52727018c5f9d980d35a49ee1c62840fe6749d0c Mon Sep 17 00:00:00 2001 From: chrishobcroft Date: Thu, 10 Sep 2026 19:48:02 +0200 Subject: [PATCH] payments: keep the state a skipped cycle hands back A 482 from the remote signer is a completed cycle, not a failed one: a signer with livepeer/go-livepeer#4077 fixed debits the service consumed since its last update and returns new state alongside the error. Dropping it means resending the same balance every cycle, minting nothing, and being released by the orchestrator once the opening credit runs out. Read the state off the 482 body when it is present and store it. Signers that answer with a bare error are unaffected: the caller keeps what it had. --- src/livepeer_gateway/errors.py | 14 +++++- src/livepeer_gateway/http.py | 22 ++++++++- src/livepeer_gateway/remote_signer.py | 13 ++++- tests/test_live_payment_session.py | 71 +++++++++++++++++++++++++++ 4 files changed, 116 insertions(+), 4 deletions(-) diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py index 175ca06..8ac2d83 100644 --- a/src/livepeer_gateway/errors.py +++ b/src/livepeer_gateway/errors.py @@ -44,7 +44,19 @@ class SignerRefreshRequired(LivepeerGatewayError): class SkipPaymentCycle(LivepeerGatewayError): - """Raised when the signer returns HTTP 482 to skip a payment cycle.""" + """Raised when the signer returns HTTP 482 to skip a payment cycle. + + A skipped cycle still ADVANCES the session: the signer debits the service + consumed since its last update and hands back the new state. Carrying that + state here is what lets the caller store it, so the balance runs down and + the next cycle mints when it needs to. Dropping it leaves the caller + replaying the same state forever, which is a session that never pays again + and dies on the orchestrator minutes later. + """ + + def __init__(self, message: str, *, state: dict | None = None) -> None: + super().__init__(message) + self.state = state class PaymentError(LivepeerGatewayError): diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 4558c73..62243f2 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -95,6 +95,22 @@ def _json_request_parts( return resolved_method, req_headers, body +def _state_from_body(body: str) -> dict | None: + """The signed state a 482 carries, if it carries one. + + A skipped payment cycle is not a failed one: the signer has debited the + service consumed since its last update and handed back new state. Older + signers answer with a bare error and no state, so this returns None for + those and the caller keeps what it had. + """ + try: + data = json.loads(body) + except (TypeError, ValueError): + return None + state = data.get("state") if isinstance(data, dict) else None + return state if isinstance(state, dict) else None + + def _raise_http_json_error( status: int, url: str, @@ -108,7 +124,8 @@ def _raise_http_json_error( ) if status == 482: raise SkipPaymentCycle( - f"Signer returned HTTP 482 (skip payment cycle) (url={url}){body_part}" + f"Signer returned HTTP 482 (skip payment cycle) (url={url}){body_part}", + state=_state_from_body(body), ) raise LivepeerHTTPError( status, @@ -166,7 +183,8 @@ def request_json_sync( ) from e if e.code == 482: raise SkipPaymentCycle( - f"Signer returned HTTP 482 (skip payment cycle) (url={url}){body_part}" + f"Signer returned HTTP 482 (skip payment cycle) (url={url}){body_part}", + state=_state_from_body(raw_body), ) from e raise LivepeerHTTPError( e.code, diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 1c027bf..5c31593 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -212,7 +212,18 @@ async def _payment_request(self) -> GetPaymentResponse: payload["state"] = self._state headers = dict(self._signer_headers) if self._signer_headers else None - data = await post_json(url, payload, headers=headers) + try: + data = await post_json(url, payload, headers=headers) + except SkipPaymentCycle as e: + # A SKIPPED CYCLE STILL ADVANCES THE SESSION. This signer keeps no + # state of its own -- the balance and the clock travel with us -- + # so dropping the state it hands back means resending the same + # balance every cycle, minting nothing, and being released by the + # orchestrator for `insufficient balance` once the upfront credit + # runs out. Measured at exactly three minutes. + if e.state is not None: + self._state = e.state + raise payment = data.get("payment") if not isinstance(payment, str) or not payment: raise PaymentError( diff --git a/tests/test_live_payment_session.py b/tests/test_live_payment_session.py index e96af45..4ba3ba1 100644 --- a/tests/test_live_payment_session.py +++ b/tests/test_live_payment_session.py @@ -8,7 +8,9 @@ from livepeer_gateway.errors import ( LivepeerHTTPError, SignerRefreshRequired, + SkipPaymentCycle, ) +from livepeer_gateway.http import _state_from_body from livepeer_gateway.remote_signer import ( LivePaymentChallenge, LivePaymentSession, @@ -291,3 +293,72 @@ async def _post_json( assert first.address == "opaque-sender" assert first.sig == "opaque-signature" assert len(calls) == 1 + + +class TestSkippedCycleState: + """A 482 from the signer carries the advanced state, and the session keeps it. + + The remote signer is stateless: the balance and the clock travel with the + client. If a skipped cycle's state is dropped, the next request replays the + old balance, nothing is ever minted again, and the orchestrator releases the + session once its credit runs out (livepeer/go-livepeer#4077). + """ + + def test_state_from_body(self) -> None: + body = '{"error": {"message": "no tickets"}, "state": {"state": "two", "sig": "s"}}' + assert _state_from_body(body) == {"state": "two", "sig": "s"} + assert _state_from_body('{"error": {"message": "no tickets"}}') is None + assert _state_from_body("no tickets") is None + + async def test_live_session_keeps_skip_state(self) -> None: + calls: list[dict[str, object]] = [] + replies: list[object] = [ + {"payment": "p1", "segCreds": "s1", "state": {"state": "one"}}, + SkipPaymentCycle("skip", state={"state": "two"}), + {"payment": "p3", "segCreds": "s3", "state": {"state": "three"}}, + ] + + async def _post_json(url, payload, *, headers=None, timeout=5.0): + calls.append(dict(payload)) + reply = replies.pop(0) + if isinstance(reply, Exception): + raise reply + return reply + + with mock.patch("livepeer_gateway.http.post_json", side_effect=_post_json): + session = LivePaymentSession( + "https://signer.example.com", type="live", challenge=_challenge() + ) + await session.get_payment() + with pytest.raises(SkipPaymentCycle): + await session.get_payment() + await session.get_payment() + + assert calls[1]["state"] == {"state": "one"} + assert calls[2]["state"] == {"state": "two"} + + async def test_live_session_keeps_state_when_skip_has_none(self) -> None: + calls: list[dict[str, object]] = [] + replies: list[object] = [ + {"payment": "p1", "segCreds": "s1", "state": {"state": "one"}}, + SkipPaymentCycle("skip"), + {"payment": "p3", "segCreds": "s3", "state": {"state": "three"}}, + ] + + async def _post_json(url, payload, *, headers=None, timeout=5.0): + calls.append(dict(payload)) + reply = replies.pop(0) + if isinstance(reply, Exception): + raise reply + return reply + + with mock.patch("livepeer_gateway.http.post_json", side_effect=_post_json): + session = LivePaymentSession( + "https://signer.example.com", type="live", challenge=_challenge() + ) + await session.get_payment() + with pytest.raises(SkipPaymentCycle): + await session.get_payment() + await session.get_payment() + + assert calls[2]["state"] == {"state": "one"}