diff --git a/src/livepeer_gateway/__init__.py b/src/livepeer_gateway/__init__.py index 3a82a4d..a88d68c 100644 --- a/src/livepeer_gateway/__init__.py +++ b/src/livepeer_gateway/__init__.py @@ -15,7 +15,12 @@ wait_for_training, list_capabilities, ) -from .errors import LivepeerGatewayError, NoOrchestratorAvailableError, PaymentError +from .errors import ( + InsufficientBalance, + LivepeerGatewayError, + NoOrchestratorAvailableError, + PaymentError, +) from .events import Events from .media_publish import ( AudioOutputConfig, @@ -62,6 +67,7 @@ "get_orch_info", "LiveVideoToVideo", "LivepeerGatewayError", + "InsufficientBalance", "NoOrchestratorAvailableError", "OrchestratorRejection", "PaymentError", diff --git a/src/livepeer_gateway/byoc.py b/src/livepeer_gateway/byoc.py index 4d27781..6ce93dc 100644 --- a/src/livepeer_gateway/byoc.py +++ b/src/livepeer_gateway/byoc.py @@ -42,7 +42,12 @@ from urllib.request import Request, urlopen from .orchestrator import _http_origin, _parse_http_url, discover_orchestrators -from .errors import LivepeerGatewayError, NoOrchestratorAvailableError, OrchestratorRejection +from .errors import ( + InsufficientBalance, + LivepeerGatewayError, + NoOrchestratorAvailableError, + OrchestratorRejection, +) _LOG = logging.getLogger(__name__) @@ -212,6 +217,10 @@ def _create_byoc_payment( payment_data = json.loads(resp.read()) except HTTPError as e: body = e.read().decode("utf-8", errors="replace")[:200] + if e.code == 483: + raise InsufficientBalance( + f"Signer returned HTTP 483 (insufficient balance): {body}" + ) from e raise LivepeerGatewayError(f"BYOC payment generation failed: HTTP {e.code}: {body}") from e result = {} @@ -394,6 +403,8 @@ def submit_byoc_job( ) headers.update(payment_headers) _LOG.info("BYOC job %s: payment tickets created for %s", job_id, orch_origin) + except InsufficientBalance: + raise except Exception as e: _LOG.warning("BYOC job %s: payment creation failed for %s: %s", job_id, orch_origin, e) rejections.append(OrchestratorRejection(url=orch_origin, reason=f"payment failed: {e}")) @@ -678,6 +689,8 @@ def submit_training_job( headers.update(payment_headers) _LOG.info("Training job %s: payment tickets created for %s", job_id, orch_origin) + except InsufficientBalance: + raise except Exception as e: _LOG.warning("Training job %s: payment creation failed for %s: %s", job_id, orch_origin, e) diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py index 14fb8b4..d6500b1 100644 --- a/src/livepeer_gateway/errors.py +++ b/src/livepeer_gateway/errors.py @@ -31,5 +31,9 @@ class SkipPaymentCycle(LivepeerGatewayError): """Raised when the signer returns HTTP 482 to skip a payment cycle.""" +class InsufficientBalance(LivepeerGatewayError): + """Raised when the signer returns HTTP 483 (insufficient balance / allowance).""" + + class PaymentError(LivepeerGatewayError): """Raised when a PaymentSession operation fails.""" diff --git a/src/livepeer_gateway/lv2v.py b/src/livepeer_gateway/lv2v.py index 5948b03..cda4812 100644 --- a/src/livepeer_gateway/lv2v.py +++ b/src/livepeer_gateway/lv2v.py @@ -11,6 +11,7 @@ from .channel_writer import ChannelWriter from .control import Control, ControlConfig, ControlMode from .errors import ( + InsufficientBalance, LivepeerGatewayError, NoOrchestratorAvailableError, OrchestratorRejection, @@ -380,6 +381,9 @@ def start_lv2v( if mode == ControlMode.MESSAGE and job.control is not None: job.control.start_keepalive() return job + except InsufficientBalance: + # Account-level signer denial — other orch candidates cannot succeed. + raise except LivepeerGatewayError as e: _LOG.debug( "start_lv2v candidate failed, trying fallback if available: %s (%s)", diff --git a/src/livepeer_gateway/orchestrator.py b/src/livepeer_gateway/orchestrator.py index 844cd7b..8577b90 100644 --- a/src/livepeer_gateway/orchestrator.py +++ b/src/livepeer_gateway/orchestrator.py @@ -13,6 +13,7 @@ from .capabilities import capabilities_to_query from .errors import ( + InsufficientBalance, LivepeerGatewayError, SignerRefreshRequired, SkipPaymentCycle, @@ -116,6 +117,10 @@ def request_json( raise SkipPaymentCycle( f"Signer returned HTTP 482 (skip payment cycle) (url={url}){body_part}" ) from e + if e.code == 483: + raise InsufficientBalance( + f"Signer returned HTTP 483 (insufficient balance) (url={url}){body_part}" + ) from e raise LivepeerGatewayError( f"HTTP JSON error: HTTP {e.code} from endpoint (url={url}){body_part}" ) from e diff --git a/tests/test_insufficient_balance.py b/tests/test_insufficient_balance.py new file mode 100644 index 0000000..101a732 --- /dev/null +++ b/tests/test_insufficient_balance.py @@ -0,0 +1,93 @@ +"""HTTP 483 (insufficient balance) must fail fast — no orch fallback.""" +from __future__ import annotations + +from io import BytesIO +from unittest.mock import MagicMock, patch +from urllib.error import HTTPError + +import pytest + +from livepeer_gateway.byoc import _create_byoc_payment +from livepeer_gateway.errors import InsufficientBalance, NoOrchestratorAvailableError +from livepeer_gateway.lv2v import StartJobRequest, start_lv2v +from livepeer_gateway.orchestrator import request_json + + +def test_request_json_maps_483_to_insufficient_balance() -> None: + body = b'{"error":"Starter allowance exhausted"}' + err = HTTPError( + "https://signer.example/generate-live-payment", + 483, + "Insufficient Balance", + hdrs=None, # type: ignore[arg-type] + fp=BytesIO(body), + ) + with patch("livepeer_gateway.orchestrator.urlopen", side_effect=err): + with pytest.raises(InsufficientBalance, match="Starter allowance exhausted"): + request_json( + "https://signer.example/generate-live-payment", + payload={}, + ) + + +def test_create_byoc_payment_preserves_483_as_insufficient_balance() -> None: + info = MagicMock() + info.HasField.return_value = True + info.ticket_params.face_value = b"\x01" + info.SerializeToString.return_value = b"orchestrator-info" + err = HTTPError( + "https://signer.example/generate-live-payment", + 483, + "Insufficient Balance", + hdrs=None, # type: ignore[arg-type] + fp=BytesIO(b"Starter allowance exhausted"), + ) + + with ( + patch("livepeer_gateway.orch_info.get_orch_info", return_value=info), + patch("livepeer_gateway.byoc.urlopen", side_effect=err), + ): + with pytest.raises(InsufficientBalance, match="Starter allowance exhausted"): + _create_byoc_payment( + orch_origin="https://orch.example:8936", + capability="noop", + livepeer_hdr="header", + signer_url="https://signer.example", + ) + + +def test_start_lv2v_fails_fast_on_483_without_orch_fallback() -> None: + info_a = MagicMock() + info_a.transcoder = "https://orch-a.example:8935" + info_b = MagicMock() + info_b.transcoder = "https://orch-b.example:8935" + + cursor = MagicMock() + cursor.next.side_effect = [ + ("https://orch-a.example:8935", info_a), + ("https://orch-b.example:8935", info_b), + NoOrchestratorAvailableError("exhausted", rejections=[]), + ] + + payment_session = MagicMock() + payment_session.get_payment.side_effect = InsufficientBalance( + "Signer returned HTTP 483 (insufficient balance) " + "(url=https://signer.example/generate-live-payment); " + "body='Starter allowance exhausted'" + ) + + with ( + patch("livepeer_gateway.lv2v.orchestrator_selector", return_value=cursor), + patch("livepeer_gateway.lv2v.PaymentSession", return_value=payment_session), + patch("livepeer_gateway.lv2v.build_capabilities"), + ): + with pytest.raises(InsufficientBalance, match="Starter allowance exhausted"): + start_lv2v( + None, + StartJobRequest(model_id="noop"), + signer_url="https://signer.example", + discovery_url="https://discovery.example/raw", + ) + + assert cursor.next.call_count == 1 + assert payment_session.get_payment.call_count == 1