impl BackendState { fn new(mode: HealthMode) -> Self { let now = Instant::now(); Self { healthy_cached: AtomicBool::new(mode == HealthMode::Off), health: Mutex::new(Health { healthy: mode == HealthMode::Off, active_healthy: mode == HealthMode::Off, passive_healthy: true, checked: false, failures: 0, passive_failures: 0, successes: 0, passive_successes: 0, next_probe: now, probing: false, healthy_since: now, result: if mode == HealthMode::Off { "disabled" } else { "checking" }, passive_result: "none", last_probe_ms: 0, latency_us: 0, ewma_us: 0, transitions: 0, unhealthy_source: if mode == HealthMode::Off { "none" } else { "active" }, unhealthy_reason: if mode == HealthMode::Off { "none" } else { "checking" }, active_unhealthy_source: if mode == HealthMode::Off { "none" } else { "active" }, active_unhealthy_reason: if mode == HealthMode::Off { "none" } else { "checking" }, passive_unhealthy_source: "none", passive_unhealthy_reason: "none", passive_first_failure: None, active_first_failure: None, passive_probe_successes: 0, passive_ejected_at: None, passive_cooldown: Duration::from_secs(5), last_io_counted_at: None, }), traffic: Mutex::new(Traffic { at: now, total: 0, rate: 0.0, }), last_failure: Mutex::new(FailureObservation::default()), active: AtomicU64::new(0), requests: AtomicU64::new(0), errors: AtomicU64::new(0), bytes_in: AtomicU64::new(0), bytes_out: AtomicU64::new(0), last_status: AtomicU16::new(0), active_health_failures_total: AtomicU64::new(0), passive_health_failures_total: AtomicU64::new(0), http_5xx_total: AtomicU64::new(0), timeouts_total: AtomicU64::new(0), connect_errors_total: AtomicU64::new(0), io_errors_total: AtomicU64::new(0), slow_responses_total: AtomicU64::new(0), retry_attempts_total: AtomicU64::new(0), retry_successes_total: AtomicU64::new(0), capacity_rejections_total: AtomicU64::new(0), queue_timeouts_total: AtomicU64::new(0), dns_errors_total: AtomicU64::new(0), tls_errors_total: AtomicU64::new(0), response_header_timeouts_total: AtomicU64::new(0), read_timeouts_total: AtomicU64::new(0), send_timeouts_total: AtomicU64::new(0), last_ejected_ms: AtomicU64::new(0), last_recovered_ms: AtomicU64::new(0), } } fn has_capacity(&self, limit: u32) -> bool { limit == 0 || self.active.load(Ordering::Relaxed) < u64::from(limit) } fn try_acquire_active(&self, limit: u32) -> bool { if limit == 0 { self.active.fetch_add(1, Ordering::Relaxed); return true; } let limit = u64::from(limit); let mut current = self.active.load(Ordering::Relaxed); loop { if current >= limit { return false; } match self.active.compare_exchange_weak( current, current + 1, Ordering::AcqRel, Ordering::Relaxed, ) { Ok(_) => return true, Err(next) => current = next, } } } pub fn healthy(&self) -> bool { if self.healthy_cached.load(Ordering::Acquire) { return true; } let mut h = lock(&self.health); if !h.passive_healthy && h.active_healthy && !h.checked && h.passive_ejected_at .is_some_and(|at| at.elapsed() >= h.passive_cooldown) { let previous = h.healthy; Self::clear_passive_ejection(&mut h); self.finish_transition(&mut h, previous); } h.healthy } #[cfg(test)] fn set_healthy_for_test(&self, healthy: bool) { lock(&self.health).healthy = healthy; self.healthy_cached.store(healthy, Ordering::Release); } fn clear_passive_ejection(h: &mut Health) { h.passive_healthy = true; h.passive_failures = 0; h.passive_successes = 0; h.passive_probe_successes = 0; h.passive_first_failure = None; h.passive_ejected_at = None; h.passive_unhealthy_source = "none"; h.passive_unhealthy_reason = "none"; } fn weight(&self, configured: u16, slow_start: u64) -> f64 { let h = lock(&self.health); if slow_start == 0 { return configured as f64; } let ratio = (h.healthy_since.elapsed().as_secs_f64() / slow_start as f64).clamp(0.05, 1.0); (configured as f64 * ratio).max(0.05) } fn latency(&self) -> f64 { lock(&self.health).ewma_us.max(1000) as f64 / 1000.0 } fn traffic_rate(&self) -> f64 { lock(&self.traffic).rate } fn sample_traffic(&self) { let mut sample = lock(&self.traffic); let elapsed = sample.at.elapsed().as_secs_f64(); if elapsed < 1.0 { return; } let total = self .bytes_in .load(Ordering::Relaxed) .saturating_add(self.bytes_out.load(Ordering::Relaxed)); let delta = total.saturating_sub(sample.total); if delta == 0 { sample.rate = 0.0; } else { let rate = delta as f64 / elapsed; sample.rate = sample.rate * 0.5 + rate * 0.5; } sample.total = total; sample.at = Instant::now(); } fn claim_probe(&self, h: &HealthCheck) -> bool { if h.mode == HealthMode::Off { return false; } let mut state = lock(&self.health); if state.probing || Instant::now() < state.next_probe { return false; } state.probing = true; state.next_probe = Instant::now() + Duration::from_millis(h.interval_ms); true } fn update_latency(h: &mut Health, elapsed: Duration) { let latency = elapsed.as_micros().min(u64::MAX as u128) as u64; h.latency_us = latency; h.ewma_us = if h.ewma_us == 0 { latency } else { h.ewma_us.saturating_mul(7).saturating_add(latency) / 8 }; } fn finish_transition(&self, h: &mut Health, previous: bool) { h.healthy = h.active_healthy && h.passive_healthy; self.healthy_cached.store(h.healthy, Ordering::Release); if previous != h.healthy { h.transitions = h.transitions.saturating_add(1); if h.healthy { self.last_recovered_ms .store(crate::logging::now_ms(), Ordering::Relaxed); h.healthy_since = Instant::now(); h.unhealthy_source = "none"; h.unhealthy_reason = "none"; return; } else { self.last_ejected_ms .store(crate::logging::now_ms(), Ordering::Relaxed); } } if h.healthy { return; } let current_source_is_still_unhealthy = (h.unhealthy_source.starts_with("active") && !h.active_healthy) || (h.unhealthy_source.starts_with("passive") && !h.passive_healthy); if current_source_is_still_unhealthy { return; } if !h.active_healthy { h.unhealthy_source = h.active_unhealthy_source; h.unhealthy_reason = h.active_unhealthy_reason; } else if !h.passive_healthy { h.unhealthy_source = h.passive_unhealthy_source; h.unhealthy_reason = h.passive_unhealthy_reason; } } fn observe_active(&self, settings: &HealthCheck, result: &'static str, elapsed: Duration) { if settings.mode == HealthMode::Off { return; } let mut h = lock(&self.health); let previous = h.healthy; Self::update_latency(&mut h, elapsed); h.probing = false; h.checked = true; h.last_probe_ms = crate::logging::now_ms(); h.result = result; if result == "ok" { h.active_first_failure = None; h.failures = 0; h.successes = h.successes.saturating_add(1); if h.successes >= settings.rise { h.active_healthy = true; } if !h.passive_healthy && h.passive_ejected_at.is_some() && h.active_healthy { h.passive_probe_successes = h.passive_probe_successes.saturating_add(1); if h.passive_probe_successes >= settings.rise.max(2) { Self::clear_passive_ejection(&mut h); } } } else { h.passive_probe_successes = 0; if h.active_first_failure.is_none() { h.active_first_failure = Some(FailureObservation { at_ms: crate::logging::now_ms(), class: "active_health", detail: result.into(), }); } self.active_health_failures_total .fetch_add(1, Ordering::Relaxed); h.successes = 0; h.failures = h.failures.saturating_add(1); if h.failures >= settings.fall { h.active_healthy = false; } } let source = match settings.mode { HealthMode::Tcp => "active_tcp", HealthMode::Http => "active_http", HealthMode::Off => "active", }; if result != "ok" { self.record_failure("active_health", &format!("{source}:{result}")); } if h.active_healthy { h.active_unhealthy_source = "none"; h.active_unhealthy_reason = "none"; } else { h.active_unhealthy_source = source; h.active_unhealthy_reason = result; if previous || h.unhealthy_source == "active" || h.unhealthy_source == "none" { h.unhealthy_source = source; h.unhealthy_reason = result; } } self.finish_transition(&mut h, previous); } fn first_active_failure(&self) -> Option { lock(&self.health).active_first_failure.clone() } fn record_failure(&self, class: &'static str, detail: &str) { match class { "dns" => { self.dns_errors_total.fetch_add(1, Ordering::Relaxed); } "tls" => { self.tls_errors_total.fetch_add(1, Ordering::Relaxed); } "response_header_timeout" => { self.response_header_timeouts_total .fetch_add(1, Ordering::Relaxed); } "read_timeout" => { self.read_timeouts_total.fetch_add(1, Ordering::Relaxed); } "send_timeout" => { self.send_timeouts_total.fetch_add(1, Ordering::Relaxed); } _ => {} } let mut last = lock(&self.last_failure); last.at_ms = crate::logging::now_ms(); last.class = class; last.detail.clear(); last.detail.extend(detail.chars().take(1024)); } fn record_passive_observation(&self, reason: &'static str) { match reason { "http_5xx" => { self.http_5xx_total.fetch_add(1, Ordering::Relaxed); } "timeout" => { self.timeouts_total.fetch_add(1, Ordering::Relaxed); } "connect_error" => { self.connect_errors_total.fetch_add(1, Ordering::Relaxed); } "io_error" => { self.io_errors_total.fetch_add(1, Ordering::Relaxed); } "slow_response" => { self.slow_responses_total.fetch_add(1, Ordering::Relaxed); } _ => {} } } #[cfg(test)] fn observe_passive( &self, active: &HealthCheck, passive: &PassiveHealth, reason: &'static str, elapsed: Duration, ) { self.observe_passive_with_detail(active, passive, reason, reason, reason, elapsed); } fn observe_passive_with_detail( &self, active: &HealthCheck, passive: &PassiveHealth, reason: &'static str, class: &'static str, detail: &str, elapsed: Duration, ) -> Option<(FailureObservation, u16)> { self.record_passive_observation(reason); let mut h = lock(&self.health); Self::update_latency(&mut h, elapsed); h.passive_result = reason; if !passive.enabled { return None; } let previous = h.healthy; if reason == "ok" { h.passive_failures = 0; h.passive_first_failure = None; h.last_io_counted_at = None; h.passive_successes = h.passive_successes.saturating_add(1); if h.passive_successes >= passive.rise(active) { Self::clear_passive_ejection(&mut h); } self.finish_transition(&mut h, previous); return None; } if !passive.counts(reason) { return None; } if reason == "io_error" { let now = Instant::now(); if h.last_io_counted_at .is_some_and(|last| now.duration_since(last) < Duration::from_millis(250)) { return None; } h.last_io_counted_at = Some(now); } self.passive_health_failures_total .fetch_add(1, Ordering::Relaxed); if h.passive_failures == 0 { h.passive_first_failure = Some(FailureObservation { at_ms: crate::logging::now_ms(), class, detail: detail.chars().take(1024).collect(), }); } h.passive_successes = 0; h.passive_probe_successes = 0; h.passive_failures = h.passive_failures.saturating_add(1); if h.passive_failures >= passive.fall(active) && h.passive_healthy { h.passive_healthy = false; h.passive_ejected_at = Some(Instant::now()); h.passive_cooldown = Duration::from_millis(active.interval_ms.clamp(5_000, 30_000)); } let source = match reason { "connect_error" => "passive_connect_error", "timeout" => "passive_timeout", "io_error" => "passive_io_error", "http_5xx" => "passive_http_5xx", "slow_response" => "passive_slow", _ => "passive", }; if !h.passive_healthy { h.passive_unhealthy_source = source; h.passive_unhealthy_reason = reason; if previous { h.unhealthy_source = source; h.unhealthy_reason = reason; } } self.finish_transition(&mut h, previous); if previous && !h.healthy { h.passive_first_failure .clone() .map(|first| (first, h.passive_failures)) } else { None } } } #[derive(Debug)] pub struct Member { pub backend: Arc, pub definition: PoolMember, pub state: Arc, health_probe: Option>, health_settings: Arc, passive_settings: Arc, token: String, } #[derive(Debug)] pub struct Pool { pub id: String, pub settings: Settings, pub members: Vec>, pub fingerprint: Vec, cookie_name: String, cursor: AtomicU64, scheduling: Mutex>, } #[derive(Debug)] pub struct Lease { pub member: Arc, health: Arc, passive_health: Arc, error_recorded: AtomicBool, }