From ba5cb4c4bf1c8bff981df7e5e6a53b8d6ec07016 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Tue, 14 Jul 2026 20:39:46 -0400 Subject: [PATCH 1/3] fix: fail fast on signer HTTP 483 insufficient balance Map signer 483 to InsufficientBalance and skip orchestrator fallback for LV2V/BYOC payment failures. Also add package LOG_LEVEL configure_logging. --- examples/write_frames.py | 3 +- src/livepeer_gateway/__init__.py | 15 +++++- src/livepeer_gateway/byoc.py | 11 ++++- src/livepeer_gateway/errors.py | 4 ++ src/livepeer_gateway/logging_config.py | 33 +++++++++++++ src/livepeer_gateway/lv2v.py | 4 ++ src/livepeer_gateway/orchestrator.py | 5 ++ tests/test_insufficient_balance.py | 66 ++++++++++++++++++++++++++ 8 files changed, 137 insertions(+), 4 deletions(-) create mode 100644 src/livepeer_gateway/logging_config.py create mode 100644 tests/test_insufficient_balance.py diff --git a/examples/write_frames.py b/examples/write_frames.py index 05b33ef..5185986 100644 --- a/examples/write_frames.py +++ b/examples/write_frames.py @@ -4,6 +4,7 @@ import av +from livepeer_gateway import configure_logging from livepeer_gateway.errors import LivepeerGatewayError from livepeer_gateway.lv2v import StartJobRequest, start_lv2v from livepeer_gateway.media_publish import MediaPublishConfig, VideoOutputConfig @@ -49,6 +50,7 @@ def _solid_rgb_frame(width: int, height: int, rgb: tuple[int, int, int]) -> av.V async def main() -> None: + configure_logging() args = _parse_args() frame_interval = 1.0 / max(1e-6, args.fps) @@ -88,4 +90,3 @@ async def main() -> None: if __name__ == "__main__": asyncio.run(main()) - diff --git a/src/livepeer_gateway/__init__.py b/src/livepeer_gateway/__init__.py index 3a82a4d..e8060e3 100644 --- a/src/livepeer_gateway/__init__.py +++ b/src/livepeer_gateway/__init__.py @@ -15,7 +15,14 @@ wait_for_training, list_capabilities, ) -from .errors import LivepeerGatewayError, NoOrchestratorAvailableError, PaymentError +from .errors import ( + InsufficientBalance, + LivepeerGatewayError, + NoOrchestratorAvailableError, + OrchestratorRejection, + PaymentError, +) +from .logging_config import apply_package_log_level, configure_logging from .events import Events from .media_publish import ( AudioOutputConfig, @@ -33,7 +40,6 @@ VideoDecodedMediaFrame, ) from .media_output import MediaOutput, MediaOutputStats -from .errors import OrchestratorRejection from .lv2v import LiveVideoToVideo, StartJobRequest, start_lv2v from .orch_info import get_orch_info from .orchestrator import discover_orchestrators @@ -62,9 +68,11 @@ "get_orch_info", "LiveVideoToVideo", "LivepeerGatewayError", + "InsufficientBalance", "NoOrchestratorAvailableError", "OrchestratorRejection", "PaymentError", + "configure_logging", "MediaPublish", "MediaPublishConfig", "MediaPublishTrack", @@ -99,3 +107,6 @@ "TrickleSubscriberStats", "VideoDecodedMediaFrame", ] + +# Apply LOG_LEVEL to the package logger on import (no basicConfig). +apply_package_log_level() diff --git a/src/livepeer_gateway/byoc.py b/src/livepeer_gateway/byoc.py index 4d27781..4701305 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__) @@ -394,6 +399,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 +685,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/logging_config.py b/src/livepeer_gateway/logging_config.py new file mode 100644 index 0000000..3ddc459 --- /dev/null +++ b/src/livepeer_gateway/logging_config.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import logging +import os + + +_PACKAGE_LOGGER = "livepeer_gateway" +_DEFAULT_LEVEL = logging.INFO +_LOG_FORMAT = "%(levelname)s %(name)s: %(message)s" + + +def _resolve_level(level: str | int | None = None) -> int: + if level is None: + level = os.environ.get("LOG_LEVEL", "INFO") + if isinstance(level, int): + return level + return getattr(logging, str(level).upper(), _DEFAULT_LEVEL) + + +def apply_package_log_level(level: str | int | None = None) -> None: + """Set the livepeer_gateway logger level from arg or LOG_LEVEL env (default INFO).""" + logging.getLogger(_PACKAGE_LOGGER).setLevel(_resolve_level(level)) + + +def configure_logging(level: str | int | None = None) -> None: + """Configure livepeer_gateway log level from arg or LOG_LEVEL env (default INFO). + + Ensures a basic root handler exists so package logs are visible. + """ + resolved = _resolve_level(level) + apply_package_log_level(resolved) + if not logging.getLogger().handlers: + logging.basicConfig(level=resolved, format=_LOG_FORMAT) 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..64135f2 --- /dev/null +++ b/tests/test_insufficient_balance.py @@ -0,0 +1,66 @@ +"""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.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_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 From a903ab9fccec7ea9de0e8c058a215f5abc4a7bdb Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Tue, 14 Jul 2026 20:42:04 -0400 Subject: [PATCH 2/3] refactor: drop logging changes from insufficient-balance PR Keep this branch scoped to HTTP 483 / InsufficientBalance fail-fast only. --- examples/write_frames.py | 3 +-- src/livepeer_gateway/__init__.py | 7 +----- src/livepeer_gateway/logging_config.py | 33 -------------------------- 3 files changed, 2 insertions(+), 41 deletions(-) delete mode 100644 src/livepeer_gateway/logging_config.py diff --git a/examples/write_frames.py b/examples/write_frames.py index 5185986..05b33ef 100644 --- a/examples/write_frames.py +++ b/examples/write_frames.py @@ -4,7 +4,6 @@ import av -from livepeer_gateway import configure_logging from livepeer_gateway.errors import LivepeerGatewayError from livepeer_gateway.lv2v import StartJobRequest, start_lv2v from livepeer_gateway.media_publish import MediaPublishConfig, VideoOutputConfig @@ -50,7 +49,6 @@ def _solid_rgb_frame(width: int, height: int, rgb: tuple[int, int, int]) -> av.V async def main() -> None: - configure_logging() args = _parse_args() frame_interval = 1.0 / max(1e-6, args.fps) @@ -90,3 +88,4 @@ async def main() -> None: if __name__ == "__main__": asyncio.run(main()) + diff --git a/src/livepeer_gateway/__init__.py b/src/livepeer_gateway/__init__.py index e8060e3..a88d68c 100644 --- a/src/livepeer_gateway/__init__.py +++ b/src/livepeer_gateway/__init__.py @@ -19,10 +19,8 @@ InsufficientBalance, LivepeerGatewayError, NoOrchestratorAvailableError, - OrchestratorRejection, PaymentError, ) -from .logging_config import apply_package_log_level, configure_logging from .events import Events from .media_publish import ( AudioOutputConfig, @@ -40,6 +38,7 @@ VideoDecodedMediaFrame, ) from .media_output import MediaOutput, MediaOutputStats +from .errors import OrchestratorRejection from .lv2v import LiveVideoToVideo, StartJobRequest, start_lv2v from .orch_info import get_orch_info from .orchestrator import discover_orchestrators @@ -72,7 +71,6 @@ "NoOrchestratorAvailableError", "OrchestratorRejection", "PaymentError", - "configure_logging", "MediaPublish", "MediaPublishConfig", "MediaPublishTrack", @@ -107,6 +105,3 @@ "TrickleSubscriberStats", "VideoDecodedMediaFrame", ] - -# Apply LOG_LEVEL to the package logger on import (no basicConfig). -apply_package_log_level() diff --git a/src/livepeer_gateway/logging_config.py b/src/livepeer_gateway/logging_config.py deleted file mode 100644 index 3ddc459..0000000 --- a/src/livepeer_gateway/logging_config.py +++ /dev/null @@ -1,33 +0,0 @@ -from __future__ import annotations - -import logging -import os - - -_PACKAGE_LOGGER = "livepeer_gateway" -_DEFAULT_LEVEL = logging.INFO -_LOG_FORMAT = "%(levelname)s %(name)s: %(message)s" - - -def _resolve_level(level: str | int | None = None) -> int: - if level is None: - level = os.environ.get("LOG_LEVEL", "INFO") - if isinstance(level, int): - return level - return getattr(logging, str(level).upper(), _DEFAULT_LEVEL) - - -def apply_package_log_level(level: str | int | None = None) -> None: - """Set the livepeer_gateway logger level from arg or LOG_LEVEL env (default INFO).""" - logging.getLogger(_PACKAGE_LOGGER).setLevel(_resolve_level(level)) - - -def configure_logging(level: str | int | None = None) -> None: - """Configure livepeer_gateway log level from arg or LOG_LEVEL env (default INFO). - - Ensures a basic root handler exists so package logs are visible. - """ - resolved = _resolve_level(level) - apply_package_log_level(resolved) - if not logging.getLogger().handlers: - logging.basicConfig(level=resolved, format=_LOG_FORMAT) From 01e9affa00f54d8636a326df1109a6d62b9676a8 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Tue, 14 Jul 2026 20:49:26 -0400 Subject: [PATCH 3/3] fix(byoc): preserve signer insufficient balance errors Let HTTP 483 reach the existing fail-fast handlers instead of wrapping it as a generic payment failure. --- src/livepeer_gateway/byoc.py | 4 ++++ tests/test_insufficient_balance.py | 27 +++++++++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/src/livepeer_gateway/byoc.py b/src/livepeer_gateway/byoc.py index 4701305..6ce93dc 100644 --- a/src/livepeer_gateway/byoc.py +++ b/src/livepeer_gateway/byoc.py @@ -217,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 = {} diff --git a/tests/test_insufficient_balance.py b/tests/test_insufficient_balance.py index 64135f2..101a732 100644 --- a/tests/test_insufficient_balance.py +++ b/tests/test_insufficient_balance.py @@ -7,6 +7,7 @@ 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 @@ -29,6 +30,32 @@ def test_request_json_maps_483_to_insufficient_balance() -> None: ) +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"