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"}