feat(call_runner): send raw bytes as the request body

CHANGELOG.md

@@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Added
+
+- `call_runner` accepts a raw request body: `content` (bytes) with an optional
+  `content_type` (default `application/octet-stream`), as an alternative to the JSON
+  `payload`. Use it for apps that take an image, audio or other binary input.
+
 ### Removed
 
 - gRPC orchestrator information and selection including protobuf bindings

src/livepeer_gateway/http.py

@@ -78,8 +78,12 @@ def _json_request_parts(
     *,
     method: str | None = None,
     payload: dict[str, Any] | None = None,
+    content: bytes | None = None,
+    content_type: str | None = None,
     headers: dict[str, str] | None = None,
 ) -> tuple[str, dict[str, str], bytes | None]:
+    if payload is not None and content is not None:
+        raise ValueError("pass either payload (JSON) or content (raw bytes), not both")
     req_headers: dict[str, str] = {
         "Accept": "application/json",
         "User-Agent": "livepeer-python-gateway/0.1",
@@ -88,10 +92,13 @@ def _json_request_parts(
     if payload is not None:
         req_headers["Content-Type"] = "application/json"
         body = json.dumps(payload).encode("utf-8")
+    elif content is not None:
+        req_headers["Content-Type"] = content_type or "application/octet-stream"
+        body = bytes(content)
     if headers:
         req_headers.update(headers)
 
-    resolved_method = method.upper() if method else ("POST" if payload is not None else "GET")
+    resolved_method = method.upper() if method else ("POST" if body is not None else "GET")
     return resolved_method, req_headers, body
 
 
@@ -228,6 +235,8 @@ async def _request_body(
     *,
     method: str | None = None,
     payload: dict[str, Any] | None = None,
+    content: bytes | None = None,
+    content_type: str | None = None,
     headers: dict[str, str] | None = None,
     timeout: float = 5.0,
 ) -> tuple[bytes, str]:
@@ -245,6 +254,8 @@ async def _request_body(
         url,
         method=method,
         payload=payload,
+        content=content,
+        content_type=content_type,
         headers=headers,
     )
 
@@ -254,7 +265,7 @@ async def _request_body(
         async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session:
             async with session.request(resolved_method, url, data=body, headers=req_headers) as resp:
                 raw = await resp.read()
-                content_type = resp.content_type or ""
+                response_content_type = resp.content_type or ""
                 if resp.status >= 400:
                     _raise_http_json_error(
                         resp.status, url, raw.decode(errors="replace")
@@ -283,7 +294,7 @@ async def _request_body(
             f"HTTP JSON error: unexpected error: {e.__class__.__name__}: {e} (url={url})"
         ) from e
 
-    return raw, content_type
+    return raw, response_content_type
 
 
 async def request_json(
@@ -321,6 +332,8 @@ async def open_stream(
     *,
     method: str | None = None,
     payload: dict[str, Any] | None = None,
+    content: bytes | None = None,
+    content_type: str | None = None,
     headers: dict[str, str] | None = None,
     connect_timeout: float = 10.0,
 ) -> tuple[aiohttp.ClientSession, aiohttp.ClientResponse]:
@@ -336,9 +349,12 @@ async def open_stream(
         url,
         method=method,
         payload=payload,
+        content=content,
+        content_type=content_type,
         headers=headers,
     )
 
+
     timeout = aiohttp.ClientTimeout(total=None, sock_connect=connect_timeout, sock_read=None)
     session = aiohttp.ClientSession(timeout=timeout, connector=aiohttp.TCPConnector(ssl=False))
     try:

src/livepeer_gateway/live_runner.py

@@ -756,6 +756,8 @@ async def call_runner(
     *,
     runner: LiveRunnerInstance | None = ...,
     payload: dict[str, Any] | None = ...,
+    content: bytes | None = ...,
+    content_type: str | None = ...,
     method: str = ...,
     signer_url: str | None = ...,
     signer_headers: dict[str, str] | None = ...,
@@ -773,6 +775,8 @@ async def call_runner(
     *,
     runner: LiveRunnerInstance | None = ...,
     payload: dict[str, Any] | None = ...,
+    content: bytes | None = ...,
+    content_type: str | None = ...,
     method: str = ...,
     signer_url: str | None = ...,
     signer_headers: dict[str, str] | None = ...,
@@ -789,6 +793,8 @@ async def call_runner(
     *,
     runner: LiveRunnerInstance | None = None,
     payload: dict[str, Any] | None = None,
+    content: bytes | None = None,
+    content_type: str | None = None,
     method: str = "POST",
     signer_url: str | None = None,
     signer_headers: dict[str, str] | None = None,
@@ -806,12 +812,26 @@ async def call_runner(
     ``application/json`` and ``+json`` types parse into ``result.data``; anything else
     (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``.
 
+    The request body is JSON (``payload``) or raw bytes (``content``), not both. ``content``
+    is sent as is with ``Content-Type: content_type`` (default ``application/octet-stream``),
+    for apps that take an image, audio or any other binary body.
+
+    ``timeout`` is the total time allowed for each buffered request attempt, so raise it
+    for large bodies. It restarts on every payment retry, so it does not bound the whole
+    call, and it is not applied when ``stream=True``.
+
     The request asks for no particular format, so the app picks what it returns.
     """
     runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "")
     if not runner_url:
         raise LivepeerGatewayError("Live runner call requires runner_url")
-    request_payload = payload or {}
+    if payload is not None and content is not None:
+        raise LivepeerGatewayError("Live runner call takes payload (JSON) or content (bytes), not both")
+    request_payload = None if content is not None else (payload or {})
+    # Only a raw body needs these; JSON calls keep their existing request arguments.
+    body_kwargs: dict[str, Any] = (
+        {"content": content, "content_type": content_type} if content is not None else {}
+    )
     payment_type = _runner_payment_type(runner, payment_unit) if signer_url else ""
     max_price: LiveRunnerPriceInfo | None = None
     if signer_url and runner is not None and runner.price_info is not None:
@@ -872,6 +892,7 @@ async def call_runner(
                     method=method,
                     payload=request_payload,
                     headers=request_headers or None,
+                    **body_kwargs,
                 )
                 call_stream = LiveRunnerCallStream(
                     resp.status,
@@ -894,24 +915,25 @@ async def call_runner(
                 else None
             )
             try:
-                body, content_type = await _request_body(
+                body, response_content_type = await _request_body(
                     runner_url,
                     method=method,
                     payload=request_payload,
+                    **body_kwargs,
                     **request_kwargs,
                 )
             finally:
                 await _stop_funding(pay_task)
             # Non-JSON bodies (an image, ndjson) are handed back unparsed in `content`.
-            is_json = _is_json_content_type(content_type)
+            is_json = _is_json_content_type(response_content_type)
             data: dict[str, Any] = {}
             if is_json:
                 try:
                     data = json.loads(body)
                 except (UnicodeDecodeError, json.JSONDecodeError) as e:
                     raise LivepeerGatewayError(
                         f"HTTP JSON error: endpoint did not return valid JSON: {e} "
-                        f"(url={runner_url}, content_type={content_type})"
+                        f"(url={runner_url}, content_type={response_content_type})"
                     ) from e
                 if not isinstance(data, dict):
                     raise LivepeerGatewayError(
@@ -927,7 +949,7 @@ async def call_runner(
                 ),
                 payment_session=None if payment_type == "fixed" else payment_session,
                 content=None if is_json else body,
-                content_type=content_type,
+                content_type=response_content_type,
             )
         except LivepeerHTTPError as e:
             if e.status_code != 402:

tests/test_call_runner_raw.py

@@ -1,9 +1,13 @@
-"""Tests for non-JSON (raw byte) responses in call_runner.
+"""Tests for raw bytes in call_runner, in both directions.
 
-Single-document JSON responses (``application/json`` or an RFC 6839 ``+json``
-suffix) keep today's behavior: parsed into ``result.data``, strict about being an
+Responses: single-document JSON (``application/json`` or an RFC 6839 ``+json``
+suffix) keeps today's behavior: parsed into ``result.data``, strict about being an
 object. Anything else — binary, or a multi-document format like ndjson — returns
 the body unparsed in ``result.content`` with ``result.content_type`` set.
+
+Requests: ``payload`` keeps sending JSON. ``content`` sends the bytes as is with
+``content_type`` (default ``application/octet-stream``), and the two are mutually
+exclusive.
 """
 
 from __future__ import annotations
@@ -37,6 +41,9 @@ async def main():
     return asyncio.run(main())
 
 
+# --- Responses: what the runner sends back --------------------------------------
+
+
 def test_json_response_unchanged():
     async def handler(request):
         return web.json_response({"message": "hello", "session_id": " s1 "})
@@ -180,3 +187,89 @@ async def scenario(base):
 
     with pytest.raises(LivepeerHTTPError):
         _run(app, scenario)
+
+
+# --- Requests: what the client sends ---------------------------------------------
+
+
+def _echo_app(seen: dict) -> web.Application:
+    """Record the request body, method and content type the server receives."""
+
+    async def handler(request):
+        seen["body"] = await request.read()
+        seen["content_type"] = request.headers.get("Content-Type", "")
+        seen["method"] = request.method
+        return web.json_response({"received": len(seen["body"])})
+
+    app = web.Application()
+    app.router.add_route("*", "/in", handler)
+    return app
+
+
+def test_binary_request_sent_raw():
+    seen: dict = {}
+
+    async def scenario(base):
+        return await call_runner(f"{base}/in", content=FAKE_JPEG, content_type="image/jpeg")
+
+    result = _run(_echo_app(seen), scenario)
+    assert seen["body"] == FAKE_JPEG
+    assert seen["content_type"] == "image/jpeg"
+    assert seen["method"] == "POST"
+    assert result.data == {"received": len(FAKE_JPEG)}
+
+
+def test_binary_request_defaults_to_octet_stream():
+    seen: dict = {}
+
+    async def scenario(base):
+        return await call_runner(f"{base}/in", content=b"\x00\x01\x02")
+
+    _run(_echo_app(seen), scenario)
+    assert seen["body"] == b"\x00\x01\x02"
+    assert seen["content_type"] == "application/octet-stream"
+
+
+def test_empty_binary_request_is_posted():
+    seen: dict = {}
+
+    async def scenario(base):
+        return await call_runner(f"{base}/in", content=b"")
+
+    _run(_echo_app(seen), scenario)
+    assert seen["method"] == "POST"
+    assert seen["body"] == b""
+
+
+def test_json_request_unchanged():
+    seen: dict = {}
+
+    async def scenario(base):
+        return await call_runner(f"{base}/in", payload={"x": 1})
+
+    _run(_echo_app(seen), scenario)
+    assert seen["content_type"] == "application/json"
+    assert seen["body"] == b'{"x": 1}'
+
+
+def test_binary_request_with_stream():
+    seen: dict = {}
+
+    async def scenario(base):
+        async with await call_runner(
+            f"{base}/in", content=FAKE_JPEG, content_type="image/jpeg", stream=True
+        ) as stream:
+            return b"".join([chunk async for chunk in stream.aiter_bytes()])
+
+    body = _run(_echo_app(seen), scenario)
+    assert seen["body"] == FAKE_JPEG
+    assert seen["content_type"] == "image/jpeg"
+    assert b"received" in body
+
+
+def test_payload_and_content_are_mutually_exclusive():
+    async def scenario(base):
+        return await call_runner(f"{base}/in", payload={"x": 1}, content=b"y")
+
+    with pytest.raises(LivepeerGatewayError):
+        _run(_echo_app({}), scenario)

tests/test_live_runner.py

@@ -279,6 +279,77 @@ def _request_body(
 
         assert result.runner is runner
 
+    async def test_paid_call_resends_raw_content_on_retry(self) -> None:
+        calls: list[dict[str, object]] = []
+        runner_url = "https://service.example.com/apps/runner-1/session"
+        runner = LiveRunnerInstance(
+            url=runner_url,
+            app="livepeer/app",
+            runner_id="runner-1",
+            mode="persistent",
+            orchestrator_url="https://service.example.com",
+            raw={},
+            price_info=LiveRunnerPriceInfo(100, "wei", "seconds"),
+        )
+
+        class _PaymentSession:
+            def __init__(self, signer_url: str, **kwargs: object) -> None:
+                pass
+
+            async def get_payment(self) -> object:
+                return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64")
+
+        def _request_body(
+            url: str,
+            *,
+            method: str | None = None,
+            payload: dict[str, object] | None = None,
+            content: bytes | None = None,
+            content_type: str | None = None,
+            headers: dict[str, str] | None = None,
+            timeout: float,
+        ) -> tuple[bytes, str]:
+            calls.append(
+                {
+                    "payload": payload,
+                    "content": content,
+                    "content_type": content_type,
+                    "headers": headers,
+                }
+            )
+            if len(calls) == 1:
+                raise LivepeerHTTPError(
+                    402, url, _payment_challenge_body("manifest-1"), "payment required"
+                )
+            return _json_data({"ok": True})
+
+        with (
+            mock.patch.object(live_runner, "_request_body", side_effect=_request_body),
+            mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession),
+            mock.patch.object(
+                live_runner,
+                "get_signer_info",
+                new_callable=mock.AsyncMock,
+                return_value=SimpleNamespace(address="opaque-payer", sig="opaque-signature"),
+            ),
+        ):
+            result = await call_runner(
+                runner=runner,
+                content=b"\x89PNG-bytes",
+                content_type="image/png",
+                signer_url="https://signer.example.com",
+            )
+
+        assert result.data == {"ok": True}
+        assert len(calls) == 2
+        # The raw body is sent unchanged on the first attempt and again on the paid retry.
+        for call in calls:
+            assert call["payload"] is None
+            assert call["content"] == b"\x89PNG-bytes"
+            assert call["content_type"] == "image/png"
+        assert "Livepeer-Payment" not in calls[0]["headers"]  # type: ignore[operator]
+        assert calls[1]["headers"]["Livepeer-Payment"] == "payment-b64"  # type: ignore[index]
+
     async def test_paid_call_retries_with_payment_headers(self) -> None:
         calls: list[
             tuple[