diff --git a/rust/crates/api/src/providers/anthropic.rs b/rust/crates/api/src/providers/anthropic.rs index 0b9b2f5c75..403cbf557d 100644 --- a/rust/crates/api/src/providers/anthropic.rs +++ b/rust/crates/api/src/providers/anthropic.rs @@ -889,7 +889,7 @@ async fn expect_success(response: reqwest::Response) -> Result(&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 { @@ -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 @@ -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(); @@ -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] diff --git a/rust/crates/api/src/providers/openai_compat.rs b/rust/crates/api/src/providers/openai_compat.rs index 06e0071060..e40a94ec8c 100644 --- a/rust/crates/api/src/providers/openai_compat.rs +++ b/rust/crates/api/src/providers/openai_compat.rs @@ -1763,7 +1763,7 @@ async fn expect_success(response: reqwest::Response) -> Result(&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); @@ -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 @@ -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; @@ -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() {