diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b364779..ba01265f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,74 @@ target. ## [Unreleased] +### Read before upgrading + +- **A request whose upstreams fail for the moment can take up to two + minutes to fail.** When every route of a model failed with an overload + or a server error, or could not be reached, before any of the answer + reached the client, the gateway now waits and sends the request again + instead of returning the error at once, for up to + `gateway.retry_wait_secs` (120 seconds by default) after the request + arrived. Clients and load balancers with a shorter timeout than that may + give up first. Set `gateway.retry_wait_secs` to 0 to keep the earlier + behaviour. +- **A whole answer that runs into the request timeout is now an error + with a log row.** A non-streaming request's upstream attempt is cut + 2 seconds before the gateway's 120-second request timeout (counted from + when the request reached the gateway port), and the client gets the + last upstream error, or a 504 `ProviderTimeout` when there was none, + with its `gateway_logs` row, instead of an empty 408. A retry starts + only when its wait and another attempt as long as the last one fit + before then. +- **A stream's opening events arrive with its first content.** While a + stream can still be sent again, what the upstream sends before any + content (Anthropic `message_start` and `ping`, Responses + `response.created` and `response.in_progress`, a Chat chunk with only + the role) is held back and goes out together with the first content. + While it is held, and while the gateway waits to send again, the client + receives an SSE comment (`: keep-alive`) after every 15 seconds without + data — except Gemini clients, which never receive one. The status is + 200 as before. + +### Added + +- **Transient upstream failures are retried.** When every candidate route + has been tried and the last one failed for the moment — 500, 502, 503, + 504 or 529, a connection that could not be made or broke before the + answer's head, a stream that broke off or reported an overload or a + server error (`overloaded_error`, `api_error`, `server_error`, + `server_is_overloaded`, `slow_down`, Gemini `UNAVAILABLE`) before any + content — and none of the answer reached the client, the request waits + 2, 4, 8, then 16 seconds (each give or take 20%), or as long as the + upstream said in `retry-after-ms` or `Retry-After` (a zero wait is + ignored, and a shorter one than 250 ms is taken as 250 ms), and goes out again + to the routes that failed that way. A model with one route, and a + stream, goes back to the same route. Each attempt is picked as before: + the strategy, the circuit breaker, the route caps, session affinity + and the provider holding a conversation's encrypted history all apply, + and each failed attempt counts against its route's breaker as it did. + Retries stop when the next wait would end past the deadline; an attempt + already under way is not cut short. Rate limits and spent quota (429), + refused credentials (401, 403), any other 4xx, timeouts, the gateway's + own limits, budgets and policies, and a client that left are never + retried, and neither is a stream after part of its answer went out. +- **`gateway.retry_wait_secs`** (Settings → Gateway → Routing strategy): + how long after a request arrives it may still be sent again, 0 to 600 + seconds, default 120; 0 turns retries off. +- **`gateway_provider_retry_total{provider}`**: attempts sent again after + a transient failure, by the provider they went to, next to + `gateway_provider_fallback_total`. +- The `gateway_logs` row of a request that was sent again records + `"retries": n` in its detail, whether it then succeeded or failed. + +### Changed + +- A provider whose base URL cannot be made into a request fails as + `ProviderError` instead of `NetworkError`, and is not retried. +- An error a stream reports as OpenAI's `server_error` is logged as the + upstream's 500 (`ProviderHttpError`), as the same error answered with + HTTP 500 is, instead of 502 (`ProviderError`). + ## [3.9.1] — 2026-10-23 A conversation whose history a provider encrypted, such as a Codex diff --git a/crates/common/src/dynamic_config.rs b/crates/common/src/dynamic_config.rs index ed044a87..3a777b7d 100644 --- a/crates/common/src/dynamic_config.rs +++ b/crates/common/src/dynamic_config.rs @@ -216,6 +216,12 @@ macro_rules! dc_getters_u64_from_i64 { }; } +/// `gateway.retry_wait_secs` when the setting is missing. +pub const RETRY_WAIT_DEFAULT_SECS: i64 = 120; +/// Upper bound on `gateway.retry_wait_secs`, enforced both at write time +/// (the settings handler) and at read time. +pub const RETRY_WAIT_MAX_SECS: i64 = 600; + /// Upper bound on `security.signature_drift_secs`, enforced both at /// write time (admin settings validation) and at read time (the /// getter below). Anything beyond five minutes weakens replay @@ -440,6 +446,18 @@ impl DynamicConfig { raw.clamp(0, 86400) } + /// How long after it arrived a request may still be sent again to + /// upstreams that failed for the moment (overloaded, a server error, + /// a broken connection) before any of the answer reached the client. + /// Clamped to [0, 600]; 0 never sends a request again. + pub async fn retry_wait_secs(&self) -> u64 { + let raw = self + .get_i64("gateway.retry_wait_secs") + .await + .unwrap_or(RETRY_WAIT_DEFAULT_SECS); + raw.clamp(0, RETRY_WAIT_MAX_SECS) as u64 + } + /// Exponent for the `latency` strategy: higher = more aggressive /// (heavier penalty on slow routes). Default 2.0 (aggressive). /// Clamped to [0.5, 8.0] so a typo can't make weights collapse diff --git a/crates/gateway/Cargo.toml b/crates/gateway/Cargo.toml index 9d138611..da2b5e87 100644 --- a/crates/gateway/Cargo.toml +++ b/crates/gateway/Cargo.toml @@ -40,3 +40,4 @@ rust_decimal = { workspace = true } [dev-dependencies] http_1x = { package = "http", version = "1" } +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/gateway/src/error.rs b/crates/gateway/src/error.rs index 03b3b44b..fa771ff5 100644 --- a/crates/gateway/src/error.rs +++ b/crates/gateway/src/error.rs @@ -12,9 +12,17 @@ pub enum GatewayError { /// Upstream returned a non-2xx, non-429, non-401 status. The /// status is kept structured so error-classifier metrics stay /// readable and the gateway can classify retry-eligible 5xx - /// versus poison 4xx without parsing the message. + /// versus poison 4xx without parsing the message. `retry_after` is + /// when the upstream said to try again (`retry-after-ms` or + /// `Retry-After`), which an overloaded upstream can send with a 503: + /// the gateway waits that long before sending to it again (see + /// `proxy::retry`). #[error("Provider HTTP {status}: {message}")] - ProviderHttpError { status: u16, message: String }, + ProviderHttpError { + status: u16, + message: String, + retry_after: Option, + }, /// Upstream took longer than the configured timeout. Distinct /// from a network drop because the request reached the upstream /// — only the response was missing in time. diff --git a/crates/gateway/src/lifecycle/mod.rs b/crates/gateway/src/lifecycle/mod.rs index 98e73f59..40773b2e 100644 --- a/crates/gateway/src/lifecycle/mod.rs +++ b/crates/gateway/src/lifecycle/mod.rs @@ -34,9 +34,10 @@ use tw_dialect::ir::Dialect; use crate::guards::Guards; use crate::proxy::generate::{Wire, priced, tokens}; +use crate::proxy::retry::{KEEP_ALIVE, KEEP_ALIVE_FRAME}; use crate::proxy::shaper::{StreamShaper, rewrite_model}; use crate::proxy::{ - GatewayRequestIdentity, GatewayState, SelectionRecord, emit_gateway_log_with_extra, + GatewayRequestIdentity, GatewayState, Resend, SelectionRecord, emit_gateway_log_with_extra, finalize_health, post_flight_account, prepare_body_capture, }; @@ -158,11 +159,34 @@ pub(crate) struct ChatPostInvokeDeps { pub cache_enabled: bool, } -/// The upstream call a stream makes, not yet started. -pub(crate) type OpenUpstream = Pin< - Box> + Send>, +/// The upstream call a stream makes, not yet started: each call starts +/// one attempt. +pub(crate) type OpenUpstream = Box< + dyn FnMut() -> Pin< + Box< + dyn std::future::Future> + + Send, + >, + > + Send, >; +/// What a stream's tail resolves to once the stream ends: what the hooks +/// read, and how the stream reached its upstream. +pub(crate) type StreamTail = + Pin, Sent)> + Send>>; + +/// How a stream reached its upstream, for the tail: how many times it went +/// out again after a transient failure, and when the attempt that ended it +/// went out. +#[derive(Debug, Clone, Copy)] +pub(crate) struct Sent { + pub retries: u32, + pub last_sent: std::time::Instant, + /// Waiting to go out again: the attempt that failed was counted + /// against the route already, and nothing is under way. + pub waiting: bool, +} + /// Build the streaming pump: forward the upstream's bytes to the caller — /// converted if the route speaks another format, shaped either way — and /// return a tail future that resolves once the stream ends. @@ -176,8 +200,19 @@ pub(crate) type OpenUpstream = Pin< /// rejected dialect is still retried inside `open`, before any byte /// reaches the caller. /// -/// Nothing is buffered. Usage is sniffed and the whole answer assembled -/// alongside the bytes, not by holding them back. +/// Nothing of the answer is buffered. Usage is sniffed and the whole +/// answer assembled alongside the bytes, not by holding them back. +/// +/// **A transient failure before any of the answer went out goes out +/// again** to the same route (see `proxy::retry`; `resend` decides): the +/// upstream could not be reached, answered 5xx, broke off, or reported an +/// overload or a server error before any content. While a failure could +/// still be sent again, what the upstream sends ahead of its content is +/// held back (see [`Opening`]), so that a retry's opening is not the +/// client's second one. While it is held, and while the gateway waits to +/// send again, the client gets an SSE keep-alive comment after every +/// [`KEEP_ALIVE`] of silence. The status is 200 from the start, as it +/// always was: a failure that is the answer is said in the stream. /// /// **A dropped stream is a cancelled request.** When the client goes, /// hyper drops the body, and with it the sender the tail is waiting on; @@ -205,6 +240,7 @@ pub(crate) type OpenUpstream = Pin< #[allow(clippy::too_many_arguments)] pub(crate) fn build_chat_pump( open: OpenUpstream, + resend: Resend, mut shaper: StreamShaper, client: Dialect, client_sse: bool, @@ -212,10 +248,7 @@ pub(crate) fn build_chat_pump( guards: &Guards, request: &ChatRequestSnapshot, provider: &str, -) -> ( - axum::response::Response, - Pin> + Send>>, -) { +) -> (axum::response::Response, StreamTail) { let readers = Arc::new(Mutex::new(Readers::default())); let readers_for_tail = Arc::clone(&readers); @@ -247,149 +280,301 @@ pub(crate) fn build_chat_pump( let mut last_frame = LastFrame::new(client, client_sse); let provider_for_tail = provider.to_string(); let provider = provider.to_string(); + let sent = Arc::new(Mutex::new(Sent { + retries: 0, + last_sent: std::time::Instant::now(), + waiting: false, + })); + // Gemini clients get no keep-alive comments: the google-genai SDK reads + // every SSE line as JSON. + let keep_alive = keeps_alive(client); + let sent_for_tail = Arc::clone(&sent); let body = async_stream::stream! { let mut done_tx = Some(done_tx); - - let (upstream, wire) = match open.await { - Ok(opened) => opened, - Err(e) => { - // Headers already went out as 200, so the refusal is said - // in the stream — and logged with the upstream's own status, - // so a throttled upstream stays 429 on the audit row. - let mut out = shaper.process(&error_frame(client, e.status_code(), &e.to_string())); - out.extend(shaper.finish()); - tell(&mut done_tx, StreamOutcome::UpstreamError { - error_type: e.error_tag().to_string(), - message: e.to_string(), - status_code: e.status_code(), - }); - yield Ok::(Bytes::from(out)); - return; + let mut open = open; + let mut resend = resend; + // When the client last received something: a keep-alive is due + // `KEEP_ALIVE` after it while the gateway holds the stream back. + let mut quiet_since = tokio::time::Instant::now(); + // The wait before the next attempt, after a transient failure. + let mut backoff: Option = None; + + 'attempt: loop { + if let Some(wait) = backoff.take() { + let until = tokio::time::Instant::now() + wait; + if keep_alive { + let mut ticks = std::pin::pin!(keep_alives(quiet_since, until)); + while ticks.next().await.is_some() { + quiet_since = tokio::time::Instant::now(); + yield Ok::(Bytes::from_static(KEEP_ALIVE_FRAME)); + } + } else { + tokio::time::sleep_until(until).await; + } } - }; - if let Ok(mut r) = readers.lock() { - r.sniffer = Some(tw_dialect::usage::Sniffer::new()); - r.collector = Some(wire.collect.collector()); - r.errors = Some(ErrorWatch::new(wire.dialect)); - r.ledger = Some(wire.ledger.clone()); - } - // The hop that answered can have numbered a value of its own. - shaper.restore_with(&wire.ledger); - let mut convert = wire.convert.as_ref().map(|s| s.stream()); - // Bedrock streams AWS eventstream frames, not SSE. Unframe them at - // the door, so the sniffer, the collector and the converter all - // read the same SSE they read from every other upstream. - let mut unframe = (wire.dialect == Dialect::Bedrock) - .then(tw_bedrock::eventstream::Transcoder::new); - let mut source = upstream.bytes_stream(); - while let Some(item) = source.next().await { - let item = match item { - Ok(raw) => match unframe.as_mut() { - None => Ok(raw), - Some(t) => t - .feed(&raw) - .map(Bytes::from) - .map_err(|e| bedrock_ended(&provider, e)), - }, - Err(e) => Err(broke_off(format!("The upstream stream broke off: {e}"))), + let sent_at = std::time::Instant::now(); + if let Ok(mut s) = sent.lock() { + *s = Sent { retries: resend.budget.retries(), last_sent: sent_at, waiting: false }; + } + // What an attempt that failed had read is not this one's: its + // usage is not billed, its error not the request's. + if let Ok(mut r) = readers.lock() { + *r = Readers::default(); + } + // A retry's upstream says nothing until it answers: the client + // hears keep-alives meanwhile. + let resent = resend.budget.retries() > 0; + let mut opening = open(); + let opened = loop { + let tick = tokio::time::sleep_until(quiet_since + KEEP_ALIVE); + let next = tokio::select! { + opened = &mut opening => Some(opened), + _ = tick, if resent && keep_alive => None, + }; + match next { + Some(opened) => break opened, + None => { + quiet_since = tokio::time::Instant::now(); + yield Ok(Bytes::from_static(KEEP_ALIVE_FRAME)); + } + } + }; + let (upstream, wire) = match opened { + Ok(opened) => opened, + Err(e) => { + // Nothing of the answer went out: one that failed for + // the moment goes out again after a wait. + if let Some(wait) = resend.budget.wait_after(&e) + && resend.admit(elapsed_ms(sent_at)).await + { + resending(&provider, resend.budget.retries(), wait, e.error_tag(), &readers, &sent); + backoff = Some(wait); + continue 'attempt; + } + // Headers already went out as 200, so the refusal is said + // in the stream — and logged with the upstream's own status, + // so a throttled upstream stays 429 on the audit row. + let mut out = shaper.process(&error_frame(client, e.status_code(), &e.to_string())); + out.extend(shaper.finish()); + tell(&mut done_tx, StreamOutcome::UpstreamError { + error_type: e.error_tag().to_string(), + message: e.to_string(), + status_code: e.status_code(), + }); + yield Ok(Bytes::from(out)); + return; + } }; - match item { - Ok(chunk) => { - let sealed = match readers.lock() { - Ok(mut r) => { - r.feed(&chunk); - r.sealed_refusal() + if let Ok(mut r) = readers.lock() { + *r = Readers { + sniffer: Some(tw_dialect::usage::Sniffer::new()), + collector: Some(wire.collect.collector()), + errors: Some(ErrorWatch::new(wire.dialect)), + ledger: Some(wire.ledger.clone()), + said: None, + }; + } + // The hop that answered can have numbered a value of its own. + shaper.restore_with(&wire.ledger); + let mut convert = wire.convert.as_ref().map(|s| s.stream()); + // Bedrock streams AWS eventstream frames, not SSE. Unframe them at + // the door, so the sniffer, the collector and the converter all + // read the same SSE they read from every other upstream. + let mut unframe = (wire.dialect == Dialect::Bedrock) + .then(tw_bedrock::eventstream::Transcoder::new); + // While a failure could still be sent again, the upstream's + // opening is held back until its first content (see `Opening`). + let mut hold = resend.budget.open().then(|| Opening::new(wire.dialect)); + let mut source = upstream.bytes_stream(); + loop { + let next = tokio::select! { + item = source.next() => Some(item), + _ = tokio::time::sleep_until(quiet_since + KEEP_ALIVE), if hold.is_some() && keep_alive => None, + }; + let Some(item) = next else { + quiet_since = tokio::time::Instant::now(); + yield Ok(Bytes::from_static(KEEP_ALIVE_FRAME)); + continue; + }; + let Some(item) = item else { + // The upstream's stream ended. One whose opening is still + // held ended without content: an error it reported that + // is transient goes out again; anything else goes to the + // client as it would have. + if let Some(h) = hold.take() { + let said = readers.lock().ok().and_then(|mut r| r.reported(true)); + if let Some(said) = said + && let Some(wait) = retry_after_said(&resend, &provider, &said) + && resend.admit(elapsed_ms(sent_at)).await + { + resending(&provider, resend.budget.retries(), wait, "in-stream error", &readers, &sent); + backoff = Some(wait); + continue 'attempt; + } + match deliver(&h.release(), &mut convert, &mut shaper, inspector.as_mut(), client) { + Ok(out) => { + if !out.is_empty() { + if last_frame.is_in(&out) { + delivered.store(true, Ordering::Release); + } + yield Ok(Bytes::from(out)); + } + } + Err((out, outcome)) => { + tell(&mut done_tx, outcome); + yield Ok(Bytes::from(out)); + return; + } } - Err(_) => None, - }; - // The upstream cannot read the conversation's encrypted - // history: the client is told why in plain words, in - // place of the upstream's own error. - if let Some(said) = sealed { - let err = sealed_refused(&provider, &said); - tracing::warn!( - provider = %provider, - message = %said.message, - "upstream refused the conversation's encrypted history" - ); - let out = cut(&shaper, convert.as_mut(), client, &[], &err); - tell(&mut done_tx, StreamOutcome::UpstreamError { - error_type: err.error_tag().to_string(), - message: err.to_string(), - status_code: err.status_code(), - }); - yield Ok(Bytes::from(out)); - return; } - let client_bytes = match convert.as_mut() { - Some(c) => c.process(&chunk), - None => chunk.to_vec(), - }; - let out = shaper.process(&client_bytes); - // A tool call the inspection stops: what came before - // still goes out, then the refusal. - let stop = inspector.as_mut().and_then(|i| i.check(&out)); - if let Some((err, safe)) = stop { - let out = cut(&shaper, convert.as_mut(), client, &out[..safe], &err); - tell(&mut done_tx, StreamOutcome::UpstreamError { - error_type: err.error_tag().to_string(), - message: err.to_string(), - status_code: err.status_code(), - }); - yield Ok(Bytes::from(out)); - return; + break; + }; + let item = match item { + Ok(raw) => match unframe.as_mut() { + None => Ok(raw), + Some(t) => t + .feed(&raw) + .map(Bytes::from) + .map_err(|e| bedrock_ended(&provider, e)), + }, + Err(e) => Err(broke_off(format!("The upstream stream broke off: {e}"))), + }; + match item { + Ok(chunk) => { + let (sealed, said) = match readers.lock() { + Ok(mut r) => { + r.feed(&chunk); + (r.sealed_refusal(), r.reported(false)) + } + Err(_) => (None, None), + }; + // The upstream cannot read the conversation's encrypted + // history: the client is told why in plain words, in + // place of the upstream's own error. + if let Some(said) = sealed { + let err = sealed_refused(&provider, &said); + tracing::warn!( + provider = %provider, + message = %said.message, + "upstream refused the conversation's encrypted history" + ); + let out = cut(&shaper, convert.as_mut(), client, &[], &err); + tell(&mut done_tx, StreamOutcome::UpstreamError { + error_type: err.error_tag().to_string(), + message: err.to_string(), + status_code: err.status_code(), + }); + yield Ok(Bytes::from(out)); + return; + } + let chunk = match hold.as_mut() { + None => chunk.to_vec(), + Some(h) => { + // An error before any content: one that is + // transient goes out again. Anything else is + // the answer, and goes out with the opening. + if !h.feed(&chunk) { + let Some(said) = &said else { continue }; + if let Some(wait) = retry_after_said(&resend, &provider, said) + && resend.admit(elapsed_ms(sent_at)).await + { + resending(&provider, resend.budget.retries(), wait, "in-stream error", &readers, &sent); + backoff = Some(wait); + continue 'attempt; + } + } + hold.take().map(Opening::release).unwrap_or_default() + } + }; + match deliver(&chunk, &mut convert, &mut shaper, inspector.as_mut(), client) { + Ok(out) => { + if !out.is_empty() { + if last_frame.is_in(&out) { + delivered.store(true, Ordering::Release); + } + yield Ok(Bytes::from(out)); + } + } + // A tool call the inspection stops: what came + // before still goes out, then the refusal. + Err((out, outcome)) => { + tell(&mut done_tx, outcome); + yield Ok(Bytes::from(out)); + return; + } + } } - if !out.is_empty() { - if last_frame.is_in(&out) { - delivered.store(true, Ordering::Release); + Err((message, outcome)) => { + // Broken off before any content: the upstream's + // failure, which goes out again. + if let Some(h) = hold.take() { + if let StreamOutcome::UpstreamError { error_type, status_code, .. } = &outcome + && let Some(wait) = resend.budget.wait_after_outcome(error_type, *status_code) + && resend.admit(elapsed_ms(sent_at)).await + { + resending(&provider, resend.budget.retries(), wait, error_type, &readers, &sent); + backoff = Some(wait); + continue 'attempt; + } + // The opening it sent goes out before the error. + match deliver(&h.release(), &mut convert, &mut shaper, inspector.as_mut(), client) { + Ok(out) => { + if !out.is_empty() { + yield Ok(Bytes::from(out)); + } + } + Err((out, outcome)) => { + tell(&mut done_tx, outcome); + yield Ok(Bytes::from(out)); + return; + } + } } + // Headers are gone; the only way left to say it is in + // the stream, in the caller's own format. + let tail = match convert.as_mut() { + Some(c) => c.fail(&message), + None => error_frame(client, 502, &message), + }; + let mut out = shaper.process(&tail); + out.extend(shaper.finish()); + tell(&mut done_tx, outcome); yield Ok(Bytes::from(out)); + return; } } - Err((message, outcome)) => { - // Headers are gone; the only way left to say it is in - // the stream, in the caller's own format. - let tail = match convert.as_mut() { - Some(c) => c.fail(&message), - None => error_frame(client, 502, &message), - }; - let mut out = shaper.process(&tail); - out.extend(shaper.finish()); - tell(&mut done_tx, outcome); - yield Ok(Bytes::from(out)); - return; - } } - } - let tail = convert.as_mut().map(|c| c.finish()).unwrap_or_default(); - let mut out = shaper.process(&tail); - out.extend(shaper.finish()); - // The converter's last bytes can complete a tool call (the block's - // stop), so they are inspected too. - // A Chat call with no `finish_reason` or `[DONE]` completes only - // as the stream closes; one that stops it keeps the tail back. - let stop = inspector.as_mut().and_then(|i| { - i.check(&out) - .or_else(|| i.finish().map(|err| (err, 0))) - }); - if let Some((err, safe)) = stop { - let out = cut(&shaper, None, client, &out[..safe], &err); - tell(&mut done_tx, StreamOutcome::UpstreamError { - error_type: err.error_tag().to_string(), - message: err.to_string(), - status_code: err.status_code(), + let tail = convert.as_mut().map(|c| c.finish()).unwrap_or_default(); + let mut out = shaper.process(&tail); + out.extend(shaper.finish()); + // The converter's last bytes can complete a tool call (the block's + // stop), so they are inspected too. + // A Chat call with no `finish_reason` or `[DONE]` completes only + // as the stream closes; one that stops it keeps the tail back. + let stop = inspector.as_mut().and_then(|i| { + i.check(&out) + .or_else(|| i.finish().map(|err| (err, 0))) }); - yield Ok(Bytes::from(out)); + if let Some((err, safe)) = stop { + let out = cut(&shaper, None, client, &out[..safe], &err); + tell(&mut done_tx, StreamOutcome::UpstreamError { + error_type: err.error_tag().to_string(), + message: err.to_string(), + status_code: err.status_code(), + }); + yield Ok(Bytes::from(out)); + return; + } + let outcome = match readers.lock().ok().and_then(|mut r| r.upstream_error()) { + Some(said) => failed_partway(&provider, &said), + None => StreamOutcome::Natural, + }; + tell(&mut done_tx, outcome); + if !out.is_empty() { + yield Ok(Bytes::from(out)); + } return; } - let outcome = match readers.lock().ok().and_then(|mut r| r.upstream_error()) { - Some(said) => failed_partway(&provider, &said), - None => StreamOutcome::Natural, - }; - tell(&mut done_tx, outcome); - if !out.is_empty() { - yield Ok(Bytes::from(out)); - } }; // A Gemini caller that did not ask for SSE reads one JSON array. @@ -494,7 +679,12 @@ pub(crate) fn build_chat_pump( assembled, logged, }; - Invoked { + let sent = sent_for_tail.lock().map(|s| *s).unwrap_or(Sent { + retries: 0, + last_sent: started_at, + waiting: false, + }); + let invoked = Invoked { client_ip: identity.ip_address.clone(), identity, trace_id, @@ -504,7 +694,8 @@ pub(crate) fn build_chat_pump( }, access_candidate: mapped_model, view: CapturedView::Streaming { outcome, captured }, - } + }; + (invoked, sent) }); (response, tail) } @@ -566,6 +757,20 @@ impl Readers { self.said.as_ref().filter(|s| s.sealed).cloned() } + /// The error the upstream reported in its stream so far, left in place + /// for the tail. `ended`: the stream ended, so a last frame without a + /// blank line after it is read too. + fn reported(&mut self, ended: bool) -> Option { + if ended + && self.said.is_none() + && let Some(said) = self.errors.as_mut().and_then(ErrorWatch::flush) + { + self.said = Some(said); + self.errors = None; + } + self.said.clone() + } + /// The error the upstream reported in its stream, if it did — read to /// the end, for an upstream whose last frame has no blank line after /// it. @@ -723,7 +928,7 @@ fn status_named(name: &str) -> Option { | "insufficient_quota" | "usage_limit_reached" | "RESOURCE_EXHAUSTED" => 429, - "api_error" => 500, + "api_error" | "server_error" => 500, "server_is_overloaded" | "slow_down" | "UNAVAILABLE" => 503, "timeout_error" | "DEADLINE_EXCEEDED" => 504, "overloaded_error" => 529, @@ -749,6 +954,11 @@ fn failed_partway(provider: &str, said: &Said) -> StreamOutcome { message = %said.message, "upstream reported an error partway through a stream" ); + outcome_of(provider, said) +} + +/// [`failed_partway`], without the log line. +fn outcome_of(provider: &str, said: &Said) -> StreamOutcome { let words: String = said.message.chars().take(500).collect(); if said.sealed { let e = sealed_refused(provider, said); @@ -847,6 +1057,164 @@ fn bedrock_status(kind: &str) -> u16 { .map_or(500, |(_, status)| *status) } +/// The wait before sending again after an error the upstream reported +/// in its stream before any content, when it is transient (see +/// `proxy::retry`) and the wait fits. +fn retry_after_said(resend: &Resend, provider: &str, said: &Said) -> Option { + match outcome_of(provider, said) { + StreamOutcome::UpstreamError { + error_type, + status_code, + .. + } => resend.budget.wait_after_outcome(&error_type, status_code), + _ => None, + } +} + +/// Whether the client may be sent keep-alive comments while the gateway +/// holds its stream back. Not a Gemini client: the google-genai SDK reads +/// every SSE line as JSON, and a JSON-array stream has no place for them. +fn keeps_alive(client: Dialect) -> bool { + client != Dialect::Gemini +} + +/// A stream goes out again after `wait`: what the failed attempt left for +/// the tail is cleared — a client that leaves during the wait left a +/// request whose failed attempt is already counted against the route +/// (see `routing::Resend::admit`), with nothing to bill — and the log +/// line written. +fn resending( + provider: &str, + retry: u32, + wait: std::time::Duration, + error: &str, + readers: &Mutex, + sent: &Mutex, +) { + if let Ok(mut r) = readers.lock() { + *r = Readers::default(); + } + if let Ok(mut s) = sent.lock() { + *s = Sent { + retries: retry, + last_sent: std::time::Instant::now(), + waiting: true, + }; + } + tracing::info!( + provider, + retry, + wait_ms = wait.as_millis() as u64, + error, + "Upstream failed for the moment before any of the answer went out, sending again after a wait" + ); +} + +fn elapsed_ms(since: std::time::Instant) -> u32 { + since.elapsed().as_millis().min(u32::MAX as u128) as u32 +} + +/// A tick each time the client is due a keep-alive while the gateway waits +/// until `until`: [`KEEP_ALIVE`] after `quiet_since`, then every +/// [`KEEP_ALIVE`]. Ends at `until`. +fn keep_alives( + quiet_since: tokio::time::Instant, + until: tokio::time::Instant, +) -> impl futures::Stream { + async_stream::stream! { + let mut due = quiet_since + KEEP_ALIVE; + while due < until { + tokio::time::sleep_until(due).await; + yield (); + due += KEEP_ALIVE; + } + tokio::time::sleep_until(until).await; + } +} + +/// What the client gets for `chunk` of the upstream's bytes: converted if +/// the route speaks another format, shaped, and inspected for tool calls. +/// `Err`: a tool call the inspection stops — what came before it, then +/// the refusal — and the stream's outcome. +fn deliver( + chunk: &[u8], + convert: &mut Option, + shaper: &mut StreamShaper, + inspector: Option<&mut crate::tool_inspection::StreamInspector>, + client: Dialect, +) -> Result, (Vec, StreamOutcome)> { + let client_bytes = match convert.as_mut() { + Some(c) => c.process(chunk), + None => chunk.to_vec(), + }; + let out = shaper.process(&client_bytes); + match inspector.and_then(|i| i.check(&out)) { + Some((err, safe)) => { + let out = cut(shaper, convert.as_mut(), client, &out[..safe], &err); + Err(( + out, + StreamOutcome::UpstreamError { + error_type: err.error_tag().to_string(), + message: err.to_string(), + status_code: err.status_code(), + }, + )) + } + None => Ok(out), + } +} + +/// A stream's opening, held back from the client while a failure could +/// still be sent again (see `proxy::retry`): what an upstream sends as +/// soon as it takes the request — Anthropic's `message_start` and `ping`, +/// Responses' `response.created` and `response.in_progress`, a Chat chunk +/// with only the role, Bedrock's `messageStart` — is not yet the answer. +/// Sent ahead of a failure, it could not be taken back, and a retry would +/// send it twice. It goes out with the first content, or with an error +/// that is the answer. +/// +/// Read on the upstream's bytes in its own format, as +/// `tw_dialect::convert::Reader` reads them: anything past the start and +/// the usage is content. An opening that grows past [`Opening::MAX`] +/// without any is let through: a broken upstream must not make it grow +/// without end. +struct Opening { + reader: tw_dialect::convert::Reader, + held: Vec, +} + +impl Opening { + const MAX: usize = 1024 * 1024; + + fn new(upstream: Dialect) -> Self { + Self { + reader: tw_dialect::convert::Reader::new(upstream), + held: Vec::new(), + } + } + + /// Hold `chunk` back. True once it carries content ahead of any error, + /// and with it what was held can go out: an error after content is + /// one after the answer began, whatever piece of the upstream's bytes + /// it came in. + fn feed(&mut self, chunk: &[u8]) -> bool { + use tw_dialect::ir::Event; + self.held.extend_from_slice(chunk); + let content = self + .reader + .feed(chunk) + .iter() + .take_while(|e| !matches!(e, Event::Error { .. })) + .any(|e| !matches!(e, Event::Start { .. } | Event::Usage(_))); + content || self.held.len() > Self::MAX + } + + /// What was held back, to go out. + fn release(self) -> Vec { + self.held + } +} + /// Tell the tail how the stream ended. Only the first outcome counts. fn tell(done_tx: &mut Option>, outcome: StreamOutcome) { if let Some(tx) = done_tx.take() { @@ -1156,7 +1524,10 @@ impl Surface for ChatCompletionSurface { deps.request.request_started_at.elapsed().as_millis() as i64, logged_status, with_usage_detail( - pinned_detail(error_detail, deps.route.pinned), + retried_detail( + pinned_detail(error_detail, deps.route.pinned), + deps.route.sel_record.retries, + ), &usage, usage_estimated(&invoked.view), ), @@ -1209,6 +1580,20 @@ fn pinned_detail(detail: Option, pinned: bool) -> Option, retries: u32) -> Option { + if retries == 0 { + return detail; + } + let mut d = match detail { + Some(serde_json::Value::Object(d)) => d, + _ => serde_json::Map::new(), + }; + d.insert("retries".into(), retries.into()); + Some(serde_json::Value::Object(d)) +} + /// The usage a captured view bills — shared by `record_usage` and /// `emit_audit` so the budget and the audit row can never disagree. fn extract_usage(view: &CapturedView) -> tw_dialect::usage::Usage { @@ -1233,6 +1618,242 @@ fn usage_estimated(view: &CapturedView) -> bool { mod tests { use super::*; + fn frame(kind: &str, v: serde_json::Value) -> String { + format!("event: {kind}\ndata: {v}\n\n") + } + + /// What an upstream sends as soon as it takes the request is held; + /// the first content lets it all through. + #[test] + fn a_streams_opening_is_held_until_its_first_content() { + use serde_json::json; + + let start = frame( + "message_start", + json!({"type": "message_start", "message": {"id": "msg_1", "model": "m", + "usage": {"input_tokens": 3, "output_tokens": 1}}}), + ); + let ping = frame("ping", json!({"type": "ping"})); + let block = frame( + "content_block_start", + json!({"type": "content_block_start", "index": 0, + "content_block": {"type": "text", "text": ""}}), + ); + let mut o = Opening::new(Dialect::Anthropic); + assert!(!o.feed(start.as_bytes())); + assert!(!o.feed(ping.as_bytes())); + assert!(o.feed(block.as_bytes())); + assert_eq!(o.release(), [start, ping, block].concat().into_bytes()); + + let mut o = Opening::new(Dialect::Responses); + let created = json!({"response": {"id": "resp_1", "model": "m"}}); + assert!(!o.feed(frame("response.created", created.clone()).as_bytes())); + assert!(!o.feed(frame("response.in_progress", created).as_bytes())); + assert!( + o.feed( + frame( + "response.output_text.delta", + json!({"type": "response.output_text.delta", "output_index": 0, + "content_index": 0, "delta": "Hi"}), + ) + .as_bytes() + ) + ); + + let chunk = |delta: serde_json::Value| { + format!( + "data: {}\n\n", + json!({"id": "c", "model": "m", "choices": [{"index": 0, "delta": delta}]}) + ) + }; + let mut o = Opening::new(Dialect::Chat); + assert!(!o.feed(chunk(json!({"role": "assistant", "content": ""})).as_bytes())); + assert!(o.feed(chunk(json!({"content": "Hi"})).as_bytes())); + } + + /// An error is not content: whether it goes out again is decided + /// with the opening still held. Content ahead of it in the same piece + /// of bytes began the answer. + #[test] + fn an_error_in_the_opening_is_not_content() { + use serde_json::json; + + let start = frame( + "message_start", + json!({"type": "message_start", "message": {"id": "msg_1", "model": "m"}}), + ); + let overload = frame( + "error", + json!({"type": "error", + "error": {"type": "overloaded_error", "message": "Overloaded"}}), + ); + let text = frame( + "content_block_start", + json!({"type": "content_block_start", "index": 0, + "content_block": {"type": "text", "text": ""}}), + ); + let mut o = Opening::new(Dialect::Anthropic); + assert!(!o.feed([start.as_str(), overload.as_str()].concat().as_bytes())); + + let mut o = Opening::new(Dialect::Anthropic); + assert!(o.feed([start, text, overload].concat().as_bytes())); + } + + /// An opening that grows without any content is let through. + #[test] + fn an_opening_does_not_grow_without_end() { + let mut o = Opening::new(Dialect::Chat); + let comment = format!(": {}\n\n", "x".repeat(64 * 1024)); + let fed = (0..32) + .map(|_| o.feed(comment.as_bytes())) + .collect::>(); + assert!(!fed[0]); + assert!(fed.last().copied().unwrap_or(false)); + } + + /// Which errors an upstream reports in its stream are worth sending + /// again for: an overload or a server error, not a refusal. + #[test] + fn an_overload_or_a_server_error_in_a_stream_is_transient() { + let cases = [ + ( + Dialect::Anthropic, + "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}\n\n", + true, + ), + ( + Dialect::Anthropic, + "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"api_error\",\"message\":\"Internal\"}}\n\n", + true, + ), + ( + Dialect::Chat, + "data: {\"error\":{\"message\":\"boom\",\"type\":\"server_error\"}}\n\n", + true, + ), + ( + Dialect::Responses, + "event: response.failed\ndata: {\"type\":\"response.failed\",\"response\":{\"error\":{\"code\":\"server_is_overloaded\",\"message\":\"busy\"}}}\n\n", + true, + ), + ( + Dialect::Gemini, + "data: {\"error\":{\"code\":503,\"message\":\"busy\",\"status\":\"UNAVAILABLE\"}}\n\n", + true, + ), + ( + Dialect::Anthropic, + "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"too long\"}}\n\n", + false, + ), + ( + Dialect::Anthropic, + "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"rate_limit_error\",\"message\":\"slow down\"}}\n\n", + false, + ), + // Nothing recognisable: not sent again on a guess. + ( + Dialect::Chat, + "data: {\"error\":{\"message\":\"odd\",\"code\":\"upstream_wobble\"}}\n\n", + false, + ), + ]; + for (dialect, frames, transient) in cases { + let mut r = Readers { + errors: Some(ErrorWatch::new(dialect)), + ..Default::default() + }; + r.feed(frames.as_bytes()); + let said = r.reported(false).unwrap_or_else(|| panic!("{frames}")); + let StreamOutcome::UpstreamError { + error_type, + status_code, + .. + } = outcome_of("up", &said) + else { + panic!("not a failure"); + }; + assert_eq!( + crate::proxy::retry::transient(&error_type, status_code), + transient, + "{frames}" + ); + // Left in place for the tail. + assert_eq!(r.upstream_error(), Some(said)); + } + } + + /// A last error frame without a blank line after it is read once the + /// stream ends. + #[test] + fn an_error_in_the_last_frame_is_reported_at_the_end() { + let mut r = Readers { + errors: Some(ErrorWatch::new(Dialect::Chat)), + ..Default::default() + }; + r.feed(b"data: {\"error\":{\"message\":\"boom\",\"type\":\"server_error\"}}"); + assert_eq!(r.reported(false), None); + let said = r.reported(true).unwrap(); + assert_eq!(said.status, Some(500)); + } + + /// A Gemini client, SSE or JSON array, never gets a keep-alive comment. + #[test] + fn only_non_gemini_clients_get_keep_alives() { + assert!(!keeps_alive(Dialect::Gemini)); + for client in [Dialect::Chat, Dialect::Anthropic, Dialect::Responses] { + assert!(keeps_alive(client)); + } + } + + /// A retry clears what the failed attempt left for the tail, and says + /// the stream is waiting. + #[test] + fn deciding_a_retry_clears_the_failed_attempt() { + let readers = Mutex::new(Readers { + errors: Some(ErrorWatch::new(Dialect::Anthropic)), + sniffer: Some(tw_dialect::usage::Sniffer::new()), + ..Default::default() + }); + readers.lock().unwrap().feed( + b"event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"x\"}}\n\n", + ); + let sent = Mutex::new(Sent { + retries: 0, + last_sent: std::time::Instant::now(), + waiting: false, + }); + resending( + "up", + 1, + std::time::Duration::from_secs(2), + "x", + &readers, + &sent, + ); + let mut r = readers.lock().unwrap(); + assert!(r.sniffer.is_none() && r.upstream_error().is_none()); + let s = *sent.lock().unwrap(); + assert_eq!((s.retries, s.waiting), (1, true)); + } + + /// While the gateway waits, the client hears a keep-alive every 15 + /// seconds of silence, and the wait ends on time. + #[tokio::test(start_paused = true)] + async fn a_wait_sends_a_keep_alive_every_fifteen_seconds() { + let start = tokio::time::Instant::now(); + let quiet_since = start - std::time::Duration::from_secs(5); + let until = start + std::time::Duration::from_secs(40); + let ticks = keep_alives(quiet_since, until); + let mut ticks = std::pin::pin!(ticks); + let mut at = Vec::new(); + while ticks.next().await.is_some() { + at.push((tokio::time::Instant::now() - start).as_secs()); + } + assert_eq!(at, [10, 25]); + assert_eq!((tokio::time::Instant::now() - start).as_secs(), 40); + } + #[test] fn the_last_frame_is_seen_in_the_chunk_that_completes_it() { let mut w = LastFrame::new(Dialect::Responses, true); @@ -1451,7 +2072,7 @@ mod tests { /// the chunk's text, which can be the model's answer. #[test] fn an_error_that_names_no_status_is_the_upstreams_failure() { - let chat = "data: {\"choices\":[{\"delta\":{\"content\":\"the secret plan\"}}],\"error\":{\"code\":\"server_error\"}}\n\n"; + let chat = "data: {\"choices\":[{\"delta\":{\"content\":\"the secret plan\"}}],\"error\":{\"code\":\"upstream_wobble\"}}\n\n"; let (error_type, status, message, counted) = recorded(Dialect::Chat, chat).unwrap(); assert_eq!((error_type.as_str(), status), ("ProviderError", 502)); assert!(counted); @@ -1483,11 +2104,12 @@ mod tests { let gemini = "data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"hi\"}]}}],\"error\":null}\n\n"; assert_eq!(recorded(Dialect::Gemini, gemini), None); - // And an error after such chunks is still seen. + // And an error after such chunks is still seen: OpenAI's + // `server_error` is its 500. let then = format!( "{chat}data: {{\"error\":{{\"message\":\"boom\",\"type\":\"server_error\"}}}}\n\n" ); - assert_eq!(recorded(Dialect::Chat, &then).unwrap().1, 502); + assert_eq!(recorded(Dialect::Chat, &then).unwrap().1, 500); } /// Cut to 500 characters, as a refusal's body is. diff --git a/crates/gateway/src/proxy/cache_marks.rs b/crates/gateway/src/proxy/cache_marks.rs index e07e7b50..3ac5c265 100644 --- a/crates/gateway/src/proxy/cache_marks.rs +++ b/crates/gateway/src/proxy/cache_marks.rs @@ -73,6 +73,7 @@ pub fn is_refusal(err: &GatewayError, label: &str) -> bool { let GatewayError::ProviderHttpError { status: 400, message, + .. } = err else { return false; @@ -102,6 +103,7 @@ mod tests { GatewayError::ProviderHttpError { status, message: format!("relay: {said}"), + retry_after: None, } } @@ -134,6 +136,7 @@ mod tests { &GatewayError::ProviderHttpError { status: 400, message: "prompt caching proxy: max_tokens: Field required".into(), + retry_after: None, }, "prompt caching proxy" )); diff --git a/crates/gateway/src/proxy/generate.rs b/crates/gateway/src/proxy/generate.rs index b679b3ad..6f356e80 100644 --- a/crates/gateway/src/proxy/generate.rs +++ b/crates/gateway/src/proxy/generate.rs @@ -51,9 +51,10 @@ use super::headers::{ }; use super::log_ctx::{LogCtx, emit_gateway_error_log, emit_gateway_log}; use super::pipeline::{launch_stream_pump, run_buffered_post_invoke, run_preflight_stages}; +use super::retry::Budget; use super::routing::{ - build_selection_ctx, finalize_health, select_route_for_stream, select_route_with_failover, - set_affinity, + Resend, build_selection_ctx, finalize_health, select_route_for_stream, + select_route_with_failover, set_affinity, }; use super::sealed::Conversation; use super::shaper::{StreamShaper, rewrite_model}; @@ -113,6 +114,7 @@ pub async fn proxy_chat_completion( headers: HeaderMap, axum::Extension(identity): axum::Extension, cancel: Option>, + arrived: Option>, body: Bytes, ) -> Result { generate( @@ -120,6 +122,7 @@ pub async fn proxy_chat_completion( headers, identity, cancel.map(|c| c.0), + arrived.map(|a| a.0.0), body, CHAT, "/v1/chat/completions", @@ -134,6 +137,7 @@ pub async fn proxy_anthropic_messages( headers: HeaderMap, axum::Extension(identity): axum::Extension, cancel: Option>, + arrived: Option>, body: Bytes, ) -> Result { generate( @@ -141,6 +145,7 @@ pub async fn proxy_anthropic_messages( headers, identity, cancel.map(|c| c.0), + arrived.map(|a| a.0.0), body, MESSAGES, "/v1/messages", @@ -155,6 +160,7 @@ pub async fn proxy_responses( headers: HeaderMap, axum::Extension(identity): axum::Extension, cancel: Option>, + arrived: Option>, body: Bytes, ) -> Result { generate( @@ -162,6 +168,7 @@ pub async fn proxy_responses( headers, identity, cancel.map(|c| c.0), + arrived.map(|a| a.0.0), body, RESPONSES, "/v1/responses", @@ -180,6 +187,7 @@ pub async fn proxy_gemini( headers: HeaderMap, axum::Extension(identity): axum::Extension, cancel: Option>, + arrived: Option>, body: Bytes, ) -> Result { generate( @@ -187,6 +195,7 @@ pub async fn proxy_gemini( headers, identity, cancel.map(|c| c.0), + arrived.map(|a| a.0.0), body, GEMINI, uri.path(), @@ -640,21 +649,27 @@ pub(crate) async fn read_whole( /// /// `cancel` is the middleware's record of a request whose client leaves /// before the response exists (see `early_cancel`); `None` on a -/// WebSocket turn, whose connection records its own end. +/// WebSocket turn, whose connection records its own end. `arrived` is +/// when the gateway port's request timeout started counting (see +/// [`super::Arrived`]); `None` on a WebSocket turn, which it does not +/// bound. #[allow(clippy::too_many_arguments)] pub(crate) async fn generate( state: GatewayState, headers: HeaderMap, identity: GatewayRequestIdentity, cancel: Option, + arrived: Option, body: Bytes, surface: ClientSurface, path: &str, query: Option<&str>, ) -> Result { - run(state, headers, identity, cancel, body, surface, path, query) - .await - .map_err(|e| e.in_dialect(surface.dialect)) + run( + state, headers, identity, cancel, arrived, body, surface, path, query, + ) + .await + .map_err(|e| e.in_dialect(surface.dialect)) } #[allow(clippy::too_many_arguments)] @@ -663,6 +678,7 @@ async fn run( headers: HeaderMap, identity: GatewayRequestIdentity, cancel: Option, + arrived: Option, body: Bytes, surface: ClientSurface, path: &str, @@ -1054,14 +1070,21 @@ async fn run( cache_enabled: surface.caches, }; + // How long after it arrived the request may still go out again to + // upstreams that failed for the moment (see `super::retry`). + let retry_wait = std::time::Duration::from_secs(state.dynamic_config.retry_wait_secs().await); + if outbound.stream { - // One pick, no retry once bytes have gone to the client. A dialect - // rejection still gets its retry: it arrives before any body. + // One pick, no failover once bytes have gone to the client. A + // dialect rejection still gets its retry: it arrives before any + // body. A transient failure before any of the answer went out + // sends to the same route again (see `build_chat_pump`). let sel_ctx = build_selection_ctx(&state, &mapped_model, identity.user_id.as_deref(), pin).await; let (entry, sel_record) = select_route_for_stream(routes, &sel_ctx) .await .map_err(|e| GatewayErrorResponse::from(ctx.emit(e)))?; + let resend = Resend::new(entry, &sel_ctx, Budget::new(request_started_at, retry_wait)); set_affinity( &state.redis, @@ -1075,16 +1098,26 @@ async fn run( let hide_usage = outbound.hides_usage(); // Started on the stream's first poll — see `build_chat_pump` for - // why it must not be awaited here. + // why it must not be awaited here — and again for each retry. let open: crate::lifecycle::OpenUpstream = { let entry = entry.clone(); + let outbound = std::sync::Arc::new(outbound); let call_ctx = call_ctx.clone(); let db = state.db.clone(); let model = entry .upstream_model .clone() .unwrap_or_else(|| mapped_model.clone()); - Box::pin(async move { send(&entry, &outbound, &call_ctx, &db, &model).await }) + Box::new(move || { + let (entry, outbound, call_ctx, db, model) = ( + entry.clone(), + outbound.clone(), + call_ctx.clone(), + db.clone(), + model.clone(), + ); + Box::pin(async move { send(&entry, &outbound, &call_ctx, &db, &model).await }) + }) }; let deps = snapshot(entry, sel_record); @@ -1094,6 +1127,7 @@ async fn run( return Ok(launch_stream_pump( deps, open, + resend, shaper, surface.dialect, client_sse, @@ -1114,28 +1148,44 @@ async fn run( None, ) .await; - let (entry, completed, answer_ledger, sel_record) = - select_route_with_failover(routes, &outbound, &call_ctx, &sel_ctx, &mapped_model) - .await - .map_err(|e| { - // Every candidate failed — no winning provider to name. - emit_gateway_error_log( - &state.audit, - &metadata.request_id, - session_id.as_deref(), - identity.user_id.as_deref(), - identity.user_email.as_deref(), - identity.api_key_id.as_deref(), - identity.api_key_lineage_id.as_deref(), - identity.ip_address.as_deref(), - &mapped_model, - None, - request_started_at.elapsed().as_millis() as i64, - &e, - error_capture, - ); - GatewayErrorResponse::from(e) - })?; + // A whole answer must be back within the gateway port's request + // timeout, which started counting before this handler did (auth, the + // body): past it the client gets an empty 408 and the request no log + // row. Each attempt is cut a little before it, and a retry starts only + // when it fits (see `super::retry`). + let mut budget = Budget::new(request_started_at, retry_wait); + if let Some(arrived) = arrived { + budget = budget.within(super::retry::hard_deadline(arrived)); + } + let (entry, completed, answer_ledger, sel_record) = select_route_with_failover( + routes, + &outbound, + &call_ctx, + &sel_ctx, + &mapped_model, + &mut budget, + ) + .await + .map_err(|e| { + // Every candidate failed — no winning provider to name. + emit_gateway_error_log( + &state.audit, + &metadata.request_id, + session_id.as_deref(), + identity.user_id.as_deref(), + identity.user_email.as_deref(), + identity.api_key_id.as_deref(), + identity.api_key_lineage_id.as_deref(), + identity.ip_address.as_deref(), + &mapped_model, + None, + request_started_at.elapsed().as_millis() as i64, + &e, + budget.retries(), + error_capture, + ); + GatewayErrorResponse::from(e) + })?; // The answer as the caller will receive it: their values restored, // then its text through the content rules that check answers. diff --git a/crates/gateway/src/proxy/log_ctx.rs b/crates/gateway/src/proxy/log_ctx.rs index 6efb362c..8e3d4890 100644 --- a/crates/gateway/src/proxy/log_ctx.rs +++ b/crates/gateway/src/proxy/log_ctx.rs @@ -82,6 +82,7 @@ impl<'a> LogCtx<'a> { None, self.started.elapsed().as_millis() as i64, &err, + 0, // Pre-route-selection failures don't have a parsed request // we can safely serialize (transform errors, malformed JSON, // etc. land here). Body capture for these paths is a follow- @@ -96,6 +97,10 @@ impl<'a> LogCtx<'a> { /// blob mirrors the success-path shape but also carries `error_type` /// and `error_message` so operators can drill down without joining /// against a separate error-log stream. +/// +/// `retries`: how many times the request went out again after a +/// transient failure (see `super::retry`), said as `"retries"` when any +/// did. #[allow(clippy::too_many_arguments)] pub(super) fn emit_gateway_error_log( audit: &think_watch_common::audit::AuditLogger, @@ -110,9 +115,13 @@ pub(super) fn emit_gateway_error_log( provider: Option<&str>, latency_ms: i64, err: &GatewayError, + retries: u32, bodies: BodyCapture, ) { - let detail = error_detail(model_id, provider, latency_ms, err); + let mut detail = error_detail(model_id, provider, latency_ms, err); + if retries > 0 { + detail["retries"] = retries.into(); + } // Same `chat.completion` action as the success path — flush_gateway // drops the action when it writes ChGatewayRow, so the trace // endpoint distinguishes errors via `status_code` (>= 400) instead. @@ -272,6 +281,7 @@ mod tests { GatewayError::ProviderHttpError { status: 500, message: "bedrock: arn:aws:iam::123456789012:user/gateway".into(), + retry_after: None, }, GatewayError::UpstreamRateLimited { retry_after_secs: Some(12), diff --git a/crates/gateway/src/proxy/mod.rs b/crates/gateway/src/proxy/mod.rs index de83c925..d0fa85a9 100644 --- a/crates/gateway/src/proxy/mod.rs +++ b/crates/gateway/src/proxy/mod.rs @@ -31,6 +31,7 @@ mod models; mod pipeline; mod protocol_relearn; mod responses_ws; +pub(crate) mod retry; mod routing; pub(crate) mod sealed; pub mod shaper; @@ -40,7 +41,7 @@ pub mod transport; pub(crate) use accounting::post_flight_account; pub(crate) use body_capture::prepare_body_capture; pub(crate) use log_ctx::emit_gateway_log_with_extra; -pub(crate) use routing::{SelectionRecord, fails as upstream_failed, finalize_health}; +pub(crate) use routing::{Resend, SelectionRecord, fails as upstream_failed, finalize_health}; // pub re-exports — `server::app` mounts these as route handlers. pub use early_cancel::{EarlyCancel, EarlyCancelSlot}; @@ -51,6 +52,30 @@ pub use identity::limits_for_ai_gateway; pub use models::{list_gemini_models_handler, list_models_handler}; pub use responses_ws::proxy_responses_ws; +/// How long the gateway port gives a handler to produce its response +/// before answering 408 (the server's `TimeoutLayer`). A stream's body is +/// not bound by it — only the time to its head — but a whole answer is, +/// failover and retries included, so its retries stop by then too. +pub const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120); + +/// When a request reached the gateway port, stamped just inside the +/// server's `TimeoutLayer` (see [`stamp_arrival`]): the moment its +/// [`REQUEST_TIMEOUT`] started counting, before authentication and the +/// body were read. +#[derive(Debug, Clone, Copy)] +pub struct Arrived(pub std::time::Instant); + +/// Stamp [`Arrived`] on the request. Layered directly inside the gateway +/// port's `TimeoutLayer`. +pub async fn stamp_arrival( + mut req: axum::extract::Request, + next: axum::middleware::Next, +) -> axum::response::Response { + req.extensions_mut() + .insert(Arrived(std::time::Instant::now())); + next.run(req).await +} + /// Shared application state for the gateway proxy handlers. #[derive(Clone)] pub struct GatewayState { @@ -259,6 +284,7 @@ mod helper_tests { GatewayError::ProviderHttpError { status: 418, message: "teapot".into(), + retry_after: None, }, 418, ), @@ -329,6 +355,7 @@ mod helper_tests { GatewayError::ProviderHttpError { status: 503, message: "down".into(), + retry_after: None, }, ] { let expected = err.status_code(); diff --git a/crates/gateway/src/proxy/pipeline.rs b/crates/gateway/src/proxy/pipeline.rs index eea8871c..5d7c3d45 100644 --- a/crates/gateway/src/proxy/pipeline.rs +++ b/crates/gateway/src/proxy/pipeline.rs @@ -15,6 +15,7 @@ use super::identity::limits_for_ai_gateway; use super::{GatewayErrorResponse, GatewayRequestIdentity, GatewayState}; +use super::routing::Resend; use super::shaper::StreamShaper; use crate::lifecycle::{ ChatCompletionOutcome, ChatCompletionSurface, ChatPostInvokeDeps, Completed, OpenUpstream, @@ -112,14 +113,16 @@ fn short_circuit_to_response(outcome: ChatCompletionOutcome) -> GatewayErrorResp /// awaits the tail future, then drives `run_post_invoke` through /// record_outcome → write_cache → record_usage → emit_audit. pub(super) fn launch_stream_pump( - deps: ChatPostInvokeDeps, + mut deps: ChatPostInvokeDeps, open: OpenUpstream, + resend: Resend, shaper: StreamShaper, client: Dialect, client_sse: bool, ) -> axum::response::Response { let (response, tail) = build_chat_pump( open, + resend, shaper, client, client_sse, @@ -129,7 +132,17 @@ pub(super) fn launch_stream_pump( &deps.route.provider_name, ); tokio::spawn(async move { - let invoked = tail.await; + let (invoked, sent) = tail.await; + // A stream that went out again: the route's time is the attempt + // that ended it, and the row says how many retries there were. + if sent.retries > 0 { + deps.route.sel_record.started_at = sent.last_sent; + deps.route.sel_record.retries = sent.retries; + } + // The client left while the stream waited to go out again: the + // attempt that failed was counted against the route when the retry + // was decided, and nothing went out since. + deps.route.sel_record.counted = sent.waiting; run_post_invoke::(invoked, &deps).await; }); response diff --git a/crates/gateway/src/proxy/protocol_relearn.rs b/crates/gateway/src/proxy/protocol_relearn.rs index 42156a0c..c81c397d 100644 --- a/crates/gateway/src/proxy/protocol_relearn.rs +++ b/crates/gateway/src/proxy/protocol_relearn.rs @@ -27,7 +27,10 @@ use crate::protocol::UpstreamProtocol; /// once. Only 4xx bodies that name an API surface qualify — the shape /// upstreams actually use to report this. pub(super) fn is_protocol_mismatch(err: &GatewayError) -> bool { - let GatewayError::ProviderHttpError { status, message } = err else { + let GatewayError::ProviderHttpError { + status, message, .. + } = err + else { return false; }; // A 5xx is an upstream incident, not a dialect problem, and retrying @@ -104,10 +107,12 @@ mod tests { message: "The model 'anthropic.claude-opus-4' does not support the \ '/v1/chat/completions' API" .into(), + retry_after: None, })); assert!(is_protocol_mismatch(&GatewayError::ProviderHttpError { status: 404, message: "unsupported endpoint /v1/responses for this model".into(), + retry_after: None, })); } @@ -119,11 +124,13 @@ mod tests { assert!(!is_protocol_mismatch(&GatewayError::ProviderHttpError { status: 400, message: "messages: at least one message is required".into(), + retry_after: None, })); // Upstream incident, not a dialect problem. assert!(!is_protocol_mismatch(&GatewayError::ProviderHttpError { status: 503, message: "/v1/chat/completions is not available right now".into(), + retry_after: None, })); assert!(!is_protocol_mismatch(&GatewayError::UpstreamAuthError { status: 401, diff --git a/crates/gateway/src/proxy/responses_ws.rs b/crates/gateway/src/proxy/responses_ws.rs index 2e2ee8cf..9d67f6a9 100644 --- a/crates/gateway/src/proxy/responses_ws.rs +++ b/crates/gateway/src/proxy/responses_ws.rs @@ -110,6 +110,7 @@ async fn serve( headers.clone(), identity.clone(), None, + None, Bytes::from(Value::Object(body).to_string()), RESPONSES, "/v1/responses", diff --git a/crates/gateway/src/proxy/retry.rs b/crates/gateway/src/proxy/retry.rs new file mode 100644 index 00000000..50923680 --- /dev/null +++ b/crates/gateway/src/proxy/retry.rs @@ -0,0 +1,430 @@ +//! Sending a request again after the upstream failed for the moment. +//! +//! When every candidate route has been tried and the last one failed for +//! the moment, and none of the answer has reached the client, the request +//! waits and goes again to the routes that failed that way, instead of +//! handing the failure to the client. A model with one route — and a +//! stream, which is sent to the one route it picked — sends to it again. +//! +//! **For the moment** ([`transient`]): overloaded (503, 529, and the same +//! errors reported inside a stream), a server error (500, 502, 504), a +//! connection that could not be made or broke before the answer's head, a +//! stream that broke off before any of the answer. **Not** sent again: an +//! upstream's rate limit or spent quota (429), a refused credential (401, +//! 403), any other 4xx (a bad request, a model the upstream does not +//! have), a timeout, the gateway's own limits, budgets and policies, and a +//! client that left. Neither is anything after part of the answer went +//! out: the client would get its beginning twice. +//! +//! **How long it waits** ([`Budget`]): 2, 4, 8, then 16 seconds each +//! time, give or take a fifth so that many requests failing together do +//! not come back together; or as long as the upstream said to wait +//! (`retry-after-ms`, `Retry-After`) — but never less than +//! [`MIN_SAID_WAIT`]: an upstream that says "now" would otherwise be +//! asked again at once, hundreds of times over. It stops when the wait +//! would end past the request's deadline, `gateway.retry_wait_secs` after +//! the request arrived (0: never sent again). There is no count of its +//! own. An attempt already under way is not cut short by that deadline: +//! it only decides whether another one starts. +//! +//! A whole (non-streamed) answer has a second, hard deadline: the +//! gateway's request timeout (`super::REQUEST_TIMEOUT`), after which the +//! client gets an empty 408 and the request no log row. Each attempt is +//! cut at it, and a retry starts only when the wait and another attempt +//! as long as the last one fit before it (see [`Budget::within`]). +//! +//! Each attempt is a pick like any other: the strategy, the breaker, the +//! route caps, the session affinity and the provider holding a +//! conversation's encrypted history (`super::sealed`) all apply, and the +//! failed attempt counts against its route as before. + +use std::time::{Duration, Instant}; + +use rand::RngExt; + +use crate::error::GatewayError; + +/// What the client receives while the gateway holds back the opening of +/// a stream or waits to send again: an SSE comment, which every SSE +/// reader skips. Sent once nothing has gone out for [`KEEP_ALIVE`]. +pub(crate) const KEEP_ALIVE_FRAME: &[u8] = b": keep-alive\n\n"; + +/// How long the client may go without a byte while the gateway holds a +/// stream back. +pub(crate) const KEEP_ALIVE: Duration = Duration::from_secs(15); + +/// Did the upstream fail for the moment, so that the same request may +/// well be answered a little later? From what a stream records of its +/// error, as [`super::routing::fails`] reads it: the error's tag and +/// status. +/// +/// `transport` is a stream that broke off in transit (see +/// `lifecycle::broke_off`). `NetworkError` is a connection that could not +/// be made, a TLS failure, or one that broke before or while the answer +/// was read. A timeout (`ProviderTimeout`) is not: the upstream already +/// had all that time. +pub(crate) fn transient(error_tag: &str, status: i64) -> bool { + match error_tag { + "ProviderHttpError" => matches!(status, 500 | 502 | 503 | 504 | 529), + "NetworkError" | "transport" => true, + _ => false, + } +} + +/// [`transient`] for an error. +pub(crate) fn is_transient(err: &GatewayError) -> bool { + transient(err.error_tag(), err.status_code()) +} + +/// How long before the gateway's request timeout a whole answer's last +/// attempt is cut: time to write its error and its log row. +pub(crate) const TIMEOUT_MARGIN: Duration = Duration::from_secs(2); + +/// The hard deadline of a whole answer to a request that `arrived` at the +/// gateway port (see [`super::Arrived`]). +pub(crate) fn hard_deadline(arrived: Instant) -> Instant { + arrived + super::REQUEST_TIMEOUT.saturating_sub(TIMEOUT_MARGIN) +} + +/// The shortest wait an upstream's `retry-after-ms` / `Retry-After` is +/// taken for. Zero is taken as no word at all: the backoff applies. +pub(crate) const MIN_SAID_WAIT: Duration = Duration::from_millis(250); + +/// How long the upstream said to wait before trying again, held to +/// [`MIN_SAID_WAIT`]. +fn said_to_wait(err: &GatewayError) -> Option { + match err { + GatewayError::ProviderHttpError { retry_after, .. } => retry_after + .filter(|d| !d.is_zero()) + .map(|d| d.max(MIN_SAID_WAIT)), + _ => None, + } +} + +/// The wait before the `n`-th retry (1-based), before jitter: 2, 4, 8, +/// then 16 seconds. +fn backoff(n: u32) -> Duration { + Duration::from_secs(2u64 << n.clamp(1, 4).saturating_sub(1)) +} + +/// A request's allowance for sending again: its deadline, and how many +/// retries went out. +#[derive(Debug, Clone)] +pub(crate) struct Budget { + /// `None`: never sent again. + deadline: Option, + /// When the answer must be back by, whatever the retries: a whole + /// answer's request timeout. `None` for a stream, whose body is not + /// bound by it. + hard: Option, + retries: u32, +} + +impl Budget { + /// The allowance of a request that arrived at `arrived`: `wait` from + /// then. Zero: none. + pub(crate) fn new(arrived: Instant, wait: Duration) -> Self { + Self { + deadline: (!wait.is_zero()) + .then(|| arrived.checked_add(wait)) + .flatten(), + hard: None, + retries: 0, + } + } + + /// The same allowance for a request that must be answered by `hard`: + /// each attempt is cut there ([`Budget::hard_deadline`]), and a retry + /// starts only when its wait and another attempt as long as the last + /// one fit before it. + pub(crate) fn within(mut self, hard: Instant) -> Self { + self.hard = Some(hard); + self + } + + /// When the attempt under way must be given up, if ever. + pub(crate) fn hard_deadline(&self) -> Option { + self.hard + } + + /// Whether a failure now could still be followed by another attempt. + pub(crate) fn open(&self) -> bool { + self.deadline.is_some_and(|d| Instant::now() < d) + } + + /// How long to wait before sending again after `err`. `None`: `err` + /// is the answer — it is not transient, or the wait would end past + /// the deadline. + pub(crate) fn wait_after(&self, err: &GatewayError) -> Option { + self.wait_after_taking(err, Duration::ZERO) + } + + /// [`Budget::wait_after`] for an attempt that took `took`: under a + /// hard deadline, the next one is expected to take as long. + pub(crate) fn wait_after_taking(&self, err: &GatewayError, took: Duration) -> Option { + if !is_transient(err) { + return None; + } + self.wait(said_to_wait(err), took) + } + + /// [`Budget::wait_after`] for what a stream recorded of its error. + pub(crate) fn wait_after_outcome(&self, error_tag: &str, status: i64) -> Option { + if !transient(error_tag, status) { + return None; + } + self.wait(None, Duration::ZERO) + } + + fn wait(&self, said: Option, took: Duration) -> Option { + let jitter = rand::rng().random_range(0.8..=1.2); + self.wait_at(Instant::now(), said, jitter, took) + } + + /// The wait before the next retry, `jitter` the factor the backoff is + /// spread by: the upstream's own word if it gave one, else the + /// backoff. `None` when it would end past the deadline, or when it and + /// another attempt as long as the last (`took`) would not fit before + /// the hard deadline. + fn wait_at( + &self, + now: Instant, + said: Option, + jitter: f64, + took: Duration, + ) -> Option { + let deadline = self.deadline?; + let wait = said.unwrap_or_else(|| backoff(self.retries + 1).mul_f64(jitter)); + let resumes = now.checked_add(wait)?; + if resumes > deadline { + return None; + } + if let Some(hard) = self.hard + && resumes.checked_add(took)? > hard + { + return None; + } + Some(wait) + } + + /// The next attempt went out. + pub(crate) fn sent(&mut self) { + self.retries += 1; + } + + /// How many retries went out. + pub(crate) fn retries(&self) -> u32 { + self.retries + } +} + +/// `gateway_provider_retry_total{provider}`: an attempt that went out +/// again after a transient failure, by the provider it went to. Next to +/// `gateway_provider_fallback_total`, which counts moving on to another +/// route. +pub(crate) fn count(provider: &str) { + metrics::counter!( + "gateway_provider_retry_total", + "provider" => crate::metrics_labels::normalize_provider_label(provider), + ) + .increment(1); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn http(status: u16, retry_after: Option) -> GatewayError { + GatewayError::ProviderHttpError { + status, + message: "p: down".into(), + retry_after, + } + } + + #[test] + fn only_a_failure_of_the_moment_is_transient() { + for status in [500, 502, 503, 504, 529] { + assert!(is_transient(&http(status, None)), "{status}"); + } + for status in [400, 404, 408, 413, 422, 501] { + assert!(!is_transient(&http(status, None)), "{status}"); + } + assert!(is_transient(&GatewayError::NetworkError("reset".into()))); + // A stream broken off before any of the answer. + assert!(transient("transport", 502)); + + for e in [ + GatewayError::UpstreamRateLimited { + retry_after_secs: Some(1), + }, + GatewayError::UpstreamAuthError { + status: 401, + message: String::new(), + }, + GatewayError::ProviderTimeout("slow".into()), + GatewayError::ProviderInvalidResponse("not json".into()), + GatewayError::ProviderError("signing failed".into()), + GatewayError::TransformError("bad".into()), + GatewayError::PolicyBlocked("rule".into()), + GatewayError::rate_limited("user:requests/1m", 5), + GatewayError::budget_exhausted("user:budget/monthly", 5), + GatewayError::SealedHistoryRefused { + status: 400, + provider: "p".into(), + said: "invalid_encrypted_content".into(), + }, + ] { + assert!(!is_transient(&e), "{e:?}"); + } + // A stream the client left, or one that ended normally. + assert!(!transient("ClientCancelled", 499)); + } + + #[test] + fn the_backoff_doubles_up_to_sixteen_seconds() { + let secs: Vec = (1..=6).map(|n| backoff(n).as_secs()).collect(); + assert_eq!(secs, [2, 4, 8, 16, 16, 16]); + } + + #[test] + fn the_wait_is_spread_by_a_fifth_either_way() { + let now = Instant::now(); + let budget = Budget::new(now, Duration::from_secs(120)); + assert_eq!( + budget.wait_at(now, None, 0.8, Duration::ZERO), + Some(Duration::from_millis(1600)) + ); + assert_eq!( + budget.wait_at(now, None, 1.2, Duration::ZERO), + Some(Duration::from_millis(2400)) + ); + for _ in 0..100 { + let w = budget.wait(None, Duration::ZERO).unwrap(); + assert!( + w >= Duration::from_millis(1600) && w <= Duration::from_millis(2400), + "{w:?}" + ); + } + } + + #[test] + fn each_retry_waits_longer() { + let now = Instant::now(); + let mut budget = Budget::new(now, Duration::from_secs(600)); + let mut waits = Vec::new(); + for _ in 0..5 { + waits.push( + budget + .wait_at(now, None, 1.0, Duration::ZERO) + .unwrap() + .as_secs(), + ); + budget.sent(); + } + assert_eq!(waits, [2, 4, 8, 16, 16]); + assert_eq!(budget.retries(), 5); + } + + #[test] + fn the_upstreams_own_wait_wins_over_the_backoff() { + let now = Instant::now(); + let budget = Budget::new(now, Duration::from_secs(120)); + let said = Some(Duration::from_millis(300)); + assert_eq!(budget.wait_at(now, said, 1.2, Duration::ZERO), said); + assert_eq!( + budget.wait_at(now, Some(Duration::from_secs(30)), 1.0, Duration::ZERO), + Some(Duration::from_secs(30)) + ); + // Longer than what is left: the failure is the answer. + assert_eq!( + budget.wait_at(now, Some(Duration::from_secs(121)), 1.0, Duration::ZERO), + None + ); + } + + /// "Now" is no word at all, and "almost now" is held to a floor: an + /// upstream that says so must not be asked again hundreds of times. + #[test] + fn a_wait_of_nothing_from_the_upstream_is_not_taken_at_its_word() { + assert_eq!(said_to_wait(&http(503, Some(Duration::ZERO))), None); + assert_eq!( + said_to_wait(&http(503, Some(Duration::from_millis(1)))), + Some(MIN_SAID_WAIT) + ); + assert_eq!( + said_to_wait(&http(503, Some(Duration::from_secs(3)))), + Some(Duration::from_secs(3)) + ); + // Zero falls back to the backoff. + let budget = Budget::new(Instant::now(), Duration::from_secs(120)); + let w = budget.wait_after(&http(503, Some(Duration::ZERO))).unwrap(); + assert!(w >= Duration::from_millis(1600), "{w:?}"); + } + + /// A whole answer's hard deadline: a retry starts only when its wait + /// and another attempt as long as the last fit before it. + #[test] + fn a_retry_must_fit_before_the_hard_deadline() { + let arrived = Instant::now(); + let budget = Budget::new(arrived, Duration::from_secs(600)) + .within(arrived + Duration::from_secs(118)); + assert_eq!( + budget.hard_deadline(), + Some(arrived + Duration::from_secs(118)) + ); + // A 504 after 60 s, 62 s in: a 2 s wait and another 60 s do not fit. + let at = arrived + Duration::from_secs(62); + let minute = Duration::from_secs(60); + assert_eq!(budget.wait_at(at, None, 1.0, minute), None); + // A quick failure does. + assert_eq!( + budget.wait_at(at, None, 1.0, Duration::from_millis(50)), + Some(Duration::from_secs(2)) + ); + // Without a hard deadline (a stream), only the soft one counts. + let stream = Budget::new(arrived, Duration::from_secs(600)); + assert_eq!(stream.hard_deadline(), None); + assert!(stream.wait_at(at, None, 1.0, minute).is_some()); + } + + #[test] + fn no_wait_ends_past_the_deadline() { + let arrived = Instant::now(); + let budget = Budget::new(arrived, Duration::from_secs(10)); + assert!(budget.open()); + // 9 s in, 1 s left: a 2 s backoff does not fit. + assert_eq!( + budget.wait_at(arrived + Duration::from_secs(9), None, 1.0, Duration::ZERO), + None + ); + assert_eq!( + budget.wait_at(arrived + Duration::from_secs(8), None, 1.0, Duration::ZERO), + Some(Duration::from_secs(2)) + ); + } + + #[test] + fn a_zero_wait_never_sends_again() { + let now = Instant::now(); + let budget = Budget::new(now, Duration::ZERO); + assert!(!budget.open()); + assert_eq!(budget.wait_at(now, None, 1.0, Duration::ZERO), None); + assert_eq!(budget.wait_after(&http(503, None)), None); + } + + #[test] + fn a_failure_that_is_not_transient_is_the_answer() { + let budget = Budget::new(Instant::now(), Duration::from_secs(120)); + assert!(budget.wait_after(&http(503, None)).is_some()); + assert_eq!(budget.wait_after(&http(400, None)), None); + assert_eq!( + budget.wait_after(&GatewayError::UpstreamRateLimited { + retry_after_secs: None + }), + None + ); + assert!(budget.wait_after_outcome("transport", 502).is_some()); + assert_eq!(budget.wait_after_outcome("ProviderHttpError", 400), None); + } +} diff --git a/crates/gateway/src/proxy/routing.rs b/crates/gateway/src/proxy/routing.rs index c9d29a99..e130e2fa 100644 --- a/crates/gateway/src/proxy/routing.rs +++ b/crates/gateway/src/proxy/routing.rs @@ -16,6 +16,7 @@ use crate::route_caps::{self, Admission, CappedRoutes, RouteCaps}; use crate::router::{AffinityMode, RouteEntry}; use crate::strategy::{self, RoutingStrategy}; +use super::retry::Budget; use super::sealed::Pin; /// What the affinity layer can pin a session to. @@ -293,17 +294,27 @@ pub(crate) struct SelectionRecord { /// at finalize time — which is correct for streams since there's /// no failover loop inflating the e2e timer. pub picked_latency_ms: Option, + /// How many times the request went out again after a transient + /// failure before this route answered (see `super::retry`). Said on + /// the log row as `"retries"`. + pub retries: u32, + /// The request's last attempt was already counted against the route: + /// a stream whose client left while it waited to go out again. + pub counted: bool, } /// Record health for the picked route once the request finishes. /// Best-effort; failures are logged. pub(crate) async fn finalize_health(state: &GatewayState, sel: &SelectionRecord, success: bool) { + if sel.counted { + return; + } let total_latency_ms = sel.started_at.elapsed().as_millis().min(u32::MAX as u128) as u32; // For health, we want the picked route's *own* time, not the // cumulative including prior failed attempts. Non-stream sets // this explicitly when it picks a winner; stream falls back to - // total elapsed (which is the picked route's time anyway since - // streams don't loop). + // total elapsed, which is the picked route's time: a stream that + // went out again starts `started_at` at the attempt that answered. let health_latency_ms = sel.picked_latency_ms.unwrap_or(total_latency_ms); let breaker = resolve_breaker_config(state).await; let _ = state @@ -405,12 +416,24 @@ async fn pick_admitted<'a>( /// fails (see [`is_upstream_failure`]) tries another candidate from the /// remaining set until exhausted. When every candidate left is at its /// cap and none failed, the answer is the caps' 429. +/// +/// Under `budget`'s hard deadline (a whole answer's request timeout), +/// each attempt is cut there and the last failure, or a timeout, is the +/// answer. +/// +/// When every candidate is exhausted and the last one failed for the +/// moment (see `super::retry`), the request waits as `budget` allows and +/// goes round again, to the providers whose last failure in this request +/// was transient — picked as before, so the strategy, the breaker, the +/// caps, the affinity and a sealed conversation's pin all apply. A model +/// with one route sends to it again. pub(super) async fn select_route_with_failover<'a>( routes: &'a [RouteEntry], outbound: &super::generate::Outbound, call_ctx: &CallCtx, ctx: &SelectionCtx<'_>, caller_model: &str, + budget: &mut Budget, ) -> Result, GatewayError> { let started_at = std::time::Instant::now(); let candidates: Vec<&RouteEntry> = routes.iter().collect(); @@ -418,102 +441,185 @@ pub(super) async fn select_route_with_failover<'a>( let mut last_error: Option = None; let mut tried: Vec = Vec::new(); let mut capped = CappedRoutes::default(); + // The providers whose last failure in this request was transient, in + // the order they failed: the ones a retry goes back to. + let mut transient: Vec = Vec::new(); + // How long the last attempt took: under a hard deadline, a retry + // starts only when one as long fits after its wait. + let mut last_took = std::time::Duration::ZERO; + + loop { + let retrying = budget.retries() > 0; + let mut attempted = false; + for _ in 0..candidates.len() { + let Some(entry) = pick_admitted(&candidates, &tried, &mut capped, ctx).await? else { + break; + }; + tried.push(entry.provider_id); + attempted = true; + if retrying { + super::retry::count(&entry.provider_name); + } - for _ in 0..candidates.len() { - let Some(entry) = pick_admitted(&candidates, &tried, &mut capped, ctx).await? else { - break; - }; - tried.push(entry.provider_id); - - let upstream_model = entry.upstream_model.as_deref().unwrap_or(caller_model); - - // Per-attempt clock: record latency against this route as - // *its* time, not "everything since the request started" - // (which would double-count earlier failed attempts in - // a failover chain and skew the latency strategy). - let attempt_started_at = std::time::Instant::now(); - // A rejected dialect is retried inside `send`, on this same route. - let result = - match super::generate::send(entry, outbound, call_ctx, &ctx.state.db, upstream_model) + let upstream_model = entry.upstream_model.as_deref().unwrap_or(caller_model); + + // Per-attempt clock: record latency against this route as + // *its* time, not "everything since the request started" + // (which would double-count earlier failed attempts in + // a failover chain and skew the latency strategy). + let attempt_started_at = std::time::Instant::now(); + // A rejected dialect is retried inside `send`, on this same route. + let attempt = async { + match super::generate::send( + entry, + outbound, + call_ctx, + &ctx.state.db, + upstream_model, + ) .await - { - Ok((resp, wire)) => { - super::generate::read_whole(resp, &wire, caller_model, outbound.input_estimate) - .await - .map(|answer| (answer, wire.ledger)) + { + Ok((resp, wire)) => super::generate::read_whole( + resp, + &wire, + caller_model, + outbound.input_estimate, + ) + .await + .map(|answer| (answer, wire.ledger)), + Err(e) => Err(e), + } + }; + // Cut at the hard deadline, so that the answer is an error the + // client and the log are told, not the request timeout's 408. + let result = match budget.hard_deadline() { + Some(hard) => { + tokio::time::timeout_at(tokio::time::Instant::from_std(hard), attempt).await } - Err(e) => Err(e), + None => Ok(attempt.await), }; - let attempt_latency_ms = attempt_started_at - .elapsed() - .as_millis() - .min(u32::MAX as u128) as u32; - - match result { - Ok((response, ledger)) => { - set_affinity( - &ctx.state.redis, - ctx.user_id, - ctx.model_id, - ctx.affinity_mode, - entry, - ctx.affinity_ttl_secs, - ) - .await; - return Ok(( - entry, - response, - ledger, - SelectionRecord { - picked_route_id: entry.route_id, - started_at, - picked_latency_ms: Some(attempt_latency_ms), - }, - )); - } - Err(e) if is_upstream_failure(&e) => { + last_took = attempt_started_at.elapsed(); + let attempt_latency_ms = last_took.as_millis().min(u32::MAX as u128) as u32; + + let Ok(result) = result else { tracing::warn!( provider = %entry.provider_name, provider_id = %entry.provider_id, - error = %e, - "Route failed, trying next" + "Route still had not answered at the gateway's request timeout" ); - metrics::counter!( - "gateway_provider_fallback_total", - "from" => crate::metrics_labels::normalize_provider_label(&entry.provider_name), - ) - .increment(1); - // Record the failed attempt in health so the - // breaker can trip mid-failover. let _ = ctx .state .health .record(entry.route_id, attempt_latency_ms, true, ctx.breaker) .await; - last_error = Some(e); - continue; - } - Err(e) => { - // The request itself was refused. Sibling routes would - // refuse it the same way, so it goes back to the caller. - // An upstream that answered with a refusal is working: - // it counts as a success for the route, as it does on - // the desktop. - if matches!( - e, - GatewayError::ProviderHttpError { .. } - | GatewayError::SealedHistoryRefused { .. } - ) { + // An earlier failure says more than the cut; with none, + // the upstream simply took too long. + return Err(last_error.unwrap_or_else(|| { + GatewayError::ProviderTimeout(format!( + "{} did not answer within the gateway's request timeout", + entry.provider_name + )) + })); + }; + + match result { + Ok((response, ledger)) => { + set_affinity( + &ctx.state.redis, + ctx.user_id, + ctx.model_id, + ctx.affinity_mode, + entry, + ctx.affinity_ttl_secs, + ) + .await; + return Ok(( + entry, + response, + ledger, + SelectionRecord { + picked_route_id: entry.route_id, + started_at, + picked_latency_ms: Some(attempt_latency_ms), + retries: budget.retries(), + counted: false, + }, + )); + } + Err(e) if is_upstream_failure(&e) => { + tracing::warn!( + provider = %entry.provider_name, + provider_id = %entry.provider_id, + error = %e, + "Route failed, trying next" + ); + metrics::counter!( + "gateway_provider_fallback_total", + "from" => crate::metrics_labels::normalize_provider_label(&entry.provider_name), + ) + .increment(1); + // Record the failed attempt in health so the + // breaker can trip mid-failover. let _ = ctx .state .health - .record(entry.route_id, attempt_latency_ms, false, ctx.breaker) + .record(entry.route_id, attempt_latency_ms, true, ctx.breaker) .await; + transient.retain(|p| *p != entry.provider_id); + if super::retry::is_transient(&e) { + transient.push(entry.provider_id); + } + last_error = Some(e); + continue; + } + Err(e) => { + // The request itself was refused. Sibling routes would + // refuse it the same way, so it goes back to the caller. + // An upstream that answered with a refusal is working: + // it counts as a success for the route, as it does on + // the desktop. + if matches!( + e, + GatewayError::ProviderHttpError { .. } + | GatewayError::SealedHistoryRefused { .. } + ) { + let _ = ctx + .state + .health + .record(entry.route_id, attempt_latency_ms, false, ctx.breaker) + .await; + } + return Err(e); } - return Err(e); } } + + // Every candidate is exhausted. Go round again only when this + // round sent something, its last failure was transient, and the + // wait fits before the deadline. + let wait = match &last_error { + Some(e) if attempted && !transient.is_empty() => budget.wait_after_taking(e, last_took), + _ => None, + }; + let Some(wait) = wait else { break }; + tracing::info!( + model = ctx.model_id, + retry = budget.retries() + 1, + wait_ms = wait.as_millis() as u64, + error = ?last_error.as_ref().map(GatewayError::error_tag), + "Every route failed for the moment, sending again after a wait" + ); + tokio::time::sleep(wait).await; + budget.sent(); + // Only the providers that failed for the moment are asked again. + // Caps are counted afresh: a window full then may have room now. + tried = candidates + .iter() + .map(|e| e.provider_id) + .filter(|p| !transient.contains(p)) + .collect(); + capped = CappedRoutes::default(); } // An upstream that failed is the answer's cause even when the rest @@ -547,6 +653,8 @@ pub(super) async fn select_route_for_stream<'a>( // back to elapsed-from-started_at, which IS the // picked route's wall time for the streaming case. picked_latency_ms: None, + retries: 0, + counted: false, }, )); } @@ -556,6 +664,67 @@ pub(super) async fn select_route_for_stream<'a>( })) } +/// What a stream needs to go out again to its route after a transient +/// failure before any of the answer reached the client (see +/// `super::retry`). A stream is sent to the one route it picked, so that +/// is where it goes again, under the pin and affinity it was picked with. +pub(crate) struct Resend { + state: GatewayState, + route_id: Uuid, + provider: String, + caps: RouteCaps, + breaker: CircuitBreakerConfig, + fail_closed: bool, + /// The route's provider alone can read the conversation's encrypted + /// history: asked again with its breaker open, as it was picked. + pinned_only: bool, + pub(crate) budget: Budget, +} + +impl Resend { + pub(super) fn new(entry: &RouteEntry, ctx: &SelectionCtx<'_>, budget: Budget) -> Self { + Self { + state: ctx.state.clone(), + route_id: entry.route_id, + provider: entry.provider_name.clone(), + caps: RouteCaps::of(entry), + breaker: ctx.breaker, + fail_closed: ctx.fail_closed, + pinned_only: matches!(ctx.pin, Some(Pin::Only(p)) if p == entry.provider_id), + budget, + } + } + + /// Before the wait: may the route be sent to again? Not while its + /// breaker is open, nor at one of its caps — the pick a failover makes; + /// the retry is counted against the caps now. When it may, the attempt + /// that failed (`failed_ms` long) counts against the route, as a failed + /// attempt in a failover does, and the retry is counted. When it may + /// not, the failure is the answer and the stream's tail records it. + pub(crate) async fn admit(&mut self, failed_ms: u32) -> bool { + let health = self + .state + .health + .snapshot(self.route_id, self.breaker) + .await; + if health.state == tw_breaker::State::Open && !self.pinned_only { + return false; + } + match route_caps::admit(&self.state.redis, &self.caps, self.fail_closed).await { + Ok(Admission::Admitted) => {} + Ok(Admission::Capped { .. }) | Err(_) => return false, + } + let _ = self + .state + .health + .record(self.route_id, failed_ms, true, self.breaker) + .await; + self.budget.sent(); + super::retry::count(&self.provider); + true + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/gateway/src/proxy/sealed.rs b/crates/gateway/src/proxy/sealed.rs index 28acdeef..b8b98d2b 100644 --- a/crates/gateway/src/proxy/sealed.rs +++ b/crates/gateway/src/proxy/sealed.rs @@ -179,9 +179,9 @@ pub(crate) fn explain_refusal( e: GatewayError, ) -> GatewayError { match e { - GatewayError::ProviderHttpError { status, message } - if (400..500).contains(&status) && refused(wire, &message) => - { + GatewayError::ProviderHttpError { + status, message, .. + } if (400..500).contains(&status) && refused(wire, &message) => { let said = message .strip_prefix(label) .and_then(|m| m.strip_prefix(": ")) @@ -294,6 +294,7 @@ mod tests { let e = GatewayError::ProviderHttpError { status: 400, message: format!("relay-b: {body}"), + retry_after: None, }; let e = explain_refusal(Dialect::Responses, "relay-b", "relay-b", e); let GatewayError::SealedHistoryRefused { @@ -323,6 +324,7 @@ mod tests { let other = || GatewayError::ProviderHttpError { status: 400, message: "p: max_output_tokens is too large".into(), + retry_after: None, }; assert!(matches!( explain_refusal(Dialect::Responses, "p", "p", other()), @@ -331,6 +333,7 @@ mod tests { let sealed = || GatewayError::ProviderHttpError { status: 400, message: "p: invalid_encrypted_content".into(), + retry_after: None, }; // Not sent as Responses: nothing of OpenAI's was sealed in it. assert!(matches!( @@ -341,6 +344,7 @@ mod tests { let e = GatewayError::ProviderHttpError { status: 500, message: "p: invalid_encrypted_content".into(), + retry_after: None, }; assert!(matches!( explain_refusal(Dialect::Responses, "p", "p", e), diff --git a/crates/gateway/src/proxy/transport.rs b/crates/gateway/src/proxy/transport.rs index 3e862b06..6da9652f 100644 --- a/crates/gateway/src/proxy/transport.rs +++ b/crates/gateway/src/proxy/transport.rs @@ -209,8 +209,14 @@ impl Upstream { /// A request that never got an answer: the upstream timed out, or the /// connection could not be made or broke. Either way it says nothing /// about the request, and another route may well answer it. +/// +/// A request that could not even be built — a provider whose base URL is +/// no URL — is the route's configuration, not a failure of the moment: +/// sending it again would fail the same way (see `retry::transient`). pub(crate) fn transport_error(e: reqwest::Error) -> GatewayError { - if e.is_timeout() { + if e.is_builder() { + GatewayError::ProviderError(format!("The upstream request could not be built: {e}")) + } else if e.is_timeout() { GatewayError::ProviderTimeout(e.to_string()) } else { GatewayError::NetworkError(e.to_string()) @@ -229,19 +235,41 @@ pub(crate) fn transport_error(e: reqwest::Error) -> GatewayError { /// Every other status is kept as it is, in `ProviderHttpError`. Whether /// it is the upstream failing (5xx, 408) or the upstream refusing this /// request (any other 4xx) decides failover and the circuit breaker — -/// see `routing::is_upstream_failure`. +/// see `routing::is_upstream_failure` — and whether it is worth sending +/// again after a wait, `proxy::retry`. The error keeps how long the +/// upstream said to wait. pub(crate) fn status_error( status: u16, - retry_after_secs: Option, + retry_after: Option, message: String, ) -> GatewayError { match status { - 429 => GatewayError::UpstreamRateLimited { retry_after_secs }, + 429 => GatewayError::UpstreamRateLimited { + retry_after_secs: retry_after.map(|d| u32::try_from(d.as_secs()).unwrap_or(u32::MAX)), + }, 401 | 403 => GatewayError::UpstreamAuthError { status, message }, - _ => GatewayError::ProviderHttpError { status, message }, + _ => GatewayError::ProviderHttpError { + status, + message, + retry_after, + }, } } +/// When an answer says to try again: OpenAI's `retry-after-ms`, or +/// `Retry-After` in seconds (see [`crate::error::parse_retry_after_seconds`]). +fn retry_after(headers: &reqwest::header::HeaderMap) -> Option { + let header = |name: &str| headers.get(name).and_then(|v| v.to_str().ok()); + header("retry-after-ms") + .and_then(|v| v.trim().parse::().ok()) + .map(Duration::from_millis) + .or_else(|| { + header(reqwest::header::RETRY_AFTER.as_str()) + .and_then(crate::error::parse_retry_after_seconds) + .map(|s| Duration::from_secs(u64::from(s))) + }) +} + /// Turn a non-2xx upstream answer into the error the caller sees (see /// [`status_error`]). /// @@ -256,14 +284,16 @@ async fn check_status( ) -> Result { let status = resp.status(); if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + // Echoed to the caller in whole seconds, as the upstream wrote it. let retry_after_secs = resp .headers() .get(reqwest::header::RETRY_AFTER) .and_then(|v| v.to_str().ok()) .and_then(crate::error::parse_retry_after_seconds); - return Err(status_error(429, retry_after_secs, String::new())); + return Err(GatewayError::UpstreamRateLimited { retry_after_secs }); } if !status.is_success() { + let retry_after = retry_after(resp.headers()); let body = resp.text().await.unwrap_or_default(); tracing::warn!(provider = label, status = %status, body = %body, "upstream returned non-2xx"); const CLIENT_MAX: usize = 512; @@ -279,7 +309,7 @@ async fn check_status( }; return Err(status_error( status.as_u16(), - None, + retry_after, format!("{label}: {shown}"), )); } @@ -393,6 +423,51 @@ mod tests { .into() } + /// A base URL that is no URL fails every time: not worth sending + /// again, unlike a connection that could not be made. + #[tokio::test] + async fn a_request_that_cannot_be_built_is_not_a_failure_of_the_moment() { + let built = client().post("http://exa mple.invalid/v1").send().await; + let e = transport_error(built.unwrap_err()); + assert!(matches!(e, GatewayError::ProviderError(_)), "{e:?}"); + assert!(!crate::proxy::retry::is_transient(&e)); + + let refused = client().post("http://127.0.0.1:1/v1").send().await; + let e = transport_error(refused.unwrap_err()); + assert!(matches!(e, GatewayError::NetworkError(_)), "{e:?}"); + assert!(crate::proxy::retry::is_transient(&e)); + } + + #[tokio::test] + async fn an_overloaded_answer_keeps_how_long_to_wait() { + let with = |headers: &[(&str, &str)]| { + let mut b = http_1x::Response::builder().status(503); + for (k, v) in headers { + b = b.header(*k, *v); + } + reqwest::Response::from(b.body("busy".to_string()).unwrap()) + }; + let wait = |e: GatewayError| match e { + GatewayError::ProviderHttpError { retry_after, .. } => retry_after, + e => panic!("{e:?}"), + }; + let e = check_status( + with(&[("retry-after-ms", "250"), ("retry-after", "3")]), + "p", + ) + .await + .unwrap_err(); + assert_eq!(wait(e), Some(Duration::from_millis(250))); + let e = check_status(with(&[("retry-after", "3")]), "p") + .await + .unwrap_err(); + assert_eq!(wait(e), Some(Duration::from_secs(3))); + let e = check_status(with(&[("retry-after", "soon")]), "p") + .await + .unwrap_err(); + assert_eq!(wait(e), None); + } + #[tokio::test] async fn a_refused_credential_keeps_the_upstreams_reason_out_of_the_callers_sight() { let body = diff --git a/crates/server/src/app.rs b/crates/server/src/app.rs index 4b9d02ed..ee513950 100644 --- a/crates/server/src/app.rs +++ b/crates/server/src/app.rs @@ -409,9 +409,12 @@ pub async fn create_gateway_app(_config: &AppConfig, state: AppState) -> anyhow: .merge(key_routes) .merge(mcp_routes) .layer(RequestBodyLimitLayer::new(10 * 1024 * 1024)) // 10MB for large prompts + // When the timeout below starts counting: a whole answer's retries + // stop in time for its error to be written (see `gateway_proxy::retry`). + .layer(axum::middleware::from_fn(gateway_proxy::stamp_arrival)) .layer(TimeoutLayer::with_status_code( axum::http::StatusCode::REQUEST_TIMEOUT, - std::time::Duration::from_secs(120), // longer timeout for streaming + gateway_proxy::REQUEST_TIMEOUT, )) .layer(gateway_cors) .layer(SetResponseHeaderLayer::overriding( diff --git a/crates/server/src/handlers/admin/settings.rs b/crates/server/src/handlers/admin/settings.rs index b373948d..7e7a3cc9 100644 --- a/crates/server/src/handlers/admin/settings.rs +++ b/crates/server/src/handlers/admin/settings.rs @@ -656,6 +656,17 @@ fn validate_setting(key: &str, value: &serde_json::Value) -> Result<(), AppError )); } } + "gateway.retry_wait_secs" => { + let v = value + .as_i64() + .ok_or_else(|| AppError::BadRequest(format!("{key} must be an integer")))?; + let max = think_watch_common::dynamic_config::RETRY_WAIT_MAX_SECS; + if !(0..=max).contains(&v) { + return Err(AppError::BadRequest(format!( + "retry_wait_secs must be between 0 and {max}" + ))); + } + } "gateway.latency_strategy_k" => { let v = value .as_f64() @@ -751,6 +762,16 @@ mod tests { assert!(validate_setting("api_keys.default_expiry_days", &json!(-1)).is_err()); } + #[test] + fn the_retry_wait_takes_zero_up_to_ten_minutes() { + assert!(validate_setting("gateway.retry_wait_secs", &json!(0)).is_ok()); + assert!(validate_setting("gateway.retry_wait_secs", &json!(120)).is_ok()); + assert!(validate_setting("gateway.retry_wait_secs", &json!(600)).is_ok()); + assert!(validate_setting("gateway.retry_wait_secs", &json!(601)).is_err()); + assert!(validate_setting("gateway.retry_wait_secs", &json!(-1)).is_err()); + assert!(validate_setting("gateway.retry_wait_secs", &json!("120")).is_err()); + } + #[test] fn validates_the_public_host_with_the_gateway_address_rule() { for ok in [ diff --git a/crates/server/src/protocol_probe.rs b/crates/server/src/protocol_probe.rs index 6b0171ca..1d42d3ea 100644 --- a/crates/server/src/protocol_probe.rs +++ b/crates/server/src/protocol_probe.rs @@ -456,6 +456,7 @@ mod tests { assert!(is_inconclusive(&GatewayError::ProviderHttpError { status: 503, message: "OpenAI: overloaded".into(), + retry_after: None, })); } @@ -491,10 +492,12 @@ mod tests { message: "OpenAI: The model 'openai.gpt-5.5' does not support \ the '/v1/chat/completions' API" .into(), + retry_after: None, })); assert!(!is_inconclusive(&GatewayError::ProviderHttpError { status: 404, message: "no such model".into(), + retry_after: None, })); } diff --git a/crates/test-support/src/lib.rs b/crates/test-support/src/lib.rs index fc0791f9..ba3ed974 100644 --- a/crates/test-support/src/lib.rs +++ b/crates/test-support/src/lib.rs @@ -237,6 +237,12 @@ impl TestApp { config.validate().map_err(anyhow::Error::msg)?; let mut state = init::init_state(config.clone(), db.clone(), redis, ch_client).await?; + // An upstream that fails in a test is not sent to again: waiting + // out two minutes of retries on every failing upstream would only + // slow the suite. Tests of the retries turn them on with + // `set_setting("gateway.retry_wait_secs", …)`. + fixtures::set_setting(&db, "gateway.retry_wait_secs", serde_json::json!(0)).await?; + state.dynamic_config.reload().await?; if let Some(v) = opts.url_validator { // Every outbound fetch goes through it — webhook and Kafka // deliveries included, not only the admin handlers. diff --git a/crates/test-support/tests/analytics_clickhouse.rs b/crates/test-support/tests/analytics_clickhouse.rs index ba1e0296..5e9b849b 100644 --- a/crates/test-support/tests/analytics_clickhouse.rs +++ b/crates/test-support/tests/analytics_clickhouse.rs @@ -648,7 +648,7 @@ async fn an_error_the_upstream_reports_in_its_stream_fails_the_request() { assert_failed_partway( status, &detail, - (502, "ProviderError"), + (500, "ProviderHttpError"), "The model crashed.", ); } @@ -790,7 +790,7 @@ async fn a_caller_that_leaves_after_response_failed_saw_the_request_fail() { assert_failed_partway( rows[0].0, &rows[0].1, - (502, "ProviderError"), + (500, "ProviderHttpError"), "The model crashed.", ); } @@ -957,3 +957,91 @@ async fn v1_usage_reports_the_cost_this_month_from_the_request_log() { "b={b} one call={one_call}" ); } + +/// A request that went out again after a transient failure says how many +/// times on its log row — whole answer and stream alike. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_retried_request_says_so_on_its_log_row() { + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let app = TestApp::spawn_with_clickhouse().await; + app.set_setting("gateway.retry_wait_secs", json!(30)).await; + let model = "gpt-retried"; + let upstream = MockServer::start().await; + // Every request is overloaded once, then answered. + let overloaded = || { + ResponseTemplate::new(503) + .insert_header("retry-after-ms", "250") + .set_body_json(json!({"error": {"message": "overloaded"}})) + }; + let chunk = |delta: Value, finish: Value| { + format!( + "data: {}\n\n", + json!({"id": "c", "object": "chat.completion.chunk", "created": 1, "model": model, + "choices": [{"index": 0, "delta": delta, "finish_reason": finish}], + "usage": null}) + ) + }; + let stream = [ + chunk(json!({"content": "hi"}), Value::Null), + chunk(json!({}), json!("stop")), + "data: [DONE]\n\n".to_string(), + ] + .concat(); + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(overloaded()) + .up_to_n_times(1) + .mount(&upstream) + .await; + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "id": "chatcmpl-retried", "object": "chat.completion", "created": 1, "model": model, + "choices": [{"index": 0, "message": {"role": "assistant", "content": "hi"}, + "finish_reason": "stop"}], + "usage": {"prompt_tokens": 7, "completion_tokens": 3, "total_tokens": 10} + }))) + .up_to_n_times(1) + .mount(&upstream) + .await; + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(overloaded()) + .up_to_n_times(1) + .mount(&upstream) + .await; + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(ResponseTemplate::new(200).set_body_raw(stream, "text/event-stream")) + .mount(&upstream) + .await; + let (key, user_id) = seed_upstream(&app, &upstream.uri(), "openai", model).await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + // Two prompts: the second must not be the first one's cache hit. + for (stream, prompt) in [(false, "hi"), (true, "hello")] { + let resp = gw + .post( + "/v1/chat/completions", + json!({"model": model, "stream": stream, + "messages": [{"role": "user", "content": prompt}]}), + ) + .await + .unwrap(); + resp.assert_ok(); + } + assert_eq!( + upstream.received_requests().await.unwrap_or_default().len(), + 4 + ); + + let ch = app.state.clickhouse.as_ref().expect("clickhouse client"); + for (status, detail) in gateway_rows(ch, user_id, 2).await { + assert_eq!(status, 200, "{detail}"); + assert_eq!(detail["retries"], 1, "{detail}"); + } +} diff --git a/crates/test-support/tests/gateway_retry.rs b/crates/test-support/tests/gateway_retry.rs new file mode 100644 index 00000000..06a421da --- /dev/null +++ b/crates/test-support/tests/gateway_retry.rs @@ -0,0 +1,566 @@ +//! Sending a request again after the upstream failed for the moment. +//! +//! Every candidate route tried and the last one overloaded, failing with a +//! server error or its connection broken, before any of the answer went +//! to the client: the gateway waits and sends again, within +//! `gateway.retry_wait_secs` of the request's arrival. A model with one +//! route — and a stream — goes back to the same route. Refusals are the +//! answer as before, and nothing is sent again once part of a stream went +//! out. +//! +//! The harness turns retries off (see `TestApp::try_spawn_with`); each +//! test here turns them on. Upstreams that say how long to wait +//! (`retry-after-ms`) keep the tests quick; one test waits out the +//! backoff itself. + +use serde_json::Value; +use std::time::{Duration, Instant}; +use think_watch_test_support::prelude::*; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +/// A model routed to `uri`, and a key that may call it. +async fn seed(app: &TestApp, uri: &str, provider_type: &str, model: &str) -> String { + let user = fixtures::create_random_user(&app.db).await.unwrap(); + let provider = + fixtures::create_provider(&app.db, &unique_name("retry"), provider_type, uri, None) + .await + .unwrap(); + fixtures::create_model_and_route(&app.db, provider.id, model) + .await + .unwrap(); + app.rebuild_gateway_router().await; + fixtures::create_api_key( + &app.db, + user.user.id, + &unique_name("retry-key"), + &["ai_gateway"], + None, + None, + ) + .await + .unwrap() + .plaintext +} + +async fn with_retries(secs: i64) -> TestApp { + let app = TestApp::spawn().await; + app.set_setting("gateway.retry_wait_secs", json!(secs)) + .await; + app +} + +/// An overloaded upstream that says to come back in 250 ms. +fn overloaded() -> ResponseTemplate { + ResponseTemplate::new(503) + .insert_header("retry-after-ms", "250") + .set_body_json(json!({"error": {"message": "overloaded"}})) +} + +fn chat_ok(model: &str) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_json(json!({ + "id": "chatcmpl-retry", + "object": "chat.completion", + "created": 1_700_000_000_i64, + "model": model, + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "hello world"}, + "finish_reason": "stop" + }], + "usage": {"prompt_tokens": 7, "completion_tokens": 3, "total_tokens": 10} + })) +} + +/// `first` for the first `times` requests, `then` after. +async fn upstream( + at: &str, + first: ResponseTemplate, + times: u64, + then: ResponseTemplate, +) -> MockServer { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path(at)) + .respond_with(first) + .up_to_n_times(times) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path(at)) + .respond_with(then) + .mount(&server) + .await; + server +} + +async fn sent(server: &MockServer) -> usize { + server.received_requests().await.unwrap_or_default().len() +} + +fn chat(model: &str, stream: bool) -> Value { + json!({"model": model, "stream": stream, "messages": [{"role": "user", "content": "hi"}]}) +} + +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_whole_answer_goes_out_again_after_an_overload() { + let app = with_retries(30).await; + let server = upstream("/v1/chat/completions", overloaded(), 2, chat_ok("retry-1")).await; + let key = seed(&app, &server.uri(), "openai", "retry-1").await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let resp = gw + .post("/v1/chat/completions", chat("retry-1", false)) + .await + .unwrap(); + resp.assert_ok(); + let body: Value = resp.json().unwrap(); + assert_eq!(body["choices"][0]["message"]["content"], "hello world"); + assert_eq!(sent(&server).await, 3, "two overloads, then the answer"); +} + +/// Without a word from the upstream, the first retry waits two seconds, +/// give or take a fifth. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn without_a_word_from_the_upstream_the_retry_waits_out_the_backoff() { + let app = with_retries(30).await; + let bad_gateway = ResponseTemplate::new(502).set_body_string("bad gateway"); + let server = upstream("/v1/chat/completions", bad_gateway, 1, chat_ok("retry-2")).await; + let key = seed(&app, &server.uri(), "openai", "retry-2").await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let started = Instant::now(); + gw.post("/v1/chat/completions", chat("retry-2", false)) + .await + .unwrap() + .assert_ok(); + let took = started.elapsed(); + assert!(took >= Duration::from_millis(1600), "{took:?}"); + assert_eq!(sent(&server).await, 2); +} + +/// An upstream that says to come back at once is not asked again at +/// once: a zero wait is no word at all, and the backoff applies. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_wait_of_zero_from_the_upstream_still_waits_out_the_backoff() { + let app = with_retries(30).await; + let now = ResponseTemplate::new(503) + .insert_header("retry-after-ms", "0") + .insert_header("retry-after", "0") + .set_body_json(json!({"error": {"message": "overloaded"}})); + let server = upstream("/v1/chat/completions", now, 1, chat_ok("retry-zero")).await; + let key = seed(&app, &server.uri(), "openai", "retry-zero").await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let started = Instant::now(); + gw.post("/v1/chat/completions", chat("retry-zero", false)) + .await + .unwrap() + .assert_ok(); + let took = started.elapsed(); + assert!(took >= Duration::from_millis(1600), "{took:?}"); + assert_eq!(sent(&server).await, 2); +} + +/// A refusal — a bad request, the upstream's rate limit — is the answer: +/// the upstream is asked once. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_refusal_is_not_sent_again() { + let app = with_retries(30).await; + for (status, model) in [ + (400u16, "retry-400"), + (429, "retry-429"), + (404, "retry-404"), + ] { + let refusal = ResponseTemplate::new(status) + .insert_header("retry-after-ms", "250") + .set_body_json(json!({"error": {"message": "no"}})); + let server = upstream("/v1/chat/completions", refusal, 1, chat_ok(model)).await; + let key = seed(&app, &server.uri(), "openai", model).await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let resp = gw + .post("/v1/chat/completions", chat(model, false)) + .await + .unwrap(); + resp.assert_status(status); + assert_eq!(sent(&server).await, 1, "{status}"); + } +} + +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_zero_wait_never_sends_again() { + let app = with_retries(0).await; + let server = upstream( + "/v1/chat/completions", + overloaded(), + 1, + chat_ok("retry-off"), + ) + .await; + let key = seed(&app, &server.uri(), "openai", "retry-off").await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let resp = gw + .post("/v1/chat/completions", chat("retry-off", false)) + .await + .unwrap(); + resp.assert_status(503); + assert_eq!(sent(&server).await, 1); +} + +/// The retries stop when the next wait would end past the deadline, and +/// the last failure is the answer. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn the_retries_stop_at_the_deadline() { + let app = with_retries(1).await; + let slow_down = || { + ResponseTemplate::new(503) + .insert_header("retry-after-ms", "400") + .set_body_json(json!({"error": {"message": "overloaded"}})) + }; + let server = upstream("/v1/chat/completions", slow_down(), 100, slow_down()).await; + let key = seed(&app, &server.uri(), "openai", "retry-deadline").await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let started = Instant::now(); + let resp = gw + .post("/v1/chat/completions", chat("retry-deadline", false)) + .await + .unwrap(); + resp.assert_status(503); + assert!(started.elapsed() < Duration::from_secs(3)); + // 0 ms, ~400 ms, ~800 ms; a wait to ~1200 ms ends past the deadline. + let n = sent(&server).await; + assert!((2..=3).contains(&n), "{n}"); +} + +/// Two routes failing for the moment: both are asked again, and the one +/// that has recovered answers. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn every_route_that_failed_for_the_moment_is_asked_again() { + let app = with_retries(30).await; + let down = upstream("/v1/chat/completions", overloaded(), 100, overloaded()).await; + let back = upstream( + "/v1/chat/completions", + overloaded(), + 1, + chat_ok("retry-two"), + ) + .await; + let user = fixtures::create_random_user(&app.db).await.unwrap(); + for server in [&down, &back] { + let p = fixtures::create_provider( + &app.db, + &unique_name("retry-two"), + "openai", + &server.uri(), + None, + ) + .await + .unwrap(); + fixtures::create_model_route(&app.db, p.id, "retry-two", 100) + .await + .unwrap(); + } + app.rebuild_gateway_router().await; + let key = fixtures::create_api_key( + &app.db, + user.user.id, + &unique_name("retry-two-key"), + &["ai_gateway"], + None, + None, + ) + .await + .unwrap(); + + let gw = app.gateway_client(); + gw.set_bearer(&key.plaintext); + gw.post("/v1/chat/completions", chat("retry-two", false)) + .await + .unwrap() + .assert_ok(); + assert_eq!(sent(&back).await, 2, "failed once, answered the retry"); + assert!(sent(&down).await >= 1); +} + +/// A stream whose upstream cannot open goes out again: the client gets +/// one whole answer. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_stream_that_cannot_open_goes_out_again() { + let app = with_retries(30).await; + let chunk = |delta: Value, finish: Value| { + format!( + "data: {}\n\n", + json!({"id": "c", "object": "chat.completion.chunk", "created": 1, + "model": "retry-stream", + "choices": [{"index": 0, "delta": delta, "finish_reason": finish}]}) + ) + }; + let answer = [ + chunk(json!({"role": "assistant"}), Value::Null), + chunk(json!({"content": "hello"}), Value::Null), + chunk(json!({}), json!("stop")), + "data: [DONE]\n\n".to_string(), + ] + .concat(); + let ok = ResponseTemplate::new(200).set_body_raw(answer, "text/event-stream"); + let server = upstream("/v1/chat/completions", overloaded(), 1, ok).await; + let key = seed(&app, &server.uri(), "openai", "retry-stream").await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let resp = gw + .post("/v1/chat/completions", chat("retry-stream", true)) + .await + .unwrap(); + resp.assert_ok(); + let text = resp.text(); + assert!(text.contains("hello"), "{text}"); + assert!(!text.contains("overloaded"), "{text}"); + assert_eq!(text.matches("\"role\":\"assistant\"").count(), 1, "{text}"); + assert_eq!(sent(&server).await, 2); +} + +fn anthropic_event(kind: &str, v: Value) -> String { + format!("event: {kind}\ndata: {v}\n\n") +} + +fn message_start(model: &str) -> String { + anthropic_event( + "message_start", + json!({"type": "message_start", "message": { + "id": "msg_1", "type": "message", "role": "assistant", "model": model, + "content": [], "stop_reason": null, + "usage": {"input_tokens": 12, "output_tokens": 1} + }}), + ) +} + +fn overload_event() -> String { + anthropic_event( + "error", + json!({"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}), + ) +} + +fn text_block(text: &str) -> String { + [ + anthropic_event( + "content_block_start", + json!({"type": "content_block_start", "index": 0, + "content_block": {"type": "text", "text": ""}}), + ), + anthropic_event( + "content_block_delta", + json!({"type": "content_block_delta", "index": 0, + "delta": {"type": "text_delta", "text": text}}), + ), + ] + .concat() +} + +fn messages(model: &str) -> Value { + json!({"model": model, "max_tokens": 100, "stream": true, + "messages": [{"role": "user", "content": "hi"}]}) +} + +/// Anthropic opens its stream and then reports an overload, before any +/// content: the opening was held back, the stream goes out again, and +/// the client sees one answer. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_stream_overloaded_before_any_content_goes_out_again() { + let app = with_retries(30).await; + let model = "retry-claude"; + let failed = [message_start(model), overload_event()].concat(); + let answer = [ + message_start(model), + text_block("Hello"), + anthropic_event( + "content_block_stop", + json!({"type": "content_block_stop", "index": 0}), + ), + anthropic_event( + "message_delta", + json!({"type": "message_delta", "delta": {"stop_reason": "end_turn"}, + "usage": {"output_tokens": 2}}), + ), + anthropic_event("message_stop", json!({"type": "message_stop"})), + ] + .concat(); + let server = upstream( + "/v1/messages", + ResponseTemplate::new(200).set_body_raw(failed, "text/event-stream"), + 1, + ResponseTemplate::new(200).set_body_raw(answer, "text/event-stream"), + ) + .await; + let key = seed(&app, &server.uri(), "anthropic", model).await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let resp = gw.post("/v1/messages", messages(model)).await.unwrap(); + resp.assert_ok(); + let text = resp.text(); + assert_eq!(text.matches("event: message_start").count(), 1, "{text}"); + assert!(text.contains("Hello"), "{text}"); + assert!(text.contains("event: message_stop"), "{text}"); + assert!(!text.contains("overloaded_error"), "{text}"); + assert_eq!(sent(&server).await, 2); +} + +/// Part of the answer went out before the overload: the client gets it +/// and the error, and the upstream is not asked again. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_stream_that_failed_after_content_is_not_sent_again() { + let app = with_retries(30).await; + let model = "retry-claude-late"; + let failed = [message_start(model), text_block("Hel"), overload_event()].concat(); + let server = upstream( + "/v1/messages", + ResponseTemplate::new(200).set_body_raw(failed.clone(), "text/event-stream"), + 1, + ResponseTemplate::new(200).set_body_raw(failed, "text/event-stream"), + ) + .await; + let key = seed(&app, &server.uri(), "anthropic", model).await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let resp = gw.post("/v1/messages", messages(model)).await.unwrap(); + resp.assert_ok(); + let text = resp.text(); + assert!(text.contains("Hel"), "{text}"); + assert!(text.contains("overloaded_error"), "{text}"); + assert_eq!(sent(&server).await, 1); +} + +/// A stream the upstream refuses before any content — a bad request — +/// is the answer: the opening and the refusal reach the client, once. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_stream_refused_before_any_content_is_not_sent_again() { + let app = with_retries(30).await; + let model = "retry-claude-refused"; + let refused = [ + message_start(model), + anthropic_event( + "error", + json!({"type": "error", "error": {"type": "invalid_request_error", + "message": "prompt is too long"}}), + ), + ] + .concat(); + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/messages")) + .respond_with(ResponseTemplate::new(200).set_body_raw(refused, "text/event-stream")) + .mount(&server) + .await; + let key = seed(&app, &server.uri(), "anthropic", model).await; + + let gw = app.gateway_client(); + gw.set_bearer(&key); + let resp = gw.post("/v1/messages", messages(model)).await.unwrap(); + resp.assert_ok(); + let text = resp.text(); + assert_eq!(text.matches("event: message_start").count(), 1, "{text}"); + assert!(text.contains("prompt is too long"), "{text}"); + assert_eq!(sent(&server).await, 1); +} + +/// While a stream waits to go out again, a Chat client hears a keep-alive +/// comment after 15 seconds of silence; a Gemini client reading SSE does +/// not — the google-genai SDK reads every line as JSON. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn only_non_gemini_clients_get_keep_alives_while_a_stream_waits() { + let app = with_retries(60).await; + let chunk = |model: &str, delta: Value, finish: Value| { + format!( + "data: {}\n\n", + json!({"id": "c", "object": "chat.completion.chunk", "created": 1, "model": model, + "choices": [{"index": 0, "delta": delta, "finish_reason": finish}]}) + ) + }; + let stream_of = |model: &str| { + [ + chunk(model, json!({"content": "hello"}), Value::Null), + chunk(model, json!({}), json!("stop")), + "data: [DONE]\n\n".to_string(), + ] + .concat() + }; + // Overloaded once, saying to come back in 16 s: longer than the + // keep-alive interval. + let busy = || { + ResponseTemplate::new(503) + .insert_header("retry-after-ms", "16000") + .set_body_json(json!({"error": {"message": "overloaded"}})) + }; + let mut keys = Vec::new(); + for model in ["retry-ka-chat", "retry-ka-gemini"] { + let ok = ResponseTemplate::new(200).set_body_raw(stream_of(model), "text/event-stream"); + let server = upstream("/v1/chat/completions", busy(), 1, ok).await; + keys.push((seed(&app, &server.uri(), "openai", model).await, server)); + } + + let chat = async { + reqwest::Client::new() + .post(format!("{}/v1/chat/completions", app.gateway_url)) + .bearer_auth(&keys[0].0) + .json(&chat("retry-ka-chat", true)) + .send() + .await + .unwrap() + .text() + .await + .unwrap() + }; + let gemini = async { + reqwest::Client::new() + .post(format!( + "{}/v1beta/models/retry-ka-gemini:streamGenerateContent?alt=sse", + app.gateway_url + )) + .header("x-goog-api-key", &keys[1].0) + .json(&json!({"contents": [{"role": "user", "parts": [{"text": "hi"}]}]})) + .send() + .await + .unwrap() + .text() + .await + .unwrap() + }; + let (chat, gemini) = tokio::join!(chat, gemini); + + assert!(chat.contains(": keep-alive\n\n"), "{chat}"); + assert!(chat.contains("hello"), "{chat}"); + assert!(gemini.contains("hello"), "{gemini}"); + for line in gemini.lines().filter(|l| !l.is_empty()) { + let data = line + .strip_prefix("data: ") + .unwrap_or_else(|| panic!("not a data line: {line:?}\n{gemini}")); + serde_json::from_str::(data).unwrap_or_else(|e| panic!("{e}: {data}")); + } + for (_, server) in &keys { + assert_eq!(sent(server).await, 2); + } +} diff --git a/db/seeds.sql b/db/seeds.sql index 596ef90b..d670ba24 100644 --- a/db/seeds.sql +++ b/db/seeds.sql @@ -354,6 +354,8 @@ INSERT INTO system_settings (key, value, category, description) VALUES 'Default session affinity mode (none/provider/route)'), ('gateway.default_affinity_ttl_secs','300', 'gateway', 'Default affinity key TTL in seconds (0-86400)'), + ('gateway.retry_wait_secs', '120', 'gateway', + 'How long after a request arrives it may still be sent again to upstreams that failed for the moment (overloaded, server error, broken connection) before any of the answer reached the client, in seconds (0-600; 0 = never)'), ('gateway.latency_strategy_k', '2.0', 'gateway', 'Exponent for the latency-strategy weighting (higher = more aggressive). Default 2.0 (aggressive).'), ('gateway.cb_enabled', 'true', 'gateway', diff --git a/web/src/i18n/en.json b/web/src/i18n/en.json index b49de707..7006d45a 100644 --- a/web/src/i18n/en.json +++ b/web/src/i18n/en.json @@ -1816,6 +1816,8 @@ "defaultAffinityTtlHint": "How long an affinity assignment lasts. 0 disables affinity entirely.", "latencyStrategyK": "Latency aggressiveness (k)", "latencyStrategyKHint": "Exponent in 1/latency^k. Higher = more aggressive bias toward fast routes. Default 2.0 (aggressive). Range 0.5–8.0.", + "retryWaitSecs": "Retry window (seconds)", + "retryWaitSecsHint": "When every route failed for the moment (overloaded, a server error, a broken connection) before any of the answer reached the client, the request is sent again after a short wait, for up to this long after it arrived. 0 disables retries. Range 0–600.", "strategy": { "weighted": "Weighted random", "latency": "Latency-first", diff --git a/web/src/i18n/zh.json b/web/src/i18n/zh.json index 272d5000..6f9af408 100644 --- a/web/src/i18n/zh.json +++ b/web/src/i18n/zh.json @@ -1816,6 +1816,8 @@ "defaultAffinityTtlHint": "黏性键的存活时间。0 = 完全关闭黏性。", "latencyStrategyK": "延迟激进度 (k)", "latencyStrategyKHint": "权重 ∝ 1/延迟^k。k 越大对慢路由的惩罚越激进。默认 2.0(激进)。范围 0.5–8.0。", + "retryWaitSecs": "重试时限 (秒)", + "retryWaitSecsHint": "所有路由都暂时失败（过载、服务器错误、连接中断）且回答尚未有任何内容发给客户端时，稍等后重发请求，直到请求到达后的这段时间为止。0 = 不重试。范围 0–600。", "strategy": { "weighted": "加权随机", "latency": "延迟优先", diff --git a/web/src/routes/admin/settings.tsx b/web/src/routes/admin/settings.tsx index d5b63fc1..a7e7f614 100644 --- a/web/src/routes/admin/settings.tsx +++ b/web/src/routes/admin/settings.tsx @@ -105,6 +105,7 @@ export function SettingsPage() { const [defaultAffinityMode, setDefaultAffinityMode] = useState('provider'); const [defaultAffinityTtlSecs, setDefaultAffinityTtlSecs] = useState(300); const [latencyStrategyK, setLatencyStrategyK] = useState(2.0); + const [retryWaitSecs, setRetryWaitSecs] = useState(120); const [cbEnabled, setCbEnabled] = useState(true); const [cbErrorPct, setCbErrorPct] = useState(50); const [cbMinSamples, setCbMinSamples] = useState(10); @@ -162,6 +163,7 @@ export function SettingsPage() { setDefaultAffinityMode(str(getSettingValue(data, 'gateway', 'default_affinity_mode'), 'provider')); setDefaultAffinityTtlSecs(num(getSettingValue(data, 'gateway', 'default_affinity_ttl_secs'), 300)); setLatencyStrategyK(Number(getSettingValue(data, 'gateway', 'latency_strategy_k')) || 2.0); + setRetryWaitSecs(num(getSettingValue(data, 'gateway', 'retry_wait_secs'), 120)); setCbEnabled(getSettingValue(data, 'gateway', 'cb_enabled') !== false); setCbErrorPct(num(getSettingValue(data, 'gateway', 'cb_error_pct'), 50)); setCbMinSamples(num(getSettingValue(data, 'gateway', 'cb_min_samples'), 10)); @@ -258,6 +260,7 @@ export function SettingsPage() { const defaultAffinityModeSave = useFieldAutosave({ value: defaultAffinityMode, isLoaded, persist: (v) => patchOne('gateway.default_affinity_mode', v), debounceMs: 0 }); const defaultAffinityTtlSave = useFieldAutosave({ value: defaultAffinityTtlSecs, isLoaded, persist: (v) => patchOne('gateway.default_affinity_ttl_secs', v) }); const latencyStrategyKSave = useFieldAutosave({ value: latencyStrategyK, isLoaded, persist: (v) => patchOne('gateway.latency_strategy_k', v) }); + const retryWaitSecsSave = useFieldAutosave({ value: retryWaitSecs, isLoaded, persist: (v) => patchOne('gateway.retry_wait_secs', v) }); const cbEnabledSave = useFieldAutosave({ value: cbEnabled, isLoaded, persist: (v) => patchOne('gateway.cb_enabled', v), debounceMs: 0 }); const cbErrorPctSave = useFieldAutosave({ value: cbErrorPct, isLoaded, persist: (v) => patchOne('gateway.cb_error_pct', v) }); const cbMinSamplesSave = useFieldAutosave({ value: cbMinSamples, isLoaded, persist: (v) => patchOne('gateway.cb_min_samples', v) }); @@ -662,6 +665,15 @@ export function SettingsPage() { />

{t('settings.latencyStrategyKHint')}

+ } + />