diff --git a/crates/protocols/src/worker.rs b/crates/protocols/src/worker.rs index 6e84d64540..af0301d9c1 100644 --- a/crates/protocols/src/worker.rs +++ b/crates/protocols/src/worker.rs @@ -1062,9 +1062,17 @@ pub struct ResilienceUpdate { pub disable_circuit_breaker: Option, // ── Retryable status codes ── - /// Custom retryable HTTP status codes. - /// When set, replaces the default set (408, 429, 500, 502, 503, 504). + /// HTTP status codes this worker counts as circuit-breaker failures. + /// When set, replaces the default set (408, 429, 500, 502, 503, 504) + /// verbatim - entries are not merged in. This does not gate retries: + /// whether a response is retried is a router-global rule, independent of + /// this set, so narrowing it cannot make a status non-retryable. pub retryable_status_codes: Option>, + /// Capacity-pushback HTTP status codes: still retryable on another + /// worker, but never counted as circuit-breaker failures (backpressure + /// is a routing signal, not a fault). When set, replaces the default + /// set (429). + pub capacity_status_codes: Option>, } impl ResilienceUpdate { @@ -1082,6 +1090,7 @@ impl ResilienceUpdate { && self.cb_window_secs.is_none() && self.disable_circuit_breaker.is_none() && self.retryable_status_codes.is_none() + && self.capacity_status_codes.is_none() } } diff --git a/model_gateway/src/worker/resilience.rs b/model_gateway/src/worker/resilience.rs index a94329b8c0..71fb473fc4 100644 --- a/model_gateway/src/worker/resilience.rs +++ b/model_gateway/src/worker/resilience.rs @@ -12,6 +12,12 @@ use crate::{config::types::RetryConfig, worker::circuit_breaker::CircuitBreakerC /// Default retryable HTTP status codes. pub const DEFAULT_RETRYABLE_STATUS_CODES: &[u16] = &[408, 429, 500, 502, 503, 504]; +/// Default capacity-pushback status codes: retried elsewhere but recorded as +/// neither success nor failure by the circuit breaker. 429 is unambiguous +/// backpressure; 503 stays a fault by default (indistinguishable from an +/// outage) and can be added per worker for backends that use it as pushback. +pub const DEFAULT_CAPACITY_STATUS_CODES: &[u16] = &[429]; + /// Fully resolved resilience configuration for a worker. /// /// Created by merging `RouterConfig` defaults with `WorkerSpec.resilience` overrides. @@ -26,6 +32,9 @@ pub struct ResolvedResilience { pub circuit_breaker_enabled: bool, /// Set of HTTP status codes considered retryable. pub retryable_status_codes: HashSet, + /// Statuses treated as capacity pushback: retryable, but excluded from + /// circuit-breaker accounting entirely. + pub capacity_status_codes: HashSet, } impl Default for ResolvedResilience { @@ -35,6 +44,7 @@ impl Default for ResolvedResilience { retry_enabled: true, circuit_breaker_enabled: true, retryable_status_codes: DEFAULT_RETRYABLE_STATUS_CODES.iter().copied().collect(), + capacity_status_codes: DEFAULT_CAPACITY_STATUS_CODES.iter().copied().collect(), } } } @@ -91,18 +101,34 @@ pub fn resolve_resilience( .map(|d| !d) .unwrap_or(base_cb_enabled); - // Resolve retryable status codes + // Both sets replace their defaults wholesale, and the capacity set is + // deliberately not unioned into the retryable one. Retryability is not + // decided here: every retry path goes through + // `routers::common::retry::is_retryable_status`, a router-global rule that + // always covers 429, so a partial override here cannot make capacity + // pushback non-retryable. `retryable_status_codes` has exactly one + // consumer - circuit-breaker failure classification in + // `Worker::record_outcome` - and that check is unreachable for capacity + // codes, which return early just above it. Merging the sets would add no + // behaviour and would break the documented replace-the-default contract. let retryable_status_codes = overrides .retryable_status_codes .as_ref() .map(|codes| codes.iter().copied().collect()) .unwrap_or_else(|| DEFAULT_RETRYABLE_STATUS_CODES.iter().copied().collect()); + let capacity_status_codes = overrides + .capacity_status_codes + .as_ref() + .map(|codes| codes.iter().copied().collect()) + .unwrap_or_else(|| DEFAULT_CAPACITY_STATUS_CODES.iter().copied().collect()); + let resolved = ResolvedResilience { retry, retry_enabled, circuit_breaker_enabled: cb_enabled, retryable_status_codes, + capacity_status_codes, }; (resolved, cb_config) @@ -110,7 +136,10 @@ pub fn resolve_resilience( #[cfg(test)] mod tests { + use axum::http::StatusCode; + use super::*; + use crate::routers::common::retry::is_retryable_status; #[test] fn test_default_resilience() { @@ -218,4 +247,61 @@ mod tests { // Non-overridden assert_eq!(cb_config.success_threshold, base_cb.success_threshold); } + + #[test] + fn capacity_codes_default_to_429_only() { + let resolved = ResolvedResilience::default(); + assert!(resolved.capacity_status_codes.contains(&429)); + assert_eq!(resolved.capacity_status_codes.len(), 1); + // 503 stays a plain retryable fault by default. + assert!(!resolved.capacity_status_codes.contains(&503)); + assert!(resolved.retryable_status_codes.contains(&503)); + } + + #[test] + fn capacity_codes_override_replaces_default() { + // The override omits the default 429 so this pins replacement, not + // union: a superset override would pass either way. + let overrides = ResilienceUpdate { + capacity_status_codes: Some(vec![503]), + ..Default::default() + }; + let (resolved, _cb) = resolve_resilience( + &RetryConfig::default(), + &CircuitBreakerConfig::default(), + true, + true, + &overrides, + ); + assert!(resolved.capacity_status_codes.contains(&503)); + assert!(!resolved.capacity_status_codes.contains(&429)); + assert_eq!(resolved.capacity_status_codes.len(), 1); + } + + #[test] + fn partial_retryable_override_leaves_capacity_pushback_retryable() { + // Pins why the capacity set is not merged into the retryable set: a + // per-worker retryable override that drops 429 cannot make pushback + // non-retryable, because the retry gate is router-global. + let overrides = ResilienceUpdate { + retryable_status_codes: Some(vec![500, 502]), + ..Default::default() + }; + let (resolved, _cb) = resolve_resilience( + &RetryConfig::default(), + &CircuitBreakerConfig::default(), + true, + true, + &overrides, + ); + + // The override replaces the default set verbatim - 429 is not merged in. + assert!(!resolved.retryable_status_codes.contains(&429)); + // Yet 429 is still retried, because retryability never consults this set. + assert!(is_retryable_status(StatusCode::TOO_MANY_REQUESTS)); + // And it still records no circuit-breaker sample, because + // `record_outcome` returns on the capacity set before it ever reads + // `retryable_status_codes`. + assert!(resolved.capacity_status_codes.contains(&429)); + } } diff --git a/model_gateway/src/worker/worker.rs b/model_gateway/src/worker/worker.rs index d93abf1f61..d28a80e596 100644 --- a/model_gateway/src/worker/worker.rs +++ b/model_gateway/src/worker/worker.rs @@ -453,18 +453,28 @@ pub trait Worker: Send + Sync + fmt::Debug + 'static { /// Record the outcome of a request based on the HTTP status code. /// - /// The worker decides whether the status is a CB failure using its - /// per-worker `retryable_status_codes` set (default: 408, 429, 5xx). + /// Statuses in the per-worker `capacity_status_codes` set (default: 429) + /// record nothing at all — neither failure nor success. Any other status + /// is a circuit-breaker failure when it appears in `retryable_status_codes`, + /// which by default leaves 408, 500, 502, 503 and 504 tripping the breaker. + /// 429 is in that set too, but the capacity check returns before it is read. /// Callers just pass the status — no need to interpret it. /// /// For transport/connection errors where no HTTP response is received, /// pass the status code returned to the client (e.g., 502 for a send /// error, 504 for a timeout). fn record_outcome(&self, status_code: u16) { - let is_failure = self - .resilience() - .retryable_status_codes - .contains(&status_code); + let resilience = self.resilience(); + // Capacity pushback (429 by default) is a routing signal, not a + // worker fault: the request is retried elsewhere, but no + // circuit-breaker sample is recorded in either direction — opening + // the breaker on backpressure would amplify a load spike into + // unavailability, and crediting a success would close a half-open + // breaker on a request the worker refused. + if resilience.capacity_status_codes.contains(&status_code) { + return; + } + let is_failure = resilience.retryable_status_codes.contains(&status_code); self.record_circuit_breaker_outcome(!is_failure); } @@ -2320,6 +2330,56 @@ mod tests { assert!(!worker.circuit_breaker_can_execute()); } + #[test] + fn test_capacity_pushback_never_trips_circuit_breaker() { + let worker = BasicWorkerBuilder::new("http://test:8080") + .worker_type(WorkerType::Regular) + .health_config(no_health_check()) + .build(); + + // A storm of 429 capacity pushback must not open the breaker... + for _ in 0..20 { + worker.record_outcome(429); + } + assert!(worker.is_available()); + assert_eq!(worker.circuit_breaker_state(), CircuitState::Closed); + + // ...while genuine failures still do. + for _ in 0..5 { + worker.record_outcome(500); + } + assert!(!worker.is_available()); + } + + #[test] + fn test_capacity_pushback_does_not_close_half_open_breaker() { + let config = CircuitBreakerConfig { + failure_threshold: 2, + success_threshold: 1, + timeout_duration: Duration::from_millis(50), + window_duration: Duration::from_secs(60), + }; + let worker = BasicWorkerBuilder::new("http://test:8080") + .worker_type(WorkerType::Regular) + .circuit_breaker_config(config) + .health_config(no_health_check()) + .build(); + + worker.record_outcome(500); + worker.record_outcome(500); + assert!(!worker.is_available()); + thread::sleep(Duration::from_millis(80)); + assert!(worker.is_available()); + assert_eq!(worker.circuit_breaker_state(), CircuitState::HalfOpen); + + // 429 records no sample: the breaker must stay half-open, not close. + worker.record_outcome(429); + assert_eq!(worker.circuit_breaker_state(), CircuitState::HalfOpen); + + worker.record_outcome(200); + assert_eq!(worker.circuit_breaker_state(), CircuitState::Closed); + } + #[test] fn test_worker_with_circuit_breaker_config() { let config = CircuitBreakerConfig {