diff --git a/crates/libsy-llm-client/Cargo.toml b/crates/libsy-llm-client/Cargo.toml index 9a9ca6e0a..cb8ea4f93 100644 --- a/crates/libsy-llm-client/Cargo.toml +++ b/crates/libsy-llm-client/Cargo.toml @@ -34,6 +34,7 @@ tracing.workspace = true tracing-opentelemetry.workspace = true [dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } # SDK + in-memory exporter to assert what the observability layer records. opentelemetry_sdk = { version = "0.32", features = ["metrics", "testing", "trace"] } thiserror.workspace = true diff --git a/crates/libsy-llm-client/README.md b/crates/libsy-llm-client/README.md index 8b0ef430d..c81f749dd 100644 --- a/crates/libsy-llm-client/README.md +++ b/crates/libsy-llm-client/README.md @@ -73,6 +73,7 @@ fn build_client() -> switchyard_llm_client::Result { extra_body: BTreeMap::new(), reasoning_effort: None, max_retries: 2, + failure_cooldown: std::time::Duration::from_secs(5), timeout: None, }; @@ -258,6 +259,12 @@ fn build_multi_format_client( transport failures, timeouts, HTTP 408/429, and 5xx responses. Buffered body transport failures are retried; streaming body failures are not replayed after the response has been returned. +- `HttpBackendConfig::failure_cooldown` skips completion calls to a backend briefly + after an exhausted transient failure. Zero disables it. State is shared per model + within the client. Calls resume when the cooldown expires. Deployment TOML defaults + `failure_cooldown_ms` to `5000` (5 seconds); set it to `0` to disable cooldown. + With `forward_auth`, HTTP 429 keeps its request-local retries and fallback while + leaving shared cooldown state unchanged. - `HttpBackendConfig::timeout` bounds one complete response, including retries, retry delays, and every stream read. Expiry returns `LlmClientError::Timeout`, either from the call or from the returned stream, which then ends. `None` leaves diff --git a/crates/libsy-llm-client/src/backend.rs b/crates/libsy-llm-client/src/backend.rs index 064b13546..bbe23b0d1 100644 --- a/crates/libsy-llm-client/src/backend.rs +++ b/crates/libsy-llm-client/src/backend.rs @@ -72,6 +72,8 @@ pub struct HttpBackendConfig { pub reasoning_effort: Option, /// Additional attempts after the initial upstream request. pub max_retries: u32, + /// Cooldown after an exhausted transient completion failure. Zero disables it. + pub failure_cooldown: Duration, /// Deadline for one complete response, including retries, retry delays, and stream reads. /// `None` leaves the wait unbounded. pub timeout: Option, @@ -88,6 +90,7 @@ impl fmt::Debug for HttpBackendConfig { .field("omit_body_fields", &self.omit_body_fields) .field("reasoning_effort", &self.reasoning_effort) .field("max_retries", &self.max_retries) + .field("failure_cooldown", &self.failure_cooldown) .field("timeout", &self.timeout) .finish() } @@ -319,6 +322,11 @@ impl Backend { self.config().max_retries } + /// Cooldown after an exhausted transient completion failure. + pub fn failure_cooldown(&self) -> Duration { + self.config().failure_cooldown + } + /// Deadline for all attempts and the complete response; `None` leaves the wait unbounded. pub fn timeout(&self) -> Option { self.config().timeout @@ -431,6 +439,7 @@ mod tests { omit_body_fields: BTreeSet::new(), reasoning_effort: None, max_retries: 0, + failure_cooldown: Duration::ZERO, timeout: None, } } diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index f2d0301dd..6ba089d9c 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -6,6 +6,7 @@ use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::future::ready; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime}; use async_trait::async_trait; @@ -22,6 +23,7 @@ use switchyard_translation::{ TranslationError, WireFormat, decode_aggregated_response, decode_request, decode_stream, encode_aggregated_response_with_extensions, encode_request, encode_stream_with_extensions, }; +use tokio::time::Instant; use tracing::Instrument; use crate::backend::{Backend, openai_url}; @@ -126,19 +128,24 @@ pub struct TranslatingLlmClient { model_to_config: HashMap, client: reqwest::Client, forward_auth_client: reqwest::Client, + cooldown_epoch: Instant, + // Milliseconds since cooldown_epoch. Each atomic is independent, so relaxed ordering suffices. + unavailable_until: HashMap, } impl TranslatingLlmClient { /// Builds a client over the given [`ModelConfig`]s, with a fresh shared HTTP /// client and the built-in translation codecs. pub fn new(model_configs: &[ModelConfig]) -> Result { + // Pre-populate so concurrent calls update only atomics and keep map access lock-free. + let mut unavailable_until = HashMap::new(); for config in model_configs { - config - .default_backend - .validate_configured_headers(&config.model_name)?; - for backend in config.other_backends.iter().flatten() { + for backend in std::iter::once(&config.default_backend) + .chain(config.other_backends.iter().flatten()) + { backend.validate_configured_headers(&config.model_name)?; } + unavailable_until.insert(config.model_name.clone(), AtomicU64::new(0)); } let build_client = |builder: reqwest::ClientBuilder| { builder.build().map_err(|error| LlmClientError::Transport { @@ -159,6 +166,8 @@ impl TranslatingLlmClient { model_to_config, client, forward_auth_client, + cooldown_epoch: Instant::now(), + unavailable_until, }) } @@ -245,6 +254,20 @@ impl TranslatingLlmClient { model: &ModelId, endpoint: UpstreamEndpoint, ) -> Result { + let cooldown = backend.failure_cooldown(); + let unavailable_until = + if matches!(endpoint, UpstreamEndpoint::Completion) && !cooldown.is_zero() { + self.unavailable_until.get(model) + } else { + None + }; + if let Some(until) = unavailable_until + && until.load(Ordering::Relaxed) > duration_millis(self.cooldown_epoch.elapsed()) + { + tracing::debug!(model = %model, wire_format = %wire_format, "skipping backend during failure cooldown"); + return Err(LlmClientError::TemporarilyUnavailable); + } + let mut body = encode_request(&llm_request, wire_format) .map_err(|error| LlmClientError::RequestEncoding(error.to_string()))?; // `encode_request` round-trips a preserved same-format body verbatim, @@ -276,8 +299,21 @@ impl TranslatingLlmClient { let url = endpoint.url(backend); record_gen_ai_request(&url, model, streaming); - self.send_with_retries(&url, backend, &body, metadata, model, streaming) - .await + let result = self + .send_with_retries(&url, backend, &body, metadata, model, streaming) + .await; + // Forwarded credentials can hit a user's quota while the backend remains healthy. + if let Some(until) = unavailable_until + && let Err(error) = &result + && is_transient_failure(error) + && !(backend.is_forwarding_auth() + && matches!(error, LlmClientError::UpstreamHttp { status, .. } if *status == StatusCode::TOO_MANY_REQUESTS)) + { + let deadline = duration_millis(self.cooldown_epoch.elapsed().saturating_add(cooldown)); + until.fetch_max(deadline, Ordering::Relaxed); + } + // Let cooldown expire naturally; a successful in-flight call may overlap a newer failure. + result } // Sends the encoded body, retrying retryable failures within the backend's retry budget. @@ -733,13 +769,17 @@ fn deadline_error(timeout: Duration) -> LlmClientError { impl AttemptFailure { fn is_retryable(&self) -> bool { - match &self.error { - LlmClientError::Transport { .. } | LlmClientError::Timeout { .. } => true, - LlmClientError::UpstreamHttp { status, .. } => { - metrics::is_retryable_http_status(status.as_u16()) - } - _ => false, + is_transient_failure(&self.error) + } +} + +fn is_transient_failure(error: &LlmClientError) -> bool { + match error { + LlmClientError::Transport { .. } | LlmClientError::Timeout { .. } => true, + LlmClientError::UpstreamHttp { status, .. } => { + metrics::is_retryable_http_status(status.as_u16()) } + _ => false, } } @@ -1307,6 +1347,7 @@ mod tests { omit_body_fields: BTreeSet::new(), reasoning_effort: None, max_retries: 0, + failure_cooldown: Duration::ZERO, timeout: None, } } @@ -2399,6 +2440,51 @@ mod tests { Ok(()) } + #[tokio::test] + async fn rate_limit_cooldown_depends_on_shared_credentials() + -> std::result::Result<(), Box> { + for forward_auth in [true, false] { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(429).insert_header("retry-after", "0")) + .expect(if forward_auth { 6 } else { 3 }) + .mount(&server) + .await; + let backend = HttpBackendConfig { + forward_auth, + failure_cooldown: Duration::from_secs(60), + ..config_with_retries(&format!("{}/v1", server.uri()), 2) + }; + let client = TranslatingLlmClient::new(&[ModelConfig::new( + "gpt", + Backend::OpenAiChat(backend), + None, + )])?; + for credential in ["Bearer first-user", "Bearer second-user"] { + let mut headers = http::HeaderMap::new(); + headers.insert("authorization", http::HeaderValue::from_static(credential)); + let result = client + .call_rewrite_model(request_with_headers("gpt", headers), None) + .await; + if !forward_auth && credential == "Bearer second-user" { + assert!(matches!( + result, + Err(LlmClientError::TemporarilyUnavailable) + )); + } else { + assert!(matches!( + result, + Err(LlmClientError::UpstreamHttp { + status: StatusCode::TOO_MANY_REQUESTS, + .. + }) + )); + } + } + } + Ok(()) + } + #[tokio::test] async fn retry_exhaustion_returns_the_final_upstream_error() -> std::result::Result<(), Box> { diff --git a/crates/libsy-llm-client/src/observability.rs b/crates/libsy-llm-client/src/observability.rs index c55d46ad8..03a32951c 100644 --- a/crates/libsy-llm-client/src/observability.rs +++ b/crates/libsy-llm-client/src/observability.rs @@ -201,6 +201,7 @@ fn llm_client_error_type(error: &LlmClientError) -> Cow<'static, str> { LlmClientError::ResponseTranslation(_) => Cow::Borrowed("response_translation"), LlmClientError::Configuration { .. } => Cow::Borrowed("configuration"), LlmClientError::Transport { .. } => Cow::Borrowed("transport"), + LlmClientError::TemporarilyUnavailable => Cow::Borrowed("temporarily_unavailable"), LlmClientError::Timeout { .. } => Cow::Borrowed("timeout"), LlmClientError::ContextWindowExceeded { .. } => Cow::Borrowed("context_window_exceeded"), LlmClientError::UpstreamHttp { status, .. } => Cow::Owned(status.as_str().to_owned()), diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 691e169cf..e7effb469 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -371,7 +371,9 @@ fn fallback_reason(error: &LibsyError) -> Option { }; match source { LlmClientError::ContextWindowExceeded { .. } => Some(RoutingFallbackReason::ContextWindow), - LlmClientError::Transport { .. } => Some(RoutingFallbackReason::Unavailable), + LlmClientError::Transport { .. } | LlmClientError::TemporarilyUnavailable => { + Some(RoutingFallbackReason::Unavailable) + } // A policy denial can be specific to one provider. Preserve its HTTP // error, but allow another candidate to serve the request. LlmClientError::UpstreamHttp { status, body } @@ -1213,6 +1215,7 @@ mod tests { omit_body_fields: BTreeSet::new(), reasoning_effort: None, max_retries: 0, + failure_cooldown: std::time::Duration::ZERO, timeout: None, }; let backend = if responses { @@ -1359,6 +1362,7 @@ mod tests { omit_body_fields: BTreeSet::new(), reasoning_effort: None, max_retries: 0, + failure_cooldown: std::time::Duration::ZERO, timeout: None, }), None, @@ -1991,7 +1995,8 @@ mod tests { } #[tokio::test] - async fn retry_budget_is_exhausted_before_falling_through() -> Result<()> { + async fn retry_exhaustion_falls_back_and_optional_cooldown_skips_failed_backend() -> Result<()> + { let server = MockServer::start().await; let calls = Arc::new(Mutex::new(Vec::new())); let observed_calls = Arc::clone(&calls); @@ -2021,37 +2026,60 @@ mod tests { .mount(&server) .await; - let backend = || { - Backend::OpenAiChat(HttpBackendConfig { - base_url: format!("{}/v1", server.uri()), - api_key: None, - forward_auth: false, - extra_headers: BTreeMap::new(), - extra_body: BTreeMap::new(), - omit_body_fields: BTreeSet::new(), - reasoning_effort: None, - max_retries: 2, - timeout: None, - }) - }; - let client = Arc::new( + let client_with_cooldown = |cooldown| { + let backend = || { + Backend::OpenAiChat(HttpBackendConfig { + base_url: format!("{}/v1", server.uri()), + api_key: None, + forward_auth: false, + extra_headers: BTreeMap::new(), + extra_body: BTreeMap::new(), + omit_body_fields: BTreeSet::new(), + reasoning_effort: None, + max_retries: 2, + failure_cooldown: cooldown, + timeout: None, + }) + }; TranslatingLlmClient::new(&[ ModelConfig::new("weak", backend(), None), ModelConfig::new("strong", backend(), None), ]) - .map_err(|error| LibsyError::external("building test client", error))?, + .map(Arc::new) + .map_err(|error| LibsyError::external("building test client", error)) + }; + let call = async |client: Arc| { + run( + Arc::new(CandidateAlgorithm {}), + ClientRouter::single(client), + request(), + to_category_map(&["weak", "strong"]), + None, + ) + .await?; + Ok::<_, LibsyError>(std::mem::take(&mut *calls.lock())) + }; + + let client = client_with_cooldown(std::time::Duration::ZERO)?; + assert_eq!( + call(client.clone()).await?, + ["weak", "weak", "weak", "strong"] ); - let algorithm = Arc::new(CandidateAlgorithm {}); - run( - algorithm, - ClientRouter::single(client), - request(), - to_category_map(&["weak", "strong"]), - None, - ) - .await?; + assert_eq!(call(client).await?, ["weak", "weak", "weak", "strong"]); + + let cooldown = std::time::Duration::from_secs(60); + let client = client_with_cooldown(cooldown)?; + assert_eq!( + call(client.clone()).await?, + ["weak", "weak", "weak", "strong"] + ); + assert_eq!(call(client.clone()).await?, ["strong"]); + + tokio::time::pause(); + tokio::time::advance(cooldown).await; + tokio::time::resume(); - assert_eq!(&*calls.lock(), &["weak", "weak", "weak", "strong"]); + assert_eq!(call(client).await?, ["weak", "weak", "weak", "strong"]); Ok(()) } @@ -2130,6 +2158,7 @@ mod tests { omit_body_fields: BTreeSet::new(), reasoning_effort: None, max_retries: 0, + failure_cooldown: std::time::Duration::ZERO, timeout: None, }) }; diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index f8626d2bd..8d312e72d 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -63,6 +63,10 @@ pub enum LlmClientError { #[error("Responses state ID is already recorded for another model; its owner was not changed")] ResponseStateConflict, + /// The selected backend is temporarily unavailable. Another candidate may serve the request. + #[error("backend temporarily unavailable")] + TemporarilyUnavailable, + /// The upstream could not be reached or the request could not be sent. #[error("upstream transport error: {source}")] Transport { diff --git a/crates/switchyard-runner/src/config.rs b/crates/switchyard-runner/src/config.rs index f37b4cda9..1c976a47b 100644 --- a/crates/switchyard-runner/src/config.rs +++ b/crates/switchyard-runner/src/config.rs @@ -576,6 +576,9 @@ struct LlmClientConfig { extra_headers: BTreeMap, #[serde(default = "default_max_retries")] max_retries: u32, + /// Cooldown after an exhausted transient completion failure. Defaults to 5 seconds; zero disables it. + #[serde(default = "default_failure_cooldown_ms")] + failure_cooldown_ms: u64, /// Deadline in milliseconds for all attempts and the complete response. Unset is unbounded. timeout_ms: Option, } @@ -700,6 +703,7 @@ fn build_backend( omit_body_fields: omit_body_fields.clone(), reasoning_effort, max_retries: config.max_retries, + failure_cooldown: Duration::from_millis(config.failure_cooldown_ms), timeout: config.timeout_ms.map(Duration::from_millis), }; let backend = match config.format { @@ -715,6 +719,10 @@ const fn default_max_retries() -> u32 { DEFAULT_MAX_RETRIES } +const fn default_failure_cooldown_ms() -> u64 { + 5000 +} + fn validate_value(label: &str, value: &str) -> RunnerResult<()> { if value.trim().is_empty() || value.trim() != value { return Err(RunnerError::configuration(format!( diff --git a/crates/switchyard-runner/src/failure.rs b/crates/switchyard-runner/src/failure.rs index 8113ef25b..e73b05a29 100644 --- a/crates/switchyard-runner/src/failure.rs +++ b/crates/switchyard-runner/src/failure.rs @@ -21,6 +21,8 @@ pub enum RouteErrorKind { UpstreamHttp, /// The selected target rejected the request because its context window was exceeded. ContextWindowExceeded, + /// The selected backend is temporarily unavailable. + TemporarilyUnavailable, /// The upstream request timed out. Timeout, /// The upstream could not be reached or the request could not be sent. @@ -144,6 +146,7 @@ fn client_error_summary( LlmClientError::ContextWindowExceeded { .. } => { (RouteErrorKind::ContextWindowExceeded, None) } + LlmClientError::TemporarilyUnavailable => (RouteErrorKind::TemporarilyUnavailable, None), LlmClientError::Timeout { .. } => (RouteErrorKind::Timeout, None), LlmClientError::Transport { .. } => (RouteErrorKind::Transport, None), LlmClientError::InvalidResponse { .. } => (RouteErrorKind::InvalidResponse, None), diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index fe5bdef83..0f1738936 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -1307,6 +1307,12 @@ fn client_error(error: &LlmClientError) -> Response { "invalid_request_error", "context_length_exceeded", ), + LlmClientError::TemporarilyUnavailable => error_response( + StatusCode::SERVICE_UNAVAILABLE, + error.to_string(), + "upstream_error", + "temporarily_unavailable", + ), LlmClientError::UpstreamHttp { status, body } => upstream_error(*status, body), LlmClientError::Transport { source } | LlmClientError::InvalidResponse { source } => { error_response( diff --git a/crates/switchyard-server/tests/client_deadline.rs b/crates/switchyard-server/tests/client_deadline.rs index 58bd06fc3..896f7be0f 100644 --- a/crates/switchyard-server/tests/client_deadline.rs +++ b/crates/switchyard-server/tests/client_deadline.rs @@ -148,6 +148,8 @@ schema_version = 1 format = "openai_chat" base_url = "{}" max_retries = 1 +# Exercise the upstream deadline on every request. +failure_cooldown_ms = 0 timeout_ms = {timeout} [targets] judge = {{ id = "{judge}", llm_client = "http", extra_body = {{ stream = {judge_stream} }} }} diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 3c2656257..992a1bbd0 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -758,6 +758,7 @@ fn random_state_with_retries( omit_body_fields: BTreeSet::new(), reasoning_effort: None, max_retries, + failure_cooldown: std::time::Duration::ZERO, timeout: None, }); let target_models = routes @@ -1358,6 +1359,8 @@ schema_version = 1 format = "openai_chat" base_url = "{base_url}" max_retries = 0 +# Exercise upstream fallback on every request. +failure_cooldown_ms = 0 [targets.first] id = "{first}" diff --git a/docs/reference/toml_schema.md b/docs/reference/toml_schema.md index 4b51a7918..5c112ee35 100644 --- a/docs/reference/toml_schema.md +++ b/docs/reference/toml_schema.md @@ -54,11 +54,24 @@ route reaches no upstream. A file without a `[targets]` table is rejected with | `forward_auth` | No | `false` | Forward the caller's provider credential and application headers. All backends reachable through the route must use the same provider. | | `extra_headers` | No | `{}` | Custom HTTP headers sent to the model server. Set credentials with `api_key_env` or `forward_auth`; the server rejects headers owned by the selected auth mode. Header names are case-insensitive. | | `max_retries` | No | `2` | Retry budget, `0`–`10`. | +| `failure_cooldown_ms` | No | `5000` (5 seconds) | Skip a backend for this many milliseconds after an exhausted transient completion failure. Zero disables it. | | `timeout_ms` | No | unset | Deadline in milliseconds for all attempts, retry delays, and the complete response, including stream reads. Must be at least `1`. Unset leaves the wait unbounded. | The TOML never contains the secret itself. `api_key_env` names a variable that must exist and be non-empty when the server loads. +Cooldown is enabled by default for 5 seconds. Set `failure_cooldown_ms = 0` to disable it. +The cooldown starts after retries are exhausted. + +`failure_cooldown_ms` tracks transport failures, timeouts, HTTP 408/429, and 5xx +responses after retries. State is shared across callers per model within the client, +including callers using forwarded credentials. With `forward_auth = true`, HTTP 429 +only triggers request-local retries and fallback; other callers keep trying the model. +For shared credentials, HTTP 429 also triggers cooldown. During cooldown, +ordered fallback tries the next candidate. A terminal cooldown error returns HTTP +503. Calls resume together after expiry. Auxiliary calls and errors after a stream +is returned leave cooldown state unchanged. + `timeout_ms` applies separately to every call through the client, including judge verdicts and answers. To give a judge a short deadline without limiting the answering models, put the judge on its own `[llm_clients]` entry; two entries may