From ee4e20046dc51294fb1c875947afb9130f4bdc12 Mon Sep 17 00:00:00 2001 From: user Date: Sun, 30 Aug 2026 11:26:52 +0800 Subject: [PATCH 1/3] fix(ai): don't retry deterministic 4xx statuses The retry loop in execute_sse_request retried every non-2xx response with the same request body, including deterministic client errors such as 400. A deterministic 4xx fails identically on every attempt, so retrying it only burns the request budget/credits and adds latency without any chance of success; the bad_requests_then_success fixture even reported a "success" after a later request happened to return 200, masking the wasted retries. Add a module-private is_transient_http_status predicate that classifies server errors (5xx), rate limiting (429), and request/gateway timeouts (408/504) as transient, and treats every other status as terminal. The non-2xx branch now breaks out of the loop for a terminal status instead of continuing; only transient statuses retry, reusing the existing retry_delay_ms / exponential backoff. The successful-response path, the transport branch, and the TTFT branch are left untouched by this commit. Test: reversed bad_requests_then_success coverage to assert terminal 400 (attempts == 1) and added is_transient_http_status classification coverage; `cargo test -p bitfun-ai-adapters --jobs 4` passes. AI: generated with review; verified with the above command. --- .../adapters/ai-adapters/src/client/sse.rs | 62 +++++++++++++++++-- 1 file changed, 56 insertions(+), 6 deletions(-) diff --git a/src/crates/adapters/ai-adapters/src/client/sse.rs b/src/crates/adapters/ai-adapters/src/client/sse.rs index 560f1da90e..f588fa8a3c 100644 --- a/src/crates/adapters/ai-adapters/src/client/sse.rs +++ b/src/crates/adapters/ai-adapters/src/client/sse.rs @@ -189,6 +189,19 @@ fn retry_delay_ms(attempt: usize, headers: &HeaderMap, status: StatusCode) -> u6 } } +/// Returns true when `status` represents a transient condition that may succeed +/// on a later attempt: server errors (5xx), rate limiting (429), and request or +/// gateway timeouts (408/504). +/// +/// Deterministic client errors (400/401/403/404/413/422) are excluded because +/// retrying them reproduces the same failure and burns request budget/credits. +fn is_transient_http_status(status: StatusCode) -> bool { + status.is_server_error() + || status == StatusCode::TOO_MANY_REQUESTS + || status == StatusCode::REQUEST_TIMEOUT + || status == StatusCode::GATEWAY_TIMEOUT +} + struct ManagedResponseStream { inner: UnboundedReceiverStream>, handler_cancel: CancellationToken, @@ -320,7 +333,7 @@ where .await; } - if attempt < max_tries - 1 { + if attempt < max_tries - 1 && is_transient_http_status(status) { let delay_ms = retry_delay_ms(attempt, &headers, status); debug!( "Retrying {} after {}ms (transport_attempt {}, status {})", @@ -331,7 +344,7 @@ where ); tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; } - continue; + break; } } StreamSendOutcome::Transport(e) => { @@ -534,6 +547,41 @@ mod tests { assert!(message.contains("first effective stream output")); } + #[test] + fn is_transient_http_status_classifies_terminal_and_transient() { + // Deterministic client errors are terminal and must not be retried. + for terminal in [ + StatusCode::BAD_REQUEST, + StatusCode::UNAUTHORIZED, + StatusCode::FORBIDDEN, + StatusCode::NOT_FOUND, + StatusCode::PAYLOAD_TOO_LARGE, + StatusCode::UNPROCESSABLE_ENTITY, + ] { + assert!( + !is_transient_http_status(terminal), + "{} should be terminal", + terminal + ); + } + + // Transient conditions are retried: server errors, rate limit, and timeouts. + for transient in [ + StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::BAD_GATEWAY, + StatusCode::SERVICE_UNAVAILABLE, + StatusCode::GATEWAY_TIMEOUT, + StatusCode::TOO_MANY_REQUESTS, + StatusCode::REQUEST_TIMEOUT, + ] { + assert!( + is_transient_http_status(transient), + "{} should be transient", + transient + ); + } + } + #[test] fn remaining_ttft_timeout_subtracts_elapsed_request_time() { let start = std::time::Instant::now() - Duration::from_secs(2); @@ -568,7 +616,7 @@ mod tests { } #[tokio::test] - async fn every_bad_request_uses_existing_retry_loop() { + async fn bad_requests_are_terminal_and_not_retried() { let attempts = Arc::new(AtomicUsize::new(0)); let app = Router::new() .route("/chat/completions", post(bad_requests_then_success)) @@ -603,11 +651,13 @@ mod tests { .await; server_task.abort(); + // Deterministic 4xx (400) must be terminal: the request is not retried + // with the same body, so the fixture is called exactly once. assert!( - result.is_ok(), - "ordinary and context-overflow 400 responses should both retry" + result.is_err(), + "deterministic 400 responses should be terminal and not retried" ); - assert_eq!(attempts.load(Ordering::SeqCst), 3); + assert_eq!(attempts.load(Ordering::SeqCst), 1); } #[tokio::test] From 5319e9c541c5e52e436e64787bde20b47180cfcb Mon Sep 17 00:00:00 2001 From: user Date: Sun, 30 Aug 2026 11:28:47 +0800 Subject: [PATCH 2/3] fix(ai): don't resend requests on TTFT timeout The retry loop in execute_sse_request continued on a TTFT timeout using the same request_body. A TTFT (time-to-first-token) timeout fires after the request body has already been sent to the server and the client is waiting for the first token, so re-sending the same body in a later attempt re-sends an already-billed request (double-billing for the same logical turn). Treat a TTFT timeout as a terminal error: break out of the retry loop and surface the timeout to the caller instead of re-sending the request. The existing warn!/last_error/trace reporting is preserved, and the transport and response branches are untouched by this commit. Test: added ttft_timeout_is_terminal coverage using a hanging mock (no completed response) that asserts a single attempt and a terminal error; `cargo test -p bitfun-ai-adapters --jobs 4` passes. AI: generated with review; verified with the above command. --- .../adapters/ai-adapters/src/client/sse.rs | 69 ++++++++++++++++--- 1 file changed, 58 insertions(+), 11 deletions(-) diff --git a/src/crates/adapters/ai-adapters/src/client/sse.rs b/src/crates/adapters/ai-adapters/src/client/sse.rs index f588fa8a3c..9c2e362dc5 100644 --- a/src/crates/adapters/ai-adapters/src/client/sse.rs +++ b/src/crates/adapters/ai-adapters/src/client/sse.rs @@ -399,17 +399,12 @@ where .await; } - if attempt < max_tries - 1 { - let delay_ms = exponential_retry_delay_ms(attempt); - debug!( - "Retrying {} after {}ms (transport_attempt {})", - label, - delay_ms, - attempt + 2 - ); - tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; - } - continue; + // The request body was already sent to the server before the TTFT + // timeout fired (send() completed and the first-token wait began). + // Retrying with the same body re-sends an already-billed request + // (double-billing). Treat it as a terminal error: break out of the + // retry loop and surface the timeout to the caller. + break; } }; @@ -490,6 +485,16 @@ mod tests { } } + async fn hanging_until_timeout( + State(state): State, + Json(_body): Json, + ) -> impl IntoResponse { + state.attempts.fetch_add(1, Ordering::SeqCst); + // Never complete the response so the client's send() future blocks and the + // injected ttft_timeout fires as StreamSendOutcome::TtftTimeout. + std::future::pending::().await + } + async fn forbidden_with_retry_after(Json(body): Json) -> impl IntoResponse { assert_eq!(body["model"], "configured-model"); ( @@ -660,6 +665,48 @@ mod tests { assert_eq!(attempts.load(Ordering::SeqCst), 1); } + #[tokio::test] + async fn ttft_timeout_is_terminal_and_not_retried() { + let attempts = Arc::new(AtomicUsize::new(0)); + let app = Router::new() + .route("/chat/completions", post(hanging_until_timeout)) + .with_state(RetryFixtureState { + attempts: Arc::clone(&attempts), + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ttft fixture"); + let address = listener.local_addr().expect("ttft fixture address"); + let server_task = tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("ttft fixture should run"); + }); + let url = format!("http://{address}/chat/completions"); + let client = reqwest::Client::new(); + let request_body = serde_json::json!({"model": "configured-model"}); + + let result = execute_sse_request( + "OpenAI Streaming API", + &url, + &request_body, + 3, + Some(Duration::from_millis(100)), + None, + || client.post(&url), + |_response, tx, _tx_raw, _remaining_ttft_timeout| async move { + drop(tx); + }, + ) + .await; + + server_task.abort(); + // A TTFT timeout means the request was already sent; it must be terminal + // (no re-send of the same body) so the fixture is called exactly once. + assert!(result.is_err(), "TTFT timeout should be terminal"); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + } + #[tokio::test] async fn single_attempt_preserves_retry_after_metadata_for_outer_budget() { let app = Router::new().route("/chat/completions", post(forbidden_with_retry_after)); From 47689a0409cd04c1b4da113ef770153b747bb1d6 Mon Sep 17 00:00:00 2001 From: user Date: Sun, 30 Aug 2026 12:07:02 +0800 Subject: [PATCH 3/3] fix(ai): retry transient statuses after sleep The Response non-2xx branch computed the retry delay and slept for transient statuses (5xx/429/408), then unconditionally fell through to break. Because the sleep was inside an if that did not jump back to the top of the loop, every non-2xx response exited the retry loop -- even transient server errors. This regressed the intended "only retry transient statuses" semantics and silently removed automatic retries for 5xx/429/408 on the aggregation/compression chains. After sleeping for a transient status, continue the retry loop so the next attempt can succeed; only deterministic 4xx (and the no-retries- remaining case) fall through to the terminal break. The TTFT branch, the transport branch, and the successful-response path are untouched by this commit. Test: added transient_server_errors_are_retried coverage using a mock that returns 503 then 200 and asserts the fixture is called more than once (attempts > 1); existing terminal assertions (4xx and TTFT timeout) still pass; cargo test -p bitfun-ai-adapters --jobs 4 sse::tests passes (16 passed; 0 failed). AI: generated with review; verified with the above command. --- .../adapters/ai-adapters/src/client/sse.rs | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/src/crates/adapters/ai-adapters/src/client/sse.rs b/src/crates/adapters/ai-adapters/src/client/sse.rs index 9c2e362dc5..5be617008d 100644 --- a/src/crates/adapters/ai-adapters/src/client/sse.rs +++ b/src/crates/adapters/ai-adapters/src/client/sse.rs @@ -343,6 +343,10 @@ where status ); tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; + // Transient statuses (5xx/429/408) may succeed on a later + // attempt, so retry rather than treating this response as a + // terminal failure. + continue; } break; } @@ -485,6 +489,27 @@ mod tests { } } + async fn server_errors_then_success( + State(state): State, + Json(body): Json, + ) -> impl IntoResponse { + assert_eq!(body["model"], "configured-model"); + match state.attempts.fetch_add(1, Ordering::SeqCst) { + 0 => ( + StatusCode::SERVICE_UNAVAILABLE, + Json(serde_json::json!({ + "error": { + "message": "temporary server overload", + "type": "server_error", + "code": "server_error" + } + })), + ) + .into_response(), + _ => StatusCode::OK.into_response(), + } + } + async fn hanging_until_timeout( State(state): State, Json(_body): Json, @@ -665,6 +690,54 @@ mod tests { assert_eq!(attempts.load(Ordering::SeqCst), 1); } + #[tokio::test] + async fn transient_server_errors_are_retried() { + let attempts = Arc::new(AtomicUsize::new(0)); + let app = Router::new() + .route("/chat/completions", post(server_errors_then_success)) + .with_state(RetryFixtureState { + attempts: Arc::clone(&attempts), + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind transient fixture"); + let address = listener.local_addr().expect("transient fixture address"); + let server_task = tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("transient fixture should run"); + }); + let url = format!("http://{address}/chat/completions"); + let client = reqwest::Client::new(); + let request_body = serde_json::json!({"model": "configured-model"}); + + let result = execute_sse_request( + "OpenAI Streaming API", + &url, + &request_body, + 2, + None, + None, + || client.post(&url), + |_response, tx, _tx_raw, _remaining_ttft_timeout| async move { + drop(tx); + }, + ) + .await; + + server_task.abort(); + // A transient server error (503) must be retried before the request + // succeeds, so the fixture is expected to be called more than once. + assert!( + result.is_ok(), + "transient server error should be retried and eventually succeed" + ); + assert!( + attempts.load(Ordering::SeqCst) > 1, + "transient 5xx should be retried rather than treated as terminal" + ); + } + #[tokio::test] async fn ttft_timeout_is_terminal_and_not_retried() { let attempts = Arc::new(AtomicUsize::new(0));