payments: keep the state a skipped cycle hands back

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):

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,

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(

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