SemianiakaVY/openpencil
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

fix(ai): add request throttle + 429/503 backoff-retry to builtin http provider

Browse source 
A single provider rate-limit (HTTP 429) or overload (503/529) on one
design sub-agent request had no recovery: no client-side pacing to stay
under the RPM limit, and no backoff-retry once tripped — so the section
burned its attempts and the run reported a failed subtask while the rest
of the design was fine. Add a process-wide min-gap throttle (default
350ms, env-overridable) and transparent backoff-retry that honors
Retry-After (else exponential 1/2/4s, capped) around both the openai-
compatible and anthropic send paths. Benefits every builtin path
(orchestrator, design loop, chat). retry.rs stays unchanged: 429 remains
non-retryable at the ladder level since the http layer already backed off.
This commit is contained in:
Fini 2026-07-05 16:52:01 +08:00
parent 20606cd632
commit 768ea37d7b
1 changed files with 185 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

203
crates/op-host-services/src/chat_builtin_http.rs
View file

@ -6,7 +6,8 @@
//! streaming endpoints and converts SSE payloads into `ChatDelta`s.
use std::fmt;
use std::sync::Arc;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use futures::StreamExt;
use op_ai::chat_provider::{
@ -32,6 +33,13 @@ pub const DESIGN_LOOP_MAX_TURNS: usize = 28;
/// emit a whole screen in one turn.
pub const DESIGN_LOOP_MAX_OUTPUT_TOKENS: u32 = 6_144;
const BUILTIN_HTTP_DEFAULT_MIN_GAP: Duration = Duration::from_millis(350);
const BUILTIN_HTTP_MAX_RETRIES: u32 = 3;
const RETRY_AFTER_MAX: Duration = Duration::from_secs(30);
const BACKOFF_MAX: Duration = Duration::from_secs(8);
static BUILTIN_HTTP_LAST_REQUEST: OnceLock<Mutex<Option<Instant>>> = OnceLock::new();
#[derive(Clone)]
pub struct ConfiguredBuiltinProvider {
kind: BuiltinAgentKind,
@ -267,6 +275,96 @@ pub(crate) fn is_glm_model(model: &str) -> bool {
model.to_ascii_lowercase().contains("glm")
}
fn builtin_http_min_gap() -> Duration {
std::env::var("OPENPENCIL_BUILTIN_HTTP_MIN_GAP_MS")
.ok()
.and_then(|value| value.parse::<u64>().ok())
.map(Duration::from_millis)
.unwrap_or(BUILTIN_HTTP_DEFAULT_MIN_GAP)
}
fn throttle_wait(last: Option<Instant>, now: Instant, min_gap: Duration) -> Duration {
last.and_then(|last| last.checked_add(min_gap))
.map(|next_allowed| next_allowed.saturating_duration_since(now))
.unwrap_or(Duration::ZERO)
}
async fn throttle_builtin_http_request() {
let min_gap = builtin_http_min_gap();
if min_gap.is_zero() {
return;
}
let wait = {
let last_request = BUILTIN_HTTP_LAST_REQUEST.get_or_init(|| Mutex::new(None));
let mut last = last_request
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let now = Instant::now();
let wait = throttle_wait(*last, now, min_gap);
*last = now.checked_add(wait);
wait
};
if !wait.is_zero() {
tokio::time::sleep(wait).await;
}
}
fn is_retryable_status(status: reqwest::StatusCode) -> bool {
status == reqwest::StatusCode::TOO_MANY_REQUESTS
|| status == reqwest::StatusCode::SERVICE_UNAVAILABLE
|| status == reqwest::StatusCode::from_u16(529).expect("529 is a valid HTTP status")
}
fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> Option<Duration> {
let seconds = headers
.get(reqwest::header::RETRY_AFTER)?
.to_str()
.ok()?
.parse::<u64>()
.ok()?;
Some(Duration::from_secs(seconds).min(RETRY_AFTER_MAX))
}
fn backoff_delay(attempt: u32) -> Duration {
let delay = 1_u64.checked_shl(attempt).unwrap_or(u64::MAX);
Duration::from_secs(delay).min(BACKOFF_MAX)
}
async fn send_with_backoff(
label: &str,
url: &str,
build: impl Fn() -> reqwest::RequestBuilder,
) -> Result<reqwest::Response, String> {
for attempt in 0..=BUILTIN_HTTP_MAX_RETRIES {
throttle_builtin_http_request().await;
match build().send().await {
Ok(resp) if resp.status().is_success() => return Ok(resp),
Ok(resp) => {
let status = resp.status();
if is_retryable_status(status) && attempt < BUILTIN_HTTP_MAX_RETRIES {
let delay =
parse_retry_after(resp.headers()).unwrap_or_else(|| backoff_delay(attempt));
drop(resp);
tokio::time::sleep(delay).await;
continue;
}
let body = resp.text().await.unwrap_or_default();
return Err(format!("{label} http {status}: {}", body.trim()));
}
Err(e) => {
if attempt < BUILTIN_HTTP_MAX_RETRIES {
tokio::time::sleep(backoff_delay(attempt)).await;
continue;
}
return Err(format!("{label} POST {url}: {e}"));
}
}
}
unreachable!("backoff loop always returns before exhausting range")
}
async fn run_openai_chat(
provider: ConfiguredBuiltinProvider,
system_prompt: String,
@ -310,14 +408,13 @@ async fn run_openai_chat(
obj.insert("thinking".into(), json!({ "type": "disabled" }));
}
}
let resp = builtin_http_client()
.post(&url)
.bearer_auth(&provider.api_key)
.json(&body)
.send()
.await
.map_err(|e| format!("openai-compatible POST {url}: {e}"))?;
let resp = ensure_success(resp, "openai-compatible").await?;
let resp = send_with_backoff("openai-compatible", &url, || {
builtin_http_client()
.post(&url)
.bearer_auth(&provider.api_key)
.json(&body)
})
.await?;
pump_sse_response(resp, tx, parse_openai_sse_data).await
}
@ -364,15 +461,14 @@ async fn run_anthropic_chat(
.expect("anthropic request body is object")
.insert("system".into(), json!(system_prompt));
}
let resp = builtin_http_client()
.post(&url)
.header("x-api-key", &provider.api_key)
.header("anthropic-version", "2023-06-01")
.json(&body)
.send()
.await
.map_err(|e| format!("anthropic POST {url}: {e}"))?;
let resp = ensure_success(resp, "anthropic").await?;
let resp = send_with_backoff("anthropic", &url, || {
builtin_http_client()
.post(&url)
.header("x-api-key", &provider.api_key)
.header("anthropic-version", "2023-06-01")
.json(&body)
})
.await?;
pump_sse_response(resp, tx, parse_anthropic_sse_data).await
}
@ -645,6 +741,77 @@ mod tests {
assert!(!is_minimax_model("ark-code-latest"));
}
#[test]
fn is_retryable_status_flags_provider_rate_limit_and_overload() {
assert!(is_retryable_status(reqwest::StatusCode::TOO_MANY_REQUESTS));
assert!(is_retryable_status(
reqwest::StatusCode::SERVICE_UNAVAILABLE
));
assert!(is_retryable_status(
reqwest::StatusCode::from_u16(529).expect("status 529")
));
assert!(!is_retryable_status(reqwest::StatusCode::OK));
assert!(!is_retryable_status(reqwest::StatusCode::BAD_REQUEST));
assert!(!is_retryable_status(reqwest::StatusCode::UNAUTHORIZED));
assert!(!is_retryable_status(reqwest::StatusCode::NOT_FOUND));
assert!(!is_retryable_status(
reqwest::StatusCode::UNAVAILABLE_FOR_LEGAL_REASONS
));
}
#[test]
fn parse_retry_after_accepts_integer_seconds_and_caps_large_values() {
let mut headers = reqwest::header::HeaderMap::new();
assert_eq!(parse_retry_after(&headers), None);
headers.insert(
reqwest::header::RETRY_AFTER,
reqwest::header::HeaderValue::from_static("3"),
);
assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(3)));
headers.insert(
reqwest::header::RETRY_AFTER,
reqwest::header::HeaderValue::from_static("0"),
);
assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(0)));
headers.insert(
reqwest::header::RETRY_AFTER,
reqwest::header::HeaderValue::from_static("abc"),
);
assert_eq!(parse_retry_after(&headers), None);
headers.insert(
reqwest::header::RETRY_AFTER,
reqwest::header::HeaderValue::from_static("3600"),
);
assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(30)));
}
#[test]
fn backoff_delay_exponentially_increases_and_caps() {
assert_eq!(backoff_delay(0), Duration::from_secs(1));
assert_eq!(backoff_delay(1), Duration::from_secs(2));
assert_eq!(backoff_delay(2), Duration::from_secs(4));
assert_eq!(backoff_delay(3), Duration::from_secs(8));
assert_eq!(backoff_delay(99), Duration::from_secs(8));
}
#[test]
fn throttle_wait_respects_reserved_last_request_slot() {
let now = std::time::Instant::now();
let min_gap = Duration::from_millis(350);
assert_eq!(throttle_wait(None, now, min_gap), Duration::ZERO);
assert_eq!(throttle_wait(Some(now), now, min_gap), min_gap);
assert_eq!(
throttle_wait(Some(now - Duration::from_millis(400)), now, min_gap),
Duration::ZERO
);
}
#[test]
fn openai_sse_error_finishes_aborted() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind local SSE server");