From cefaef4d80d72ba614c07ff5129e8b7ff0f68045 Mon Sep 17 00:00:00 2001 From: sufubao Date: Tue, 4 Aug 2026 17:53:51 +0800 Subject: [PATCH] fix(api): harden OpenAI streaming/non-streaming error handling Unexpected exceptions raised inside chat_completions/completions handlers and inside the SSE stream wrapper currently propagate to Starlette, producing a 500 with a long traceback and no structured error body. The non-OpenAI /generate handlers already catch the broad Exception case; this aligns the OpenAI endpoints with that pattern. - api_http.py: add a final `except Exception` arm to chat_completions and completions, returning a structured EXPECTATION_FAILED (417) response. - api_openai.py: extend _safe_stream_wrapper to also handle ServerBusyError (503 SSE error), re-raise asyncio.CancelledError, and convert any other exception into an InternalServerError SSE chunk. Streaming failures now increment lightllm_request_failure, matching the non-streaming create_error_response path. --- lightllm/server/api_http.py | 6 ++ lightllm/server/api_openai.py | 33 ++++++- .../test_openai_stream_error_handling.py | 98 +++++++++++++++++++ 3 files changed, 134 insertions(+), 3 deletions(-) create mode 100644 unit_tests/server/test_openai_stream_error_handling.py diff --git a/lightllm/server/api_http.py b/lightllm/server/api_http.py index 9c8981730a..11417dac48 100755 --- a/lightllm/server/api_http.py +++ b/lightllm/server/api_http.py @@ -376,6 +376,9 @@ async def chat_completions(request: ChatCompletionRequest, raw_request: Request) except ClientDisconnected as e: logger.warning(str(e)) return Response(status_code=499) + except Exception as e: + logger.error("An error occurred: %s", str(e), exc_info=True) + return create_error_response(HTTPStatus.EXPECTATION_FAILED, str(e)) return resp @@ -396,6 +399,9 @@ async def completions(request: CompletionRequest, raw_request: Request) -> Respo except ClientDisconnected as e: logger.warning(str(e)) return Response(status_code=499) + except Exception as e: + logger.error("An error occurred: %s", str(e), exc_info=True) + return create_error_response(HTTPStatus.EXPECTATION_FAILED, str(e)) return resp diff --git a/lightllm/server/api_openai.py b/lightllm/server/api_openai.py index 0d934c44c9..33a97835c9 100644 --- a/lightllm/server/api_openai.py +++ b/lightllm/server/api_openai.py @@ -30,7 +30,7 @@ from .httpserver_for_pd_master.manager import HttpServerManagerForPDMaster from .api_lightllm import lightllm_get_score from lightllm.utils.envs_utils import get_env_start_args, get_lightllm_websocket_max_message_size -from lightllm.utils.error_utils import ClientDisconnected +from lightllm.utils.error_utils import ClientDisconnected, ServerBusyError from lightllm.utils.log_utils import init_logger from lightllm.server.metrics.manager import MetricClient @@ -60,6 +60,23 @@ logger = init_logger(__name__) +def _record_request_failure_metric(): + try: + from .api_http import g_objs + + if g_objs.metric_client is not None: + g_objs.metric_client.counter_inc("lightllm_request_failure") + except Exception: + logger.warning("Failed to record lightllm_request_failure metric", exc_info=True) + + +def _stream_error_chunk(message: str, err_type: str, code: Optional[int] = None): + error = {"message": message, "type": err_type} + if code is not None: + error["code"] = code + return f"data: {json.dumps({'error': error}, ensure_ascii=False)}\n\n" + + async def _safe_stream_wrapper(stream_generator): """Wrap a streaming generator to catch ValueError (e.g. input too long) and yield an SSE error event instead of letting the exception propagate to Starlette which prints a long traceback.""" @@ -67,12 +84,22 @@ async def _safe_stream_wrapper(stream_generator): async for item in stream_generator: yield item except ValueError as e: - error_data = json.dumps({"error": {"message": str(e), "type": "invalid_request_error"}}, ensure_ascii=False) - yield f"data: {error_data}\n\n" + _record_request_failure_metric() + yield _stream_error_chunk(str(e), "invalid_request_error") + except ServerBusyError as e: + logger.error("%s", str(e), exc_info=True) + _record_request_failure_metric() + yield _stream_error_chunk(str(e), "ServerBusyError", e.status_code) except ClientDisconnected as e: logger.warning(str(e)) # Client is gone — there's no point yielding more SSE chunks. Stop quietly. return + except asyncio.CancelledError: + raise + except Exception as e: + logger.error("An error occurred in streaming response: %s", str(e), exc_info=True) + _record_request_failure_metric() + yield _stream_error_chunk(str(e), "InternalServerError") def _serialize_sse_chunk(chunk, choice_nulls=(), response_nulls=()): diff --git a/unit_tests/server/test_openai_stream_error_handling.py b/unit_tests/server/test_openai_stream_error_handling.py new file mode 100644 index 0000000000..c7d2f23f85 --- /dev/null +++ b/unit_tests/server/test_openai_stream_error_handling.py @@ -0,0 +1,98 @@ +import asyncio + +import ujson as json + +from lightllm.server import api_http +from lightllm.server.api_openai import _safe_stream_wrapper +from lightllm.utils.error_utils import ServerBusyError + + +class FakeMetricClient: + def __init__(self): + self.counters = [] + + def counter_inc(self, name): + self.counters.append(name) + + +def _decode_sse_payload(chunk): + if isinstance(chunk, bytes): + chunk = chunk.decode("utf-8") + assert chunk.startswith("data: ") + return json.loads(chunk.removeprefix("data: ").strip()) + + +def _collect(gen): + async def drain(): + return [c async for c in gen] + + return asyncio.run(drain()) + + +def test_safe_stream_wrapper_converts_unexpected_exception_to_sse_error(monkeypatch): + metric_client = FakeMetricClient() + monkeypatch.setattr(api_http.g_objs, "metric_client", metric_client) + + async def failing_stream(): + if False: + yield b"" + raise RuntimeError("backend failed") + + chunks = _collect(_safe_stream_wrapper(failing_stream())) + + assert len(chunks) == 1 + payload = _decode_sse_payload(chunks[0]) + assert payload["error"]["message"] == "backend failed" + assert payload["error"]["type"] == "InternalServerError" + assert metric_client.counters == ["lightllm_request_failure"] + + +def test_safe_stream_wrapper_maps_server_busy_error(monkeypatch): + metric_client = FakeMetricClient() + monkeypatch.setattr(api_http.g_objs, "metric_client", metric_client) + + async def busy_stream(): + if False: + yield b"" + raise ServerBusyError("server overloaded") + + chunks = _collect(_safe_stream_wrapper(busy_stream())) + + assert len(chunks) == 1 + payload = _decode_sse_payload(chunks[0]) + assert payload["error"]["type"] == "ServerBusyError" + assert payload["error"]["code"] == 503 + assert metric_client.counters == ["lightllm_request_failure"] + + +def test_safe_stream_wrapper_keeps_value_error_as_invalid_request(monkeypatch): + metric_client = FakeMetricClient() + monkeypatch.setattr(api_http.g_objs, "metric_client", metric_client) + + async def bad_stream(): + if False: + yield b"" + raise ValueError("input too long") + + chunks = _collect(_safe_stream_wrapper(bad_stream())) + + payload = _decode_sse_payload(chunks[0]) + assert payload["error"]["type"] == "invalid_request_error" + assert metric_client.counters == ["lightllm_request_failure"] + + +def test_safe_stream_wrapper_swallows_client_disconnect(monkeypatch): + from lightllm.utils.error_utils import ClientDisconnected + + metric_client = FakeMetricClient() + monkeypatch.setattr(api_http.g_objs, "metric_client", metric_client) + + async def gone_stream(): + if False: + yield b"" + raise ClientDisconnected("client gone") + + chunks = _collect(_safe_stream_wrapper(gone_stream())) + + assert chunks == [] + assert metric_client.counters == []