From 44a0f577a8d8243e7cec21931968297594fd4093 Mon Sep 17 00:00:00 2001 From: Shivam Rawat Date: Sat, 4 Jul 2026 12:41:16 -0700 Subject: [PATCH] fix(proxy): stop double-logging and false-alerting on passthrough upstream errors Two bugs from the upstream-error fixes: the success handler has no status-code awareness, so removing raise_for_status() left it firing for every upstream 4xx/5xx too, meaning the new failure hook and the existing success handler both logged the same request (corrupting SpendLogs/cost tracking). Separately, the failure hook was passed the raw httpx.HTTPStatusError, which ProxyLogging's alerting only excludes HTTPException/ProxyException from, so a normal upstream 403 would trigger a "High" severity llm_exceptions alert. Gates the success handler (both non-streaming and end-of-stream) to status_code < 400, and reports upstream failures to post_call_failure_hook as an HTTPException instead of the raw httpx error, matching how auth/rate-limit errors are already excluded from alerting. Co-authored-by: Cursor --- .../pass_through_endpoints.py | 75 ++++++++----- .../streaming_handler.py | 6 +- .../test_pass_through_endpoints.py | 104 ++++++++++++++++-- .../test_streaming_handler_interrupt.py | 35 ++++++ 4 files changed, 180 insertions(+), 40 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index f76758485e3..0ac4182ebd2 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -719,11 +719,22 @@ async def _log_passthrough_upstream_failure( try: response.raise_for_status() - except httpx.HTTPStatusError as e: + except httpx.HTTPStatusError: + # Reported as an HTTPException, not the raw httpx error: ProxyLogging's + # alerting path only excludes HTTPException/ProxyException from its + # "High" severity llm_exceptions alert, treating everything else as an + # operational LLM-API failure. An upstream 4xx/5xx returned unchanged + # to the client is a user-facing error like any other, not something + # ops needs paged for, so it must be excluded the same way auth and + # rate-limit errors already are. + synthetic_exception = HTTPException( + status_code=response.status_code, + detail=f"Upstream passthrough request failed with status {response.status_code}", + ) try: await proxy_logging_obj.post_call_failure_hook( user_api_key_dict=user_api_key_dict, - original_exception=e, + original_exception=synthetic_exception, request_data=request_payload, traceback_str=traceback.format_exc(limit=MAXIMUM_TRACEBACK_LINES_TO_LOG), ) @@ -1215,25 +1226,28 @@ async def pass_through_request( status_code=response.status_code, ) - await _log_passthrough_upstream_failure( - response=response, - user_api_key_dict=user_api_key_dict, - request_payload=_build_passthrough_failure_request_payload( - parsed_body=_parsed_body, - kwargs=kwargs, - logging_obj=logging_obj, - custom_llm_provider=custom_llm_provider, - ), - ) - content = await response.aread() ## POST-CALL GUARDRAILS ## # Guardrails and managed-id rewriting only apply to successful upstream # responses; response_body itself is parsed unconditionally so the - # success-handler log payload still reflects upstream error bodies. + # failure-hook log payload below still reflects upstream error bodies. _content_modified = False response_body: Optional[dict] = get_response_body(response) + + failure_request_payload = _build_passthrough_failure_request_payload( + parsed_body=_parsed_body, + kwargs=kwargs, + logging_obj=logging_obj, + custom_llm_provider=custom_llm_provider, + ) + failure_request_payload["response_body"] = response_body + await _log_passthrough_upstream_failure( + response=response, + user_api_key_dict=user_api_key_dict, + request_payload=failure_request_payload, + ) + if response.status_code < 400 and response_body is not None and guardrails_to_run: # Build an enriched data dict: _parsed_body has been stripped of # `metadata` by both pre_call_hook and _init_kwargs_for_pass_through_endpoint, @@ -1325,23 +1339,28 @@ async def pass_through_request( ) ## LOG SUCCESS + # Upstream errors are already logged via _log_passthrough_upstream_failure + # above; the success handler has no status-code awareness of its own; so + # calling it here for a 4xx/5xx would double-log the same request as both + # a failure and a success (corrupting spend tracking). passthrough_logging_payload["response_body"] = response_body end_time = datetime.now() - GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( - async_coroutine=pass_through_endpoint_logging.pass_through_async_success_handler( - httpx_response=response, - response_body=response_body, - url_route=str(url), - result="", - start_time=start_time, - end_time=end_time, - logging_obj=logging_obj, - cache_hit=False, - request_body=_parsed_body or {}, - custom_llm_provider=custom_llm_provider, - **kwargs, + if response.status_code < 400: + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=pass_through_endpoint_logging.pass_through_async_success_handler( + httpx_response=response, + response_body=response_body, + url_route=str(url), + result="", + start_time=start_time, + end_time=end_time, + logging_obj=logging_obj, + cache_hit=False, + request_body=_parsed_body or {}, + custom_llm_provider=custom_llm_provider, + **kwargs, + ) ) - ) ## CUSTOM HEADERS - `x-litellm-*` custom_headers = ProxyBaseLLMRequestProcessing.get_custom_headers( diff --git a/litellm/proxy/pass_through_endpoints/streaming_handler.py b/litellm/proxy/pass_through_endpoints/streaming_handler.py index 1bdd57507e3..af61281243d 100644 --- a/litellm/proxy/pass_through_endpoints/streaming_handler.py +++ b/litellm/proxy/pass_through_endpoints/streaming_handler.py @@ -89,7 +89,11 @@ class PassThroughStreamingHandler: # GeneratorExit (raised on client disconnect) is not caught by # `except Exception`; the finally block ensures partial usage # still gets logged for spend tracking. See LIT-2642. - if not logging_scheduled and raw_bytes: + # Upstream 4xx/5xx responses are already logged as a failure by + # the caller before this generator starts (see + # _log_passthrough_upstream_failure); logging them again here as + # a success would double-log the same request. + if not logging_scheduled and raw_bytes and response.status_code < 400: logging_scheduled = True try: GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( diff --git a/tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py b/tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py index b7d00d084e1..1482937ab3b 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py @@ -3689,21 +3689,90 @@ async def test_pass_through_request_non_streaming_upstream_error_returned_unchan # not stringified into a ProxyException's `error.message` field. assert body == _UPSTREAM_ERROR_BODY assert set(body.keys()) != {"error"} or not isinstance(body["error"], dict) - mock_success_handler.assert_called_once() + + # Regression: the success handler has no status-code awareness, so it must + # never be called for a 4xx/5xx upstream response - otherwise the same + # request gets recorded as both a failure and a success in SpendLogs. + mock_success_handler.assert_not_called() # Regression: post_call_failure_hook (spend-tracking, alerting callbacks) # must still fire for upstream errors even though the client-facing # response is unchanged and no ProxyException is raised. + from fastapi import HTTPException + mock_proxy_logging.post_call_failure_hook.assert_called_once() failure_call_kwargs = mock_proxy_logging.post_call_failure_hook.call_args.kwargs - assert isinstance(failure_call_kwargs["original_exception"], httpx.HTTPStatusError) - assert failure_call_kwargs["original_exception"].response.status_code == 403 + # Must be reported as HTTPException, not the raw httpx error: ProxyLogging's + # alerting only excludes HTTPException/ProxyException from its "High" + # severity llm_exceptions alert, so a raw HTTPStatusError here would page + # ops for every routine upstream 4xx returned through passthrough. + assert isinstance(failure_call_kwargs["original_exception"], HTTPException) + assert failure_call_kwargs["original_exception"].status_code == 403 - # Regression: the log payload's response_body must reflect the upstream - # error JSON, not None, so downstream spend-tracking/logging integrations - # can see what the upstream actually returned. - success_call_kwargs = mock_success_handler.call_args.kwargs - assert success_call_kwargs["response_body"] == _UPSTREAM_ERROR_BODY + # Regression: the failure-hook log payload's response_body must reflect + # the upstream error JSON, not None, so downstream spend-tracking/logging + # integrations can see what the upstream actually returned. + assert failure_call_kwargs["request_data"]["response_body"] == _UPSTREAM_ERROR_BODY + + +@pytest.mark.asyncio +async def test_pass_through_request_upstream_error_failure_hook_exception_is_swallowed(): + """ + A broken failure-hook callback (e.g. a misconfigured alerting integration) + must never take down the passthrough response - the upstream error body + must still reach the client unchanged, and the callback's exception must + only be logged, not raised. + """ + upstream_content = json.dumps(_UPSTREAM_ERROR_BODY).encode("utf-8") + upstream_response = httpx.Response( + status_code=403, + headers={"content-type": "application/json"}, + content=upstream_content, + request=httpx.Request("POST", "http://target-api.com/api/denied"), + ) + + with patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_proxy_logging: + with patch( + "litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_async_httpx_client" + ) as mock_get_client: + with patch( + "litellm.proxy.pass_through_endpoints.pass_through_endpoints.ProxyBaseLLMRequestProcessing" + ) as mock_processing: + with patch( + "litellm.proxy.pass_through_endpoints.pass_through_endpoints.pass_through_endpoint_logging.pass_through_async_success_handler" + ) as mock_success_handler: + mock_proxy_logging.pre_call_hook = AsyncMock(return_value={}) + mock_proxy_logging.post_call_failure_hook = AsyncMock( + side_effect=RuntimeError("alerting integration misconfigured") + ) + mock_proxy_logging.post_call_response_headers_hook = AsyncMock( + return_value=None + ) + mock_processing.get_custom_headers.return_value = {} + mock_success_handler.return_value = None + + async_client = MagicMock() + async_client.request = AsyncMock(return_value=upstream_response) + mock_get_client.return_value = MagicMock(client=async_client) + + mock_request = MagicMock(spec=Request) + mock_request.method = "POST" + mock_request.url = "http://test-proxy.com/mock-upstream/api/denied" + mock_request.body = AsyncMock(return_value=b'{"action": "read"}') + mock_request.headers = Headers({"content-type": "application/json"}) + mock_request.query_params = QueryParams({}) + + response = await pass_through_request( + request=mock_request, + target="http://target-api.com/api/denied", + custom_headers={}, + user_api_key_dict=MagicMock(), + ) + await asyncio.sleep(0) + + mock_proxy_logging.post_call_failure_hook.assert_called_once() + assert response.status_code == 403 + assert json.loads(response.body) == _UPSTREAM_ERROR_BODY @pytest.mark.asyncio @@ -3764,12 +3833,22 @@ async def test_pass_through_request_streaming_upstream_error_returned_unchanged( assert streamed_bytes == upstream_content assert json.loads(streamed_bytes) == _UPSTREAM_ERROR_BODY + # Regression: chunk_processor's end-of-stream success logging has no + # status-code awareness, so it must never fire for a 4xx/5xx upstream + # response - otherwise the same request gets recorded as both a failure + # (via the hook below) and a success in SpendLogs. + mock_success_handler.assert_not_called() + # Regression: post_call_failure_hook must still fire for streaming - # upstream errors, mirroring the non-streaming behavior. + # upstream errors, mirroring the non-streaming behavior, and must also + # report an HTTPException (not the raw httpx error) to avoid triggering + # a "High" severity llm_exceptions alert for a routine upstream 4xx. + from fastapi import HTTPException + mock_proxy_logging.post_call_failure_hook.assert_called_once() failure_call_kwargs = mock_proxy_logging.post_call_failure_hook.call_args.kwargs - assert isinstance(failure_call_kwargs["original_exception"], httpx.HTTPStatusError) - assert failure_call_kwargs["original_exception"].response.status_code == 403 + assert isinstance(failure_call_kwargs["original_exception"], HTTPException) + assert failure_call_kwargs["original_exception"].status_code == 403 @pytest.mark.asyncio @@ -3826,6 +3905,9 @@ async def test_pass_through_request_non_streaming_success_unchanged(): # Regression guard: the failure hook must only fire for upstream errors, # never for a successful upstream response. mock_proxy_logging.post_call_failure_hook.assert_not_called() + # ...and the success handler must still fire exactly once for a 2xx, + # proving the status_code gate doesn't also swallow real successes. + mock_success_handler.assert_called_once() @pytest.mark.asyncio diff --git a/tests/test_litellm/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py b/tests/test_litellm/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py index b781190eaef..80702e605e7 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py @@ -93,6 +93,41 @@ async def test_chunk_processor_logs_on_client_disconnect(): assert call_kwargs["raw_bytes"] == [chunks[0]] +@pytest.mark.asyncio +async def test_chunk_processor_does_not_schedule_success_logging_for_upstream_error(): + """A 4xx/5xx upstream response is already logged as a failure by the caller + before this generator starts; scheduling success logging here too would + double-log the same request in SpendLogs.""" + chunks = [b'{"error": "denied"}'] + response = _make_streaming_response(chunks) + response.status_code = 403 + + mock_logging_obj = MagicMock() + mock_passthrough_handler = MagicMock() + + with patch.object( + PassThroughStreamingHandler, + "_route_streaming_logging_to_handler", + new=AsyncMock(), + ) as mock_route: + received = [] + async for chunk in PassThroughStreamingHandler.chunk_processor( + response=response, + request_body={"model": "claude-3-haiku"}, + litellm_logging_obj=mock_logging_obj, + endpoint_type=EndpointType.GENERIC, + start_time=datetime.now(), + passthrough_success_handler_obj=mock_passthrough_handler, + url_route="/bedrock/model/claude/invoke-with-response-stream", + ): + received.append(chunk) + + await asyncio.sleep(0) + + assert received == chunks + mock_route.assert_not_called() + + @pytest.mark.asyncio async def test_chunk_processor_does_not_schedule_logging_when_no_chunks(): response = _make_streaming_response([])