fix(proxy): abort the request with a 499 when the client disconnects during the body upload (#48640)

Backport of #48640 to lts/1.108.x.
Cherry-picked from 59b8c73930 (main). Dropped the pick's edit to the ValidationError branch the line does not have.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/constants.py

@@ -702,6 +702,7 @@ DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE: Final = os.getenv(
 )
 
 LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED: Final = 499
+LITELLM_CLIENT_DISCONNECT_DETAIL: Final = "Client disconnected the request"
 
 EMAIL_BUDGET_ALERT_TTL: Final = get_env_int("EMAIL_BUDGET_ALERT_TTL", 24 * 60 * 60)  # 24 hours in seconds
 EMAIL_BUDGET_ALERT_MAX_SPEND_ALERT_PERCENTAGE: Final = get_env_float(

litellm/proxy/common_request_processing.py

@@ -38,6 +38,7 @@ from litellm.anthropic_interface.exceptions import AnthropicErrorSseFrame, anthr
 from litellm.constants import (
     DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE,
     DEFAULT_MAX_RECURSE_DEPTH,
+    LITELLM_CLIENT_DISCONNECT_DETAIL,
     LITELLM_DETAILED_TIMING,
     LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED,
     MAX_LITELLM_CALL_ID_LENGTH,
@@ -1784,7 +1785,7 @@ def _has_attribute_error_in_chain(exc: Exception) -> bool:
     return False
 
 
-_CLIENT_DISCONNECT_DETAIL: Final = "Client disconnected the request"
+_CLIENT_DISCONNECT_DETAIL: Final = LITELLM_CLIENT_DISCONNECT_DETAIL
 
 
 @runtime_checkable

litellm/proxy/common_utils/http_parsing_utils.py

@@ -5,16 +5,19 @@ from types import MappingProxyType, UnionType
 from typing import Annotated, Any, Final, Literal, Union, get_args, get_origin
 
 import orjson
-from fastapi import Request, UploadFile, status
+from fastapi import HTTPException, Request, UploadFile, status
 from pydantic import TypeAdapter
 from starlette._utils import get_route_path
-from starlette.requests import HTTPConnection
+from starlette.datastructures import FormData
+from starlette.requests import ClientDisconnect, HTTPConnection
 from typing_extensions import NotRequired, ReadOnly, Required, assert_never
 
 from litellm._logging import verbose_proxy_logger
 from litellm.constants import (
     AZURE_SPEECH_PASS_THROUGH_ROUTE_PREFIX,
     CLIENT_REQUESTED_MODEL_SCOPE_KEY,
+    LITELLM_CLIENT_DISCONNECT_DETAIL,
+    LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED,
     MAX_REQUEST_BODY_SIZE_TO_REPAIR_MB,
 )
 from litellm.integrations.otel.runtime import phase_event
@@ -208,6 +211,29 @@ def _parse_form_metadata(raw_metadata: str) -> object:
     return parsed
 
 
+async def _read_form_data(request: Request) -> FormData:
+    try:
+        form_data: Final = await request.form()
+        _mark_body_received(_declared_content_length(request.headers))
+    except ClientDisconnect:
+        raise
+    except Exception as e:
+        # ``request.form()`` raises on malformed multipart (missing
+        # boundary, malformed chunk encoding, …). Surface as 400 so
+        # the auth-time pre-read does not silently cache ``{}`` while
+        # a later raw-body re-read sees the original payload —
+        # banned-param checks must see the same body the handler
+        # acts on.
+        verbose_proxy_logger.error("Invalid form payload: %s", e)
+        raise ProxyException(
+            message=f"Invalid form payload: {e}",
+            type="invalid_request_error",
+            param="request_body",
+            code=status.HTTP_400_BAD_REQUEST,
+        )
+    return form_data
+
+
 async def read_request_body(request: Request | None) -> dict:
     """
     Safely read the request body and parse it as JSON.
@@ -218,10 +244,9 @@ async def read_request_body(request: Request | None) -> dict:
     Returns:
     - dict: Parsed request data as a dictionary or an empty dictionary if parsing fails
     """
+    if request is None:
+        return {}
     try:
-        if request is None:
-            return {}
-
         if is_otlp_trace_request(request):
             return {}
 
@@ -238,24 +263,7 @@ async def read_request_body(request: Request | None) -> dict:
             _mark_body_received(len(binary_body))
             parsed_body = _parse_binary_body(binary_body)
         elif _is_form_content_type(content_type):
-            try:
-                form_data: Final = await request.form()
-                _mark_body_received(_declared_content_length(request.headers))
-            except Exception as e:
-                # ``request.form()`` raises on malformed multipart (missing
-                # boundary, malformed chunk encoding, …). Surface as 400 so
-                # the auth-time pre-read does not silently cache ``{}`` while
-                # a later raw-body re-read sees the original payload —
-                # banned-param checks must see the same body the handler
-                # acts on.
-                verbose_proxy_logger.error("Invalid form payload: %s", e)
-                raise ProxyException(
-                    message=f"Invalid form payload: {e}",
-                    type="invalid_request_error",
-                    param="request_body",
-                    code=status.HTTP_400_BAD_REQUEST,
-                )
-            form_body: Final = dict(form_data)
+            form_body: Final = dict(await _read_form_data(request))
             raw_metadata: Final = form_body.get("metadata")
             parsed_body = (
                 {**form_body, "metadata": _parse_form_metadata(raw_metadata)}
@@ -318,6 +326,17 @@ async def read_request_body(request: Request | None) -> dict:
         # Re-raise ProxyException as-is
         verbose_proxy_logger.error("Invalid JSON payload received: %s", e)
         raise
+    except ClientDisconnect as e:
+        from litellm.proxy.proxy_server import (
+            _close_dangling_otel_server_span,  # pyright: ignore[reportPrivateUsage]  # proxy_server keeps the span-close helper private; the Starlette HTTPException handler does not stamp the OTel server span opened at auth time, so the body reader must before it aborts
+        )
+
+        verbose_proxy_logger.info("Client disconnected before the request body was fully received")
+        disconnected: Final = HTTPException(
+            status_code=LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED, detail=LITELLM_CLIENT_DISCONNECT_DETAIL
+        )
+        _close_dangling_otel_server_span(request, LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED, exc=disconnected)
+        raise disconnected from e
     except Exception as e:
         # Catch unexpected errors to avoid crashes
         verbose_proxy_logger.exception("Unexpected error reading request body - %s", e)

tests/integration/observability/test_guardrail_effects.py

@@ -1,7 +1,9 @@
 import asyncio
 import json
 import re
+import socket
 import threading
+import time
 import uuid
 from collections.abc import Callable, Iterator, Mapping
 from concurrent.futures import ThreadPoolExecutor
@@ -11,6 +13,7 @@ from dataclasses import dataclass
 from pathlib import Path
 from types import MappingProxyType
 from typing import Final
+from urllib.parse import urlsplit
 
 import anthropic
 import httpx
@@ -2928,3 +2931,60 @@ def test_disconnect_scans_keep_recording_after_a_worker_is_killed(gateway: Gatew
         assert all(rig.secret(index) in line for index, line in enumerate(received)), received
         assert _post_call_statuses(rig, rig.model, rows=8) == (("success",),) * 8
         assert rig.owned.process.poll() is None
+
+
+def _openai_upstream(request: Request) -> Reply:
+    if request.target.endswith("/vector_stores"):
+        return Reply(body=json.dumps({"id": "vs_" + uuid.uuid4().hex, "object": "vector_store", "name": None}).encode())
+    return Reply(body=_chat_completion_body("control-reply"))
+
+
+def _post_half_of_the_body_then_hang_up(gateway: Gateway, path: str, body: bytes) -> None:
+    address: Final = urlsplit(str(gateway.client.base_url))
+    assert address.hostname is not None and address.port is not None, address
+    head: Final = (
+        f"POST {path} HTTP/1.1\r\nHost: {address.netloc}\r\nAuthorization: Bearer {gateway.key}\r\n"
+        f"Content-Type: application/json\r\nContent-Length: {len(body)}\r\n\r\n"
+    ).encode()
+    with socket.create_connection((address.hostname, address.port)) as connection:
+        connection.sendall(head + body[: len(body) // 2])
+        time.sleep(0.5)
+
+
+@pytest.mark.parametrize(
+    ("path", "body"),
+    (
+        pytest.param(
+            "/v1/chat/completions",
+            b'{"model": "MODEL", "messages": [{"role": "user", "content": "Say OK"}], "max_tokens": 16}',
+            id="chat-completions-where-an-empty-body-is-invalid",
+        ),
+        pytest.param(
+            "/v1/vector_stores",
+            b'{"name": "integration-store", "metadata": {"owner": "integration", "purpose": "half upload"}}',
+            id="vector-stores-where-an-empty-body-is-valid",
+        ),
+    ),
+)
+def test_client_disconnect_during_the_body_upload_runs_no_endpoint_and_writes_no_spend_row(
+    gateway: Gateway, tmp_path: Path, path: str, body: bytes
+) -> None:
+    with (
+        wire_server(_openai_upstream) as upstream,
+        owned_proxy_process(
+            gateway, tmp_path, {"OPENAI_BASE_URL": upstream.url + "/v1", "OPENAI_API_KEY": "synthetic-openai-key"}
+        ) as owned,
+        owned.gateway.scenario() as scenario,
+    ):
+        model: Final = scenario.model(api_base=upstream.url + "/v1", api_key="synthetic-openai-key")
+        _post_half_of_the_body_then_hang_up(owned.gateway, path, body.replace(b"MODEL", model.encode()))
+        control: Final = owned.gateway.chat(model)
+        rows: Final = eventually(
+            lambda: owned.gateway.rows('SELECT request_id, call_type, status FROM "LiteLLM_SpendLogs"', ()),
+            lambda values: any(row["request_id"] == control["id"] for row in values),
+            seconds=70,
+        )
+        assert [(row["call_type"], row["status"]) for row in rows] == [("acompletion", "success")], rows
+        assert [request.target for request in upstream.drain()] == ["/v1/chat/completions"], (
+            "the half-uploaded request must never reach the provider"
+        )

tests/unit/proxy/common_utils/test_http_parsing_utils.py

@@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import orjson
 import pytest
+from fastapi import HTTPException
 from fastapi.testclient import TestClient
 from starlette.datastructures import FormData
 from starlette.requests import Request
@@ -1592,3 +1593,44 @@ async def test_form_data_metadata_keeps_non_ascii_and_valid_surrogate_pairs():
     result = await read_request_body(mock_request)
 
     assert result["metadata"]["tags"] == ["café", "😀"]
+
+
+def _request_whose_client_disconnects_mid_upload(content_type: str) -> Request:
+    scope: Final = {
+        "type": "http",
+        "method": "POST",
+        "path": "/v1/chat/completions",
+        "headers": [(b"content-type", content_type.encode()), (b"content-length", b"96")],
+        "query_string": b"",
+    }
+
+    async def receive() -> Message:
+        return {"type": "http.disconnect"}
+
+    return Request(scope, receive)
+
+
+@pytest.mark.parametrize(
+    "content_type",
+    ("application/json", "multipart/form-data; boundary=synthetic-boundary", "application/x-protobuf"),
+)
+async def test_read_request_body_aborts_with_499_when_the_client_disconnects_mid_upload(content_type: str) -> None:
+    request: Final = _request_whose_client_disconnects_mid_upload(content_type)
+    with pytest.raises(HTTPException) as raised:
+        await read_request_body(request)
+    assert (raised.value.status_code, raised.value.detail) == (499, "Client disconnected the request")
+    assert _safe_get_request_parsed_body(request) is None, "a half-received body must never be cached as {}"
+
+
+async def test_read_request_body_closes_the_otel_server_span_when_the_client_disconnects_mid_upload(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from litellm.proxy import proxy_server
+
+    span: Final = MagicMock()
+    monkeypatch.setattr(proxy_server, "open_telemetry_logger", MagicMock(), raising=False)
+    request: Final = _request_whose_client_disconnects_mid_upload("application/json")
+    request.state.parent_otel_span = span
+    with pytest.raises(HTTPException):
+        await read_request_body(request)
+    assert (span.end.called, request.state.parent_otel_span) == (True, None), "the auth-time span must not dangle"