Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 89 additions & 2 deletions rust/crates/api/src/providers/anthropic.rs
Original file line number Diff line number Diff line change
Expand Up @@ -889,7 +889,7 @@ async fn expect_success(response: reqwest::Response) -> Result<reqwest::Response
let request_id = request_id_from_headers(&headers).map(String::into_boxed_str);
let body = response.text().await.unwrap_or_else(|_| String::new());
let parsed_error = serde_json::from_str::<AnthropicErrorEnvelope>(&body).ok();
let retryable = is_retryable_status(status);
let retryable = is_retryable_status(status) || is_retryable_400(status, &body);
let retry_after = parse_retry_after(&headers, status).map(Box::new);

Err(ApiError::Api {
Expand Down Expand Up @@ -923,7 +923,10 @@ fn parse_retry_after(
}

const fn is_retryable_status(status: reqwest::StatusCode) -> bool {
matches!(status.as_u16(), 408 | 409 | 429 | 500 | 502 | 503 | 504)
matches!(
status.as_u16(),
408 | 409 | 429 | 500 | 502 | 503 | 504 | 529
)
}

/// Some providers return HTTP 400 with an unparseable body when a gateway
Expand Down Expand Up @@ -1146,6 +1149,56 @@ mod tests {
format!("http://{address}/oauth/token")
}

fn spawn_status_server(status: u16, body: &'static str) -> String {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind listener");
let address = listener.local_addr().expect("local addr");
thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept connection");
let mut buffer = [0_u8; 4096];
let _ = stream.read(&mut buffer).expect("read request");
let response = format!(
"HTTP/1.1 {status} STATUS\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\r\n{}",
body.len(),
body
);
stream
.write_all(response.as_bytes())
.expect("write response");
});
format!("http://{address}/v1/messages")
}

#[tokio::test]
async fn expect_success_marks_529_overloaded_retryable() {
let url = spawn_status_server(
529,
r#"{"error":{"type":"overloaded_error","message":"Overloaded"}}"#,
);
let response = reqwest::get(&url).await.expect("response");
let error = super::expect_success(response)
.await
.expect_err("529 should error");
assert!(error.is_retryable(), "529 overloaded must be retryable");
}

#[tokio::test]
async fn expect_success_marks_gateway_flake_400_retryable() {
let url = spawn_status_server(400, "HTTP 400 from backend (no parseable body)");
let response = reqwest::get(&url).await.expect("response");
let error = super::expect_success(response)
.await
.expect_err("400 should error");
assert!(
error.is_retryable(),
"gateway-flake 400 body must be retryable"
);
let url = spawn_status_server(400, "invalid request: max_tokens must be positive");
let response = reqwest::get(&url).await.expect("response");
let error = super::expect_success(response)
.await
.expect_err("400 should error");
assert!(!error.is_retryable(), "a real 400 must not be retried");
}
#[test]
fn read_api_key_requires_presence() {
let _guard = env_lock();
Expand Down Expand Up @@ -1445,6 +1498,40 @@ mod tests {
assert!(!super::is_retryable_status(
reqwest::StatusCode::UNAUTHORIZED
));
// Anthropic returns the non-standard 529 when the API is overloaded;
// it is transient and must be retried like a 503.
assert!(super::is_retryable_status(
reqwest::StatusCode::from_u16(529).expect("529 is a valid status")
));
assert!(!super::is_retryable_status(
reqwest::StatusCode::BAD_REQUEST
));
}

#[test]
fn gateway_flake_400_bodies_are_retryable() {
// Gateways and proxies surface transient network failures as HTTP 400
// with an unparseable body; is_retryable_400 detects them so
// expect_success marks the resulting error retryable.
for body in [
"HTTP 400 from backend (no parseable body)",
"upstream connection reset by peer",
"Broken pipe while reading response",
"Empty reply from server",
] {
assert!(
super::is_retryable_400(reqwest::StatusCode::BAD_REQUEST, body),
"body should be retryable: {body}"
);
}
assert!(!super::is_retryable_400(
reqwest::StatusCode::BAD_REQUEST,
"invalid request: max_tokens must be positive"
));
assert!(!super::is_retryable_400(
reqwest::StatusCode::INTERNAL_SERVER_ERROR,
"connection reset"
));
}

#[test]
Expand Down
104 changes: 99 additions & 5 deletions rust/crates/api/src/providers/openai_compat.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1763,7 +1763,7 @@ async fn expect_success(response: reqwest::Response) -> Result<reqwest::Response
let request_id = request_id_from_headers(&headers).map(String::into_boxed_str);
let body = response.text().await.unwrap_or_default();
let parsed_error = serde_json::from_str::<ErrorEnvelope>(&body).ok();
let retryable = is_retryable_status(status);
let retryable = is_retryable_status(status) || is_retryable_400(status, &body);
let retry_after = parse_retry_after(&headers, status).map(Box::new);

let suggested_action = suggested_action_for_status(status);
Expand Down Expand Up @@ -1799,7 +1799,10 @@ fn parse_retry_after(
}

const fn is_retryable_status(status: reqwest::StatusCode) -> bool {
matches!(status.as_u16(), 408 | 409 | 429 | 500 | 502 | 503 | 504)
matches!(
status.as_u16(),
408 | 409 | 429 | 500 | 502 | 503 | 504 | 529
)
}

/// Some providers return HTTP 400 with an unparseable body when a gateway
Expand Down Expand Up @@ -1858,9 +1861,9 @@ impl StringExt for String {
mod tests {
use super::{
build_chat_completion_request, chat_completions_endpoint, is_reasoning_model,
model_requires_reasoning_content_in_history, normalize_finish_reason, normalize_response,
openai_tool_choice, parse_tool_arguments, OpenAiCompatClient, OpenAiCompatConfig,
StreamState,
is_retryable_400, is_retryable_status, model_requires_reasoning_content_in_history,
normalize_finish_reason, normalize_response, openai_tool_choice, parse_tool_arguments,
OpenAiCompatClient, OpenAiCompatConfig, StreamState,
};
use crate::error::ApiError;
use crate::test_env_lock as env_lock;
Expand All @@ -1872,6 +1875,97 @@ mod tests {
use serde_json::json;
use std::borrow::Cow;
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::TcpListener;
use std::thread;

fn spawn_status_server(status: u16, body: &'static str) -> String {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind listener");
let address = listener.local_addr().expect("local addr");
thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept connection");
let mut buffer = [0_u8; 4096];
let _ = stream.read(&mut buffer).expect("read request");
let response = format!(
"HTTP/1.1 {status} STATUS\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\r\n{}",
body.len(),
body
);
stream
.write_all(response.as_bytes())
.expect("write response");
});
format!("http://{address}/v1/messages")
}

#[tokio::test]
async fn expect_success_marks_529_overloaded_retryable() {
let url = spawn_status_server(
529,
r#"{"error":{"type":"overloaded_error","message":"Overloaded"}}"#,
);
let response = reqwest::get(&url).await.expect("response");
let error = super::expect_success(response)
.await
.expect_err("529 should error");
assert!(error.is_retryable(), "529 overloaded must be retryable");
}

#[tokio::test]
async fn expect_success_marks_gateway_flake_400_retryable() {
let url = spawn_status_server(400, "HTTP 400 from backend (no parseable body)");
let response = reqwest::get(&url).await.expect("response");
let error = super::expect_success(response)
.await
.expect_err("400 should error");
assert!(
error.is_retryable(),
"gateway-flake 400 body must be retryable"
);
let url = spawn_status_server(400, "invalid request: max_tokens must be positive");
let response = reqwest::get(&url).await.expect("response");
let error = super::expect_success(response)
.await
.expect_err("400 should error");
assert!(!error.is_retryable(), "a real 400 must not be retried");
}
#[test]
fn retryable_statuses_are_detected() {
assert!(is_retryable_status(reqwest::StatusCode::TOO_MANY_REQUESTS));
assert!(is_retryable_status(
reqwest::StatusCode::INTERNAL_SERVER_ERROR
));
assert!(!is_retryable_status(reqwest::StatusCode::UNAUTHORIZED));
// Several OpenAI-compatible gateways (and Anthropic) signal overload
// with the non-standard 529; it is transient and must be retried.
assert!(is_retryable_status(
reqwest::StatusCode::from_u16(529).expect("529 is a valid status")
));
assert!(!is_retryable_status(reqwest::StatusCode::BAD_REQUEST));
}

#[test]
fn gateway_flake_400_bodies_are_retryable() {
for body in [
"HTTP 400 from backend (no parseable body)",
"upstream connection reset by peer",
"Broken pipe while reading response",
"Empty reply from server",
] {
assert!(
is_retryable_400(reqwest::StatusCode::BAD_REQUEST, body),
"body should be retryable: {body}"
);
}
assert!(!is_retryable_400(
reqwest::StatusCode::BAD_REQUEST,
"invalid request: max_tokens must be positive"
));
assert!(!is_retryable_400(
reqwest::StatusCode::INTERNAL_SERVER_ERROR,
"connection reset"
));
}

#[test]
fn request_translation_uses_openai_compatible_shape() {
Expand Down
Loading