Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions crates/protocols/src/worker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1062,9 +1062,17 @@ pub struct ResilienceUpdate {
pub disable_circuit_breaker: Option<bool>,

// ── 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<Vec<u16>>,
/// 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<Vec<u16>>,
}

impl ResilienceUpdate {
Expand All @@ -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()
}
}

Expand Down
88 changes: 87 additions & 1 deletion model_gateway/src/worker/resilience.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -26,6 +32,9 @@ pub struct ResolvedResilience {
pub circuit_breaker_enabled: bool,
/// Set of HTTP status codes considered retryable.
pub retryable_status_codes: HashSet<u16>,
/// Statuses treated as capacity pushback: retryable, but excluded from
/// circuit-breaker accounting entirely.
pub capacity_status_codes: HashSet<u16>,
}

impl Default for ResolvedResilience {
Expand All @@ -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(),
}
}
}
Expand Down Expand Up @@ -91,26 +101,45 @@ 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());
Comment thread
coderabbitai[bot] marked this conversation as resolved.

let resolved = ResolvedResilience {
retry,
retry_enabled,
circuit_breaker_enabled: cb_enabled,
retryable_status_codes,
capacity_status_codes,
};

(resolved, cb_config)
}

#[cfg(test)]
mod tests {
use axum::http::StatusCode;

use super::*;
use crate::routers::common::retry::is_retryable_status;

#[test]
fn test_default_resilience() {
Expand Down Expand Up @@ -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));
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
72 changes: 66 additions & 6 deletions model_gateway/src/worker/worker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Comment thread
pallasathena92 marked this conversation as resolved.
// 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);
}

Expand Down Expand Up @@ -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 {
Expand Down
Loading