impl Lease {
    pub fn started(&self) {
        self.member.state.requests.fetch_add(1, Ordering::Relaxed);
    }
    pub fn response(&self, status: u16, elapsed: Duration, request_body_empty: bool) {
        self.member
            .state
            .last_status
            .store(status, Ordering::Relaxed);
        let is_5xx = status >= 500;
        let is_slow = request_body_empty
            && self.health.slow_response_ms > 0
            && elapsed > Duration::from_millis(self.health.slow_response_ms);
        if (is_5xx || is_slow) && !self.error_recorded.swap(true, Ordering::Relaxed) {
            self.member.state.errors.fetch_add(1, Ordering::Relaxed);
        }
        if is_5xx {
            self.member
                .state
                .record_failure("http_5xx", &format!("status={status}"));
        }
        let reason = if is_5xx
            && (!is_slow
                || !self.passive_health.enabled
                || self.passive_health.count_http_5xx
                || !self.passive_health.count_slow_responses)
        {
            if is_slow {
                self.member
                    .state
                    .record_passive_observation("slow_response");
            }
            "http_5xx"
        } else if is_slow {
            if is_5xx {
                self.member.state.record_passive_observation("http_5xx");
            }
            "slow_response"
        } else {
            "ok"
        };
        let before = self.member.state.healthy();
        let failure_detail = is_5xx.then(|| format!("status={status}"));
        let detail = failure_detail.as_deref().unwrap_or(reason);
        let ejected = self.member.state.observe_passive_with_detail(
            &self.health,
            &self.passive_health,
            reason,
            reason,
            detail,
            elapsed,
        );
        if let Some((first, failures)) = ejected {
            tracing::warn!(backend = %self.member.definition.id, origin = %self.member.backend.origin, source = "passive",
                reason, first_error_class = first.class,
                first_error_at_ms = first.at_ms, first_error_detail = %first.detail,
                trigger_error_detail = %detail, failures,
                "backend excluded from traffic");
        }
        let after = self.member.state.healthy();
        if before != after {
            tracing::info!(
                backend = %self.member.definition.id,
                healthy = after,
                source = "passive",
                reason,
                "backend health changed"
            );
        }
    }
    pub fn error(&self, reason: &'static str, elapsed: Duration) {
        self.error_detail(reason, reason, elapsed);
    }
    pub fn error_detail(&self, class: &'static str, detail: &str, elapsed: Duration) {
        if !self.error_recorded.swap(true, Ordering::Relaxed) {
            self.member.state.errors.fetch_add(1, Ordering::Relaxed);
        }
        self.member.state.record_failure(class, detail);
        if matches!(class, "h2_graceful_goaway" | "h2_graceful_close") {
            return;
        }
        let passive_reason = match class {
            "connect" | "dns" | "tls" => "connect_error",
            "timeout"
            | "body_timeout"
            | "response_header_timeout"
            | "read_timeout"
            | "send_timeout" => "timeout",
            _ => "io_error",
        };
        let before = self.member.state.healthy();
        let ejected = self.member.state.observe_passive_with_detail(
            &self.health,
            &self.passive_health,
            passive_reason,
            class,
            detail,
            elapsed,
        );
        if let Some((first, failures)) = ejected {
            tracing::warn!(backend = %self.member.definition.id, origin = %self.member.backend.origin, source = "passive",
                reason = passive_reason, first_error_class = first.class,
                first_error_at_ms = first.at_ms, first_error_detail = %first.detail,
                trigger_error_detail = %detail, failures,
                "backend excluded from traffic");
        }
        let after = self.member.state.healthy();
        if before != after {
            tracing::info!(
                backend = %self.member.definition.id,
                healthy = after,
                source = "passive",
                reason = passive_reason,
                "backend health changed"
            );
        }
    }
    pub fn bytes(&self, count: u64, incoming: bool) {
        if incoming {
            self.member
                .state
                .bytes_in
                .fetch_add(count, Ordering::Relaxed);
        } else {
            self.member
                .state
                .bytes_out
                .fetch_add(count, Ordering::Relaxed);
        }
    }
    pub fn retry_attempted(&self) {
        self.member
            .state
            .retry_attempts_total
            .fetch_add(1, Ordering::Relaxed);
    }
    pub fn retry_succeeded(&self) {
        self.member
            .state
            .retry_successes_total
            .fetch_add(1, Ordering::Relaxed);
    }
}
impl Drop for Lease {
    fn drop(&mut self) {
        self.member.state.active.fetch_sub(1, Ordering::Relaxed);
    }
}
pub struct Selection {
    pub lease: Arc<Lease>,
    pub cookie: Option<HeaderValue>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SelectionFailure {
    Unavailable,
    Capacity,
}

impl Pool {
    pub fn prometheus(&self) -> String {
        use std::fmt::Write;
        let mut text = String::new();
        for member in &self.members {
            let state = &member.state;
            let labels = format!("vhost=\"{}\",backend=\"{}\"", self.id, member.definition.id);
            for (metric, value) in [
                (
                    "up",
                    u64::from(state.healthy() && member.definition.enabled),
                ),
                ("draining", u64::from(member.definition.drain)),
                ("active_exchanges", state.active.load(Ordering::Relaxed)),
                ("requests_total", state.requests.load(Ordering::Relaxed)),
                ("errors_total", state.errors.load(Ordering::Relaxed)),
                ("bytes_in_total", state.bytes_in.load(Ordering::Relaxed)),
                ("bytes_out_total", state.bytes_out.load(Ordering::Relaxed)),
            ] {
                let _ = writeln!(text, "rust_proxy_backend_{metric}{{{labels}}} {value}");
            }
            for (metric, value) in [
                (
                    "active_health_failures_total",
                    state.active_health_failures_total.load(Ordering::Relaxed),
                ),
                (
                    "passive_health_failures_total",
                    state.passive_health_failures_total.load(Ordering::Relaxed),
                ),
                (
                    "http_5xx_total",
                    state.http_5xx_total.load(Ordering::Relaxed),
                ),
                (
                    "passive_http_5xx_total",
                    state.http_5xx_total.load(Ordering::Relaxed),
                ),
                (
                    "timeouts_total",
                    state.timeouts_total.load(Ordering::Relaxed),
                ),
                (
                    "passive_timeouts_total",
                    state.timeouts_total.load(Ordering::Relaxed),
                ),
                (
                    "connect_errors_total",
                    state.connect_errors_total.load(Ordering::Relaxed),
                ),
                (
                    "passive_connect_errors_total",
                    state.connect_errors_total.load(Ordering::Relaxed),
                ),
                (
                    "io_errors_total",
                    state.io_errors_total.load(Ordering::Relaxed),
                ),
                (
                    "passive_io_errors_total",
                    state.io_errors_total.load(Ordering::Relaxed),
                ),
                (
                    "slow_responses_total",
                    state.slow_responses_total.load(Ordering::Relaxed),
                ),
                (
                    "passive_slow_total",
                    state.slow_responses_total.load(Ordering::Relaxed),
                ),
                (
                    "retry_attempts_total",
                    state.retry_attempts_total.load(Ordering::Relaxed),
                ),
                (
                    "retry_successes_total",
                    state.retry_successes_total.load(Ordering::Relaxed),
                ),
                (
                    "capacity_rejections_total",
                    state.capacity_rejections_total.load(Ordering::Relaxed),
                ),
                (
                    "queue_timeouts_total",
                    state.queue_timeouts_total.load(Ordering::Relaxed),
                ),
                (
                    "dns_errors_total",
                    state.dns_errors_total.load(Ordering::Relaxed),
                ),
                (
                    "tls_errors_total",
                    state.tls_errors_total.load(Ordering::Relaxed),
                ),
                (
                    "response_header_timeouts_total",
                    state.response_header_timeouts_total.load(Ordering::Relaxed),
                ),
                (
                    "read_timeouts_total",
                    state.read_timeouts_total.load(Ordering::Relaxed),
                ),
                (
                    "send_timeouts_total",
                    state.send_timeouts_total.load(Ordering::Relaxed),
                ),
            ] {
                let _ = writeln!(text, "rust_proxy_backend_{metric}{{{labels}}} {value}");
            }
            for (metric, value) in [
                (
                    "last_ejected_timestamp_milliseconds",
                    state.last_ejected_ms.load(Ordering::Relaxed),
                ),
                (
                    "last_recovered_timestamp_milliseconds",
                    state.last_recovered_ms.load(Ordering::Relaxed),
                ),
            ] {
                let _ = writeln!(text, "rust_proxy_backend_{metric}{{{labels}}} {value}");
            }
            let _ = writeln!(
                text,
                "rust_proxy_backend_traffic_bytes_per_second{{{labels}}} {:.3}",
                state.traffic_rate()
            );
            let _ = writeln!(
                text,
                "rust_proxy_backend_latency_milliseconds{{{labels}}} {:.3}",
                state.latency()
            );
        }
        text
    }
}
async fn probe(member: Arc<Member>, settings: Arc<HealthCheck>) {
    let start = Instant::now();
    let result = tokio::time::timeout(Duration::from_millis(settings.timeout_ms), async {
        match settings.mode {
            HealthMode::Tcp => {
                crate::upstream::tcp_probe(&member.backend.origin, &member.backend.address_policy)
                    .await
                    .map(|_| "ok")
                    .unwrap_or("connect")
            }
            HealthMode::Http => {
                let request = Request::builder()
                    .uri(format!(
                        "{}{}",
                        member.backend.request_origin, settings.path
                    ))
                    .header(
                        "host",
                        settings
                            .host
                            .as_deref()
                            .unwrap_or_else(|| member.backend.host_header.to_str().unwrap_or("")),
                    )
                    .header("connection", "close")
                    .header(
                        "user-agent",
                        concat!("RustProxy-health/", env!("CARGO_PKG_VERSION")),
                    )
                    .body(Body::empty());
                match request {
                    Ok(request) => match member.backend.websocket_client.request(request).await {
                        Ok(response) => {
                            if (settings.expected_status_min..=settings.expected_status_max)
                                .contains(&response.status().as_u16())
                            {
                                "ok"
                            } else {
                                "http_status"
                            }
                        }
                        Err(_) => "connect",
                    },
                    Err(_) => "request",
                }
            }
            HealthMode::Off => "ok",
        }
    })
    .await
    .unwrap_or("timeout");
    let elapsed = start.elapsed();
    let result = if result == "ok"
        && settings.slow_response_ms > 0
        && elapsed > Duration::from_millis(settings.slow_response_ms)
    {
        "slow"
    } else {
        result
    };
    let before = member.state.healthy();
    member.state.observe_active(&settings, result, elapsed);
    let after = member.state.healthy();
    if before && !after {
        let first = member.state.first_active_failure();
        tracing::warn!(backend = %member.definition.id, origin = %member.backend.origin, source = "active",
            reason = result,
            first_error_class = first.as_ref().map_or("active_health", |f| f.class),
            first_error_at_ms = first.as_ref().map_or(0, |f| f.at_ms),
            first_error_detail = %first.as_ref().map_or(result, |f| f.detail.as_str()),
            "backend excluded from traffic");
    }
    if before != after {
        tracing::info!(backend = %member.definition.id, healthy = after, reason = result, "backend health changed");
    }
}
pub async fn health_loop(app: Weak<crate::runtime::App>, shutdown: CancellationToken) {
    let mut interval = tokio::time::interval(Duration::from_millis(250));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    let mut checks = JoinSet::new();
    let mut cursor = 0usize;
    loop {
        tokio::select! {
            biased;
            _ = shutdown.cancelled() => break,
            _ = checks.join_next(), if !checks.is_empty() => {},
            _ = interval.tick() => {
                let Some(app) = app.upgrade() else { break; };
                let snapshot = app.config.load_full();
                let members: Vec<_> = snapshot.pools.iter().flat_map(|pool|
                    pool.members.iter().map(|member|
                        (Arc::clone(member), Arc::clone(&member.health_settings)))).collect();
                for (member, _) in &members {
                    member.state.sample_traffic();
                    member.backend.client.maintain_rotation();
                    member.backend.websocket_client.maintain_rotation();
                }
                let start = cursor;
                for offset in 0..members.len() {
                    if checks.len() >= 16 { break; }
                    let index = (start + offset) % members.len();
                    let (member, health) = &members[index];
                    if member.definition.enabled && member.state.claim_probe(health) {
                        checks.spawn(probe(Arc::clone(member), health.clone()));
                        cursor = (index + 1) % members.len();
                    }
                }
            },
        }
    }
    checks.abort_all();
    while checks.join_next().await.is_some() {}
}