From 97cf00b35ff2f188e77d301463b273ff34e3def4 Mon Sep 17 00:00:00 2001 From: Lloyd Mak Date: Fri, 21 Aug 2026 02:14:28 +0000 Subject: [PATCH 1/2] Return a real HTTP status when a stream fails before its first event If the first event from an upstream provider is an error, we currently fall through to `break None`, commit HTTP 200, open an SSE body, and deliver the failure as an in-band error frame. The client sees a successful response that contains an error - and because `[DONE]` is still appended afterwards, a client keying on `[DONE]` reads it as a successful empty completion rather than a failure. Response headers are still mutable at that point: no stream has been produced and nothing has been written. So map the error through the existing `map_provider_error` / `map_domain_error_to_status` path and return the real status instead. This costs no latency. The first event is already awaited before the response is constructed, so nothing new is blocking. Reproduced against production on 2026-08-19, streaming google/gemma-4-31B-it. A schema containing `uniqueItems` returns HTTP 200 with a single SSE event carrying `HTTP error 400: Grammar error: Unimplemented keys: ["uniqueItems"]`, followed by `[DONE]`. The same shape occurs for `contains` and `propertyNames`. A partner reported 46 of these in a six-hour window and could not distinguish them from empty successes. This fixes the whole class, not one keyword: any pre-first-event upstream error now surfaces as a real status. `map_provider_error` becomes `pub` so the api crate can reach it. The alternative was duplicating error mapping into api, which is worse. Verified: cargo build, cargo clippy -p services -p api --all-targets -D warnings, and cargo test -p api --test e2e_all --no-run all clean. The e2e tests compile but were not executed here - they require PostgreSQL, which is not available in this environment. No test-pass claim is made. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Fm9f3maaY2wd6yHH8FEjR5 --- crates/api/src/routes/completions.rs | 16 +++- .../api/tests/e2e_all/first_stream_event.rs | 82 +++++++++++++++++++ crates/api/tests/e2e_all/main.rs | 1 + crates/services/src/completions/mod.rs | 2 +- 4 files changed, 99 insertions(+), 2 deletions(-) create mode 100644 crates/api/tests/e2e_all/first_stream_event.rs diff --git a/crates/api/src/routes/completions.rs b/crates/api/src/routes/completions.rs index b64015e67..9fad02a85 100644 --- a/crates/api/src/routes/completions.rs +++ b/crates/api/src/routes/completions.rs @@ -1539,7 +1539,21 @@ async fn chat_completions_inner( } true } - _ => break None, + Some(Err(error)) => { + let domain_error = services::CompletionServiceImpl::map_provider_error( + &request.model, + error, + "chat completion stream", + api_key.organization.id.0, + ); + let status_code = map_domain_error_to_status(&domain_error); + return ( + status_code, + ResponseJson::(domain_error.into()), + ) + .into_response(); + } + None => break None, }; if is_control { if leading_control.len() >= MAX_LEADING_CONTROL_EVENTS { diff --git a/crates/api/tests/e2e_all/first_stream_event.rs b/crates/api/tests/e2e_all/first_stream_event.rs new file mode 100644 index 000000000..25d975194 --- /dev/null +++ b/crates/api/tests/e2e_all/first_stream_event.rs @@ -0,0 +1,82 @@ +use crate::common::*; +use inference_providers::mock::{RequestMatcher, ResponseTemplate}; + +#[tokio::test] +async fn first_upstream_error_is_returned_as_http_error_before_sse_starts() { + // Given + let (server, _pool, mock, _db) = setup_test_server_with_pool().await; + let model = setup_qwen_model(&server).await; + let org = setup_org_with_credits(&server, 10_000_000_000i64).await; + let api_key = get_api_key_for_org(&server, org.id).await; + mock.set_stream_error_override(Some(inference_providers::CompletionError::HttpError { + status_code: 400, + message: "Grammar error: Unimplemented keys: [\"uniqueItems\"]".to_string(), + is_external: false, + })) + .await; + + // When + let response = server + .post("/v1/chat/completions") + .add_header("Authorization", format!("Bearer {api_key}")) + .json(&serde_json::json!({ + "model": model, + "messages": [{"role": "user", "content": "Return JSON."}], + "stream": true + })) + .await; + + // Then + assert_eq!(response.status_code(), 400, "{}", response.text()); + assert_eq!( + response + .headers() + .get("content-type") + .and_then(|value| value.to_str().ok()), + Some("application/json") + ); + let body = response.json::(); + assert_eq!(body["error"]["type"], "invalid_request_error"); + assert!(body["error"]["message"] + .as_str() + .is_some_and(|message| message.contains("uniqueItems"))); +} + +#[tokio::test] +async fn normal_first_upstream_chunk_remains_first_and_unmodified() { + // Given + let (server, _pool, mock, _db) = setup_test_server_with_pool().await; + let model = setup_qwen_model(&server).await; + let org = setup_org_with_credits(&server, 10_000_000_000i64).await; + let api_key = get_api_key_for_org(&server, org.id).await; + mock.when(RequestMatcher::Any) + .respond_with(ResponseTemplate::new("first second")) + .await; + + // When + let response = server + .post("/v1/chat/completions") + .add_header("Authorization", format!("Bearer {api_key}")) + .json(&serde_json::json!({ + "model": model, + "messages": [{"role": "user", "content": "Stream two words."}], + "stream": true, + "stream_options": {"continuous_usage_stats": true} + })) + .await; + + // Then + assert_eq!(response.status_code(), 200, "{}", response.text()); + let response_text = response.text(); + let first = response_text + .lines() + .find_map(|line| line.strip_prefix("data: ")) + .expect("stream should contain a first data event"); + let chunk = serde_json::from_str::(first).expect("valid first chunk"); + assert_eq!(chunk["choices"][0]["delta"]["content"], "first"); + assert_eq!( + chunk["mock_upstream_only_field"], "dropped-by-typed-parse", + "the provider's raw first chunk must bypass typed re-serialization" + ); + assert!(chunk.get("error").is_none()); +} diff --git a/crates/api/tests/e2e_all/main.rs b/crates/api/tests/e2e_all/main.rs index cb7844523..4d04aa076 100644 --- a/crates/api/tests/e2e_all/main.rs +++ b/crates/api/tests/e2e_all/main.rs @@ -41,6 +41,7 @@ mod error_msg; mod external_providers; mod feature_requests; mod files; +mod first_stream_event; mod function_tools; mod general; mod glm52_tier_routing; diff --git a/crates/services/src/completions/mod.rs b/crates/services/src/completions/mod.rs index abafd2271..7b0a9e272 100644 --- a/crates/services/src/completions/mod.rs +++ b/crates/services/src/completions/mod.rs @@ -945,7 +945,7 @@ impl CompletionServiceImpl { ] } - pub(crate) fn map_provider_error( + pub fn map_provider_error( model: &str, error: &inference_providers::CompletionError, operation: &str, From 2d5b8bdb940cc7666b0dcded91e24b0f1d5eaca3 Mon Sep 17 00:00:00 2001 From: Lloyd Mak Date: Thu, 27 Aug 2026 20:59:40 +0000 Subject: [PATCH 2/2] Detect the first-event error in the pool instead of the route Review feedback: the route should not inspect stream events. It does not need to - the pool already peeks the first event on every streaming request, to record the chat_id for sticky routing: let mut peekable = StreamingResultExt::peekable(stream); ... if let Some(Ok(event)) = peekable.peek().await { That peek predates this PR and discards the Err case, handing back a stream whose first item is an error and forcing the caller to either inspect it or commit 200 and bury the failure in-band. So the earlier version added a second, duplicate peek one layer up. The pool now returns Err from the peek it already performs, and the route's arm is removed, restoring that loop to its shape on main. Observable behaviour is unchanged: a stream whose first upstream event is an error still returns a real HTTP status via the existing error path, the same one that already yields 400 for an unknown model. Net effect is one fewer inspection than before. The orphaned-pending-client cleanup still runs on the new error path - the `if !pinned` block executes before the early return - so an Err cannot leak a pinned connection. The error is cloned out of the peek so its status and is_external survive for map_provider_error, which classifies on both (a 404 from a third-party provider maps differently from a 404 from our own vLLM). map_provider_error reverts to pub(crate); it was only made pub for the route arm this removes, and has no callers outside the services crate. Verified: the e2e assertions in first_stream_event.rs are unchanged - only the layer that detects the error moved. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Fm9f3maaY2wd6yHH8FEjR5 --- crates/api/src/routes/completions.rs | 16 +-- crates/services/src/completions/mod.rs | 2 +- .../src/inference_provider_pool/mod.rs | 112 ++++++++++++++++-- 3 files changed, 101 insertions(+), 29 deletions(-) diff --git a/crates/api/src/routes/completions.rs b/crates/api/src/routes/completions.rs index 2873515f3..9671508bf 100644 --- a/crates/api/src/routes/completions.rs +++ b/crates/api/src/routes/completions.rs @@ -1550,21 +1550,7 @@ async fn chat_completions_inner( } true } - Some(Err(error)) => { - let domain_error = services::CompletionServiceImpl::map_provider_error( - &request.model, - error, - "chat completion stream", - api_key.organization.id.0, - ); - let status_code = map_domain_error_to_status(&domain_error); - return ( - status_code, - ResponseJson::(domain_error.into()), - ) - .into_response(); - } - None => break None, + _ => break None, }; if is_control { if leading_control.len() >= MAX_LEADING_CONTROL_EVENTS { diff --git a/crates/services/src/completions/mod.rs b/crates/services/src/completions/mod.rs index 949576379..cc5e39e46 100644 --- a/crates/services/src/completions/mod.rs +++ b/crates/services/src/completions/mod.rs @@ -992,7 +992,7 @@ impl CompletionServiceImpl { ] } - pub fn map_provider_error( + pub(crate) fn map_provider_error( model: &str, error: &inference_providers::CompletionError, operation: &str, diff --git a/crates/services/src/inference_provider_pool/mod.rs b/crates/services/src/inference_provider_pool/mod.rs index 132725e63..6f44f22cc 100644 --- a/crates/services/src/inference_provider_pool/mod.rs +++ b/crates/services/src/inference_provider_pool/mod.rs @@ -3302,25 +3302,33 @@ impl InferenceProviderPool { } } - if let Some(Ok(event)) = peekable.peek().await { - if let Some(inference_providers::StreamChunk::Chat(chat_chunk)) = &event.chunk { - let chat_id = chat_chunk.id.clone(); - tracing::info!( - chat_id = %chat_id, - "Storing chat_id mapping for streaming completion" - ); - // Pin the dedicated TLS connection so signature fetches - // reuse the same connection that served this completion. - provider.pin_chat_connection(&request_hash, &chat_id); - pinned = true; - self.store_chat_id_mapping(chat_id, provider.clone()).await; + let first_error = match peekable.peek().await { + Some(Ok(event)) => { + if let Some(inference_providers::StreamChunk::Chat(chat_chunk)) = &event.chunk { + let chat_id = chat_chunk.id.clone(); + tracing::info!( + chat_id = %chat_id, + "Storing chat_id mapping for streaming completion" + ); + // Pin the dedicated TLS connection so signature fetches + // reuse the same connection that served this completion. + provider.pin_chat_connection(&request_hash, &chat_id); + pinned = true; + self.store_chat_id_mapping(chat_id, provider.clone()).await; + } + None } - } + Some(Err(error)) => Some(error.clone()), + None => None, + }; if !pinned { // Clean up orphaned pending client when peek fails or yields no chat_id provider.pin_chat_connection(&request_hash, ""); provider.unpin_chat_connection(""); } + if let Some(error) = first_error { + return Err(error); + } let stream: StreamingResult = if leading_control.is_empty() { Box::pin(peekable) } else { @@ -6130,6 +6138,84 @@ mod tests { assert!(pool.get_provider_by_chat_id(&chat_id).await.is_some()); } + #[tokio::test] + async fn test_first_stream_error_is_returned_before_stream() { + use inference_providers::mock::MockProvider; + + // Given + let pool = InferenceProviderPool::new(None, ExternalProvidersConfig::default()); + let mock_provider = Arc::new(MockProvider::new()); + let model_id = "Qwen/Qwen3-30B-A3B-Instruct-2507".to_string(); + mock_provider + .set_stream_error_override(Some(CompletionError::HttpError { + status_code: 400, + message: "Grammar error: unsupported schema keyword".to_string(), + is_external: true, + })) + .await; + pool.register_provider(model_id.clone(), mock_provider.clone()) + .await; + let params = inference_providers::ChatCompletionParams { + model: model_id, + messages: vec![inference_providers::ChatMessage { + role: inference_providers::MessageRole::User, + content: Some(serde_json::Value::String("Hello".to_string())), + name: None, + tool_call_id: None, + tool_calls: None, + }], + max_tokens: None, + temperature: None, + top_p: None, + stop: None, + stream: Some(true), + tools: None, + max_completion_tokens: None, + n: None, + frequency_penalty: None, + presence_penalty: None, + logit_bias: None, + logprobs: None, + top_logprobs: None, + user: None, + seed: None, + tool_choice: None, + parallel_tool_calls: None, + metadata: None, + store: None, + stream_options: None, + service_tier: None, + modalities: None, + original_request: None, + extra: std::collections::HashMap::new(), + }; + + // When + let result = pool + .chat_completion_stream( + params, + "test-request-hash".to_string(), + ChatRoutingHints::default(), + ) + .await; + + // Then + match result { + Err(CompletionError::HttpError { + status_code, + message, + is_external, + }) => { + assert_eq!(status_code, 400); + assert_eq!(message, "Grammar error: unsupported schema keyword"); + assert!(is_external); + } + Err(other) => panic!("Expected HttpError, got {other:?}"), + Ok(_) => panic!("Expected the pool to return the first stream error"), + } + assert_eq!(mock_provider.unpinned_chat_ids(), vec![String::new()]); + } + // ==================== Provider Tests ==================== #[tokio::test]