starred/anyllm-proxy
1
0
Fork 0
mirror of https://github.com/whit3rabbit/anyllm-proxy.git synced 2026-09-22 00:00:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: return proper HTTP status for pre-stream backend errors

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
whit3rabbit
2026-03-24 06:55:07 -05:00
co-authored by Claude Opus 4.6
parent a0527551cb
commit bf33d8d2f4
2 changed files with 43 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/proxy/src/server/routes.rs
+11 -4
View File
@@ -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 {
crates/proxy/src/server/streaming.rs
+32 -37
View File
@@ -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<Result<Event, std::convert::Infallible>>,
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<impl Stream<Item = Result<Event, std::convert::Infallible>>>,
) {
) -> Result<
(
RateLimitHeaders,
Sse<impl Stream<Item = Result<Event, std::convert::Infallible>>>,
),
crate::backend::BackendError,
> {
let (tx, rx) = mpsc::channel::<Result<Event, std::convert::Infallible>>(32);
let (rl_tx, rl_rx) = tokio::sync::oneshot::channel::<RateLimitHeaders>();
let (rl_tx, rl_rx) =
tokio::sync::oneshot::channel::<Result<RateLimitHeaders, crate::backend::BackendError>>();
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()),
)),
}
}