From bf33d8d2f46e15d72c84689ef224bf965df10025 Mon Sep 17 00:00:00 2001 From: whit3rabbit Date: Tue, 24 Mar 2026 06:55:07 -0500 Subject: [PATCH] fix: return proper HTTP status for pre-stream backend errors Previously, backend errors (401, 429, 500) that occurred before any SSE data was sent were wrapped in a 200 OK response with an error event buried in the stream. Now messages_stream returns Result so the caller can respond with the correct HTTP status code. Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/proxy/src/server/routes.rs | 15 ++++-- crates/proxy/src/server/streaming.rs | 69 +++++++++++++--------------- 2 files changed, 43 insertions(+), 41 deletions(-) diff --git a/crates/proxy/src/server/routes.rs b/crates/proxy/src/server/routes.rs index 94988aa..088a89c 100644 --- a/crates/proxy/src/server/routes.rs +++ b/crates/proxy/src/server/routes.rs @@ -335,10 +335,17 @@ async fn messages( } let mapped_model = state.map_model(&body.model); // Logging deferred until stream completes (inside messages_stream tasks). - let (rate_limits, sse) = messages_stream(state, body, ctx, mapped_model).await; - let mut response = sse.into_response(); - rate_limits.inject_anthropic_headers(response.headers_mut()); - return response; + match messages_stream(state, body, ctx, mapped_model).await { + Ok((rate_limits, sse)) => { + let mut response = sse.into_response(); + rate_limits.inject_anthropic_headers(response.headers_mut()); + return response; + } + Err(e) => { + // Pre-stream backend error: return proper HTTP status instead of 200 OK + return backend_error_to_response(e); + } + } } match &state.backend { diff --git a/crates/proxy/src/server/streaming.rs b/crates/proxy/src/server/streaming.rs index c9b688a..70a498f 100644 --- a/crates/proxy/src/server/streaming.rs +++ b/crates/proxy/src/server/streaming.rs @@ -31,27 +31,6 @@ async fn send_events( true } -/// Send an SSE error event over the channel. -/// Logs the detailed error server-side and sends a generic message to the client. -async fn send_stream_error( - tx: &mpsc::Sender>, - metrics: &Metrics, - error: impl std::fmt::Display, -) { - tracing::error!("streaming request failed: {error}"); - metrics.record_error(); - let err_event = anthropic::StreamEvent::Error { - error: anthropic::streaming::StreamError { - error_type: "api_error".to_string(), - message: "An internal error occurred while communicating with the upstream service." - .to_string(), - }, - }; - if let Ok(sse) = super::sse::stream_event_to_sse(&err_event) { - let _ = tx.send(Ok(sse)).await; - } -} - /// Maximum SSE buffer size (10 MB). Protects against unbounded memory growth /// if the backend sends data without frame delimiters. const MAX_SSE_BUFFER_SIZE: usize = 10 * 1024 * 1024; @@ -186,6 +165,8 @@ where /// Build an SSE response that streams Anthropic events translated from backend chunks. /// Returns rate limit headers alongside the SSE stream so the caller can inject them. +/// Pre-stream backend errors (e.g., 401, 429, 500 before any data) are returned as +/// `Err(BackendError)` so the caller can respond with a proper HTTP status code. /// Logging is deferred: each spawned task logs after the stream completes with actual /// latency, status, and token counts. pub(crate) async fn messages_stream( @@ -193,12 +174,16 @@ pub(crate) async fn messages_stream( body: anthropic::MessageCreateRequest, ctx: RequestCtx, mapped_model: String, -) -> ( - RateLimitHeaders, - Sse>>, -) { +) -> Result< + ( + RateLimitHeaders, + Sse>>, + ), + crate::backend::BackendError, +> { let (tx, rx) = mpsc::channel::>(32); - let (rl_tx, rl_rx) = tokio::sync::oneshot::channel::(); + let (rl_tx, rl_rx) = + tokio::sync::oneshot::channel::>(); let metrics = state.metrics.clone(); let log_shared = state.shared.clone(); @@ -217,7 +202,7 @@ pub(crate) async fn messages_stream( tokio::spawn(async move { match client.chat_completion_stream(&openai_req).await { Ok((response, rate_limits)) => { - rl_tx.send(rate_limits).ok(); + rl_tx.send(Ok(rate_limits)).ok(); let mut translator = mapping::streaming_map::StreamingTranslator::new(model); let mut done = false; @@ -260,8 +245,7 @@ pub(crate) async fn messages_stream( Err(e) => { let status = e.status_code(); let err_msg = e.to_string(); - drop(rl_tx); - send_stream_error(&tx, &metrics, e).await; + metrics.record_error(); log_request( &log_shared, ctx.log_entry( @@ -273,6 +257,9 @@ pub(crate) async fn messages_stream( Some(err_msg), ), ); + // Send the error through the oneshot so the caller can + // return a proper HTTP error response instead of 200 OK. + let _ = rl_tx.send(Err(crate::backend::BackendError::from(e))); } } }); @@ -288,7 +275,7 @@ pub(crate) async fn messages_stream( tokio::spawn(async move { match client.responses_stream(&responses_req).await { Ok((response, rate_limits)) => { - rl_tx.send(rate_limits).ok(); + rl_tx.send(Ok(rate_limits)).ok(); let mut translator = mapping::responses_streaming_map::ResponsesStreamingTranslator::new( model, @@ -331,8 +318,7 @@ pub(crate) async fn messages_stream( Err(e) => { let status = e.status_code(); let err_msg = e.to_string(); - drop(rl_tx); - send_stream_error(&tx, &metrics, e).await; + metrics.record_error(); log_request( &log_shared, ctx.log_entry( @@ -344,6 +330,7 @@ pub(crate) async fn messages_stream( Some(err_msg), ), ); + let _ = rl_tx.send(Err(crate::backend::BackendError::from(e))); } } }); @@ -358,9 +345,17 @@ pub(crate) async fn messages_stream( } } - let rate_limits = rl_rx.await.unwrap_or_default(); - ( - rate_limits, - Sse::new(ReceiverStream::new(rx)).keep_alive(KeepAlive::default()), - ) + match rl_rx.await { + Ok(Ok(rate_limits)) => Ok(( + rate_limits, + Sse::new(ReceiverStream::new(rx)).keep_alive(KeepAlive::default()), + )), + Ok(Err(backend_err)) => Err(backend_err), + // Sender dropped without sending (e.g., Anthropic passthrough branch or task panic). + // Default to empty rate limits and let the stream deliver whatever it has. + Err(_) => Ok(( + RateLimitHeaders::default(), + Sse::new(ReceiverStream::new(rx)).keep_alive(KeepAlive::default()), + )), + } }