dwara_core/dataplane/

upstream.rs

//! Pooled upstream clients (DW-008, feature analysis 4.1).
//!
//! One hyper-util legacy client (a connection pool) per configured
//! upstream, keyed by upstream name inside an [`UpstreamRegistry`]. Each
//! pool's connector is tuned per upstream:
//!
//! - **Connect timeout**: `timeouts.connect_ms` (default 5 s) wraps the
//!   whole dial: resolution plus TCP connect plus, for TLS upstreams,
//!   the TLS handshake.
//! - **Happy-eyeballs dial** (DW-030, RFC 8305): an endpoint address
//!   that resolves to multiple addresses — the dual-stack hostname
//!   case — is dialed with interleaved address-family attempts
//!   (`timeouts.happy_eyeballs_ms`, default 250 ms, `0` disables):
//!   the resolver's first address defines the preferred family
//!   (RFC 8305 leaves the preference to the resolver's order — the
//!   system's RFC 6724 sort), the first attempt starts immediately,
//!   each subsequent attempt starts one delay after the previous
//!   START (or immediately when an attempt fails with nothing else in
//!   flight), and the first success wins, cancelling the losers.
//!   Exactly ONE outcome — the overall dial's — reaches the breaker
//!   and passive-health accounting: the losing arms of one dial are
//!   dial-internal retries, never endpoint failures. The dial (and
//!   the [`happy_race`] primitive beneath it) is ours, not
//!   hyper-util's implicit default: the RFC 8305 shape is now
//!   documented, configurable, and observable in tests. DNS resolution
//!   on the pooled dial path runs through the shared async hickory
//!   `DnsCache` (PERF-05/#170)
//!   with TTL and negative caching, replacing the blocking-pool
//!   `getaddrinfo` of `tokio::net::lookup_host`; IP-literal and
//!   `/etc/hosts` endpoints skip the network. The active health probes
//!   keep the system resolver (a background, infrequent path).
//! - **Read timeout (per-attempt)**: `timeouts.read_ms` wraps each pooled
//!   request/response-header exchange (DW-014) — from the moment the
//!   request is handed to the pool (including any connection-cap queue
//!   wait and the dial) until the response HEADERS resolve. It is
//!   therefore also the per-attempt total bound: pool wait + connect +
//!   write request + read headers, all inside one `read_ms` deadline.
//!   A request whose headers do not arrive in time fails with
//!   [`UpstreamError::ReadTimeout`] (classified 504 by the proxy, and a
//!   retryable transport-class failure when retries are enabled). The
//!   response BODY is not covered by `read_ms`.
//! - **Write timeout (body idle)**: `timeouts.write_ms` bounds the
//!   response body stream as an INACTIVITY timeout (DW-014): the body
//!   wrapper [`UpstreamBody`] errors with [`UpstreamBodyError::WriteTimeout`]
//!   when the gap between two body frames exceeds `write_ms`. It is not a
//!   total streaming budget — a body that keeps trickling frames never
//!   trips it. The wrapper also reports a mid-stream abort (transport
//!   error or idle timeout after headers resolved) as a passive-health
//!   FAILURE for the picked endpoint, closing the DW-012 gap where
//!   mid-body deaths were invisible to ejection. Requires a body wrapper:
//!   the handle's response type is `Response<UpstreamBody>`, not
//!   `Response<Incoming>` (documented; the wrapper is a thin frame
//!   passthrough when both knobs are unset).
//! - **Connection cap**: `connection_cap` (default 256) bounds concurrent
//!   outbound connections to the upstream — active AND pooled-idle — via a
//!   semaphore permit acquired before dialing and stored inside the
//!   connection's IO wrapper, so it is released exactly when the
//!   connection closes (permit drop). Excess connection attempts queue on
//!   the semaphore; they are never failed, only delayed. Because the
//!   permit wait has no deadline of its own, requests can queue
//!   indefinitely while the cap is saturated; request-level timeouts
//!   (DW-014) will bound that wait. The effective cap is clamped to a
//!   minimum of 1: validated configs always pass a cap >= 1, but direct
//!   construction with cap == 0 would build a zero-permit semaphore whose
//!   dials hang silently — misuse degrades to serial connects instead.
//!   Documented choice:
//!   the cap counts connections, not in-flight requests — HTTP/1.1
//!   multiplexes several requests over one connection only sequentially,
//!   and h2 pools share one connection per origin anyway.
//! - **Pending cap** (DW-015): `max_pending` (default 256) bounds how many
//!   requests may WAIT for a connection-cap slot. Over that, the connector
//!   fails fast with [`UpstreamError::Saturated`] (classified 503 "upstream
//!   saturated" by the proxy) instead of queueing; a pending slot is held
//!   from the dial attempt until the connection-cap permit is acquired,
//!   then released (the request is connecting, no longer pending). The
//!   default bounds memory under high concurrency (#270); operators who
//!   need deeper queueing can set `max_pending` explicitly.
//! - **TLS**: `https` upstreams negotiate TLS with ALPN `http/1.1`;
//!   `http2` upstreams negotiate TLS with ALPN `h2` and lock the client to
//!   HTTP/2; `http1` upstreams dial plaintext. Server certificates are
//!   verified against the Mozilla webpki root set by default (chosen over
//!   system roots for determinism in tests; system roots are a follow-up).
//!   Private-CA upstreams are configured per upstream via
//!   `trusted_ca_file` (#121): the PEM bundle REPLACES the public roots
//!   for that upstream, and its active https health probes verify against
//!   the same roots (see the handle's [`UpstreamHandle::tls_roots`]).
//!   Programmatic callers can still add extra roots on top of the public
//!   set via [`UpstreamRegistry::with_root_certificates`].
//!
//! Load balancing (DW-011): every dispatch picks its endpoint through the
//! upstream's [`crate::dataplane::balance::UpstreamLb`] (smooth weighted round-robin,
//! least-connections, random-2, or ketama ip-hash; slow-start ramps; hot
//! endpoint-set swaps that carry per-address state). Config lifecycle:
//! build a registry from a published snapshot; reloads rebuild it from the
//! new snapshot while carrying balancer state (see
//! [`UpstreamRegistry::from_snapshot_with_previous`]), mirroring how
//! `TlsTermination` is reloaded.
//!
//! Retries (DW-014) are driven from the proxy (which owns the request
//! body replay decision); the handle exposes the resolved
//! [`crate::resilience::retries::RetryParams`] and the per-upstream
//! [`crate::resilience::retries::RetryBudget`] for that loop, and every `send` call is
//! one ATTEMPT (fresh load-balancer pick, fresh per-attempt deadlines).

use std::collections::{BTreeMap, VecDeque};
use std::future::Future;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;

use http::uri::{Authority, PathAndQuery, Scheme};
use http_body_util::BodyExt as _;
use hyper::body::{Bytes, Frame, Incoming};
use hyper::{Request, Response, Uri, Version};
use hyper_util::client::legacy::connect::{Connected, Connection as HyperConnection};
use hyper_util::client::legacy::Client;
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use rustls::pki_types::{CertificateDer, ServerName};
use std::str::FromStr as _;
use tokio::net::TcpStream;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::task::JoinSet;
use tower_service::Service;

/// The hyper-util client type used by [`UpstreamHandle`], aliased to keep
/// the sharded `clients: Vec<UpstreamClient>` field readable.
type UpstreamClient = Client<
    UpstreamConnector,
    http_body_util::combinators::UnsyncBoxBody<
        bytes::Bytes,
        Box<dyn std::error::Error + Send + Sync>,
    >,
>;

use crate::config::limits::MAX_SLOW_START_MS;
use crate::config::{Timeouts, Upstream, UpstreamProtocol};
use crate::dataplane::discovery::DnsCache;
use crate::observability::Observability;
use crate::resilience::breaker::{Breaker, BreakerParams, BreakerState};
use crate::resilience::health::HealthDispatch;
use crate::resilience::retries::{HedgeParams, RetryBudget, RetryParams};
use crate::snapshot::Snapshot;

/// Default connection cap when `connection_cap` is absent.
///
/// Raised from 64 to 256 (#270): the previous default of 64 caused
/// unbounded memory growth under high concurrency — excess requests
/// queued on the semaphore with their full request context held in
/// memory, and the allocator retained those pages. 256 is a reasonable
/// middle ground for modern workloads; operators with higher
/// concurrency needs should set `connection_cap` explicitly.
pub const DEFAULT_CONNECTION_CAP: u32 = 256;
/// Default pending-request cap when `max_pending` is absent (#270).
///
/// Previously absent = unbounded queueing, which allowed memory to
/// grow without limit when the connection cap was saturated. A
/// default of 256 bounds the queue so excess requests get 503
/// "upstream saturated" (backpressure) instead of queueing
/// indefinitely. This aligns with the AGENTS.md rule that buffering
/// must be size-capped. Operators who need deeper queueing can set
/// `max_pending` explicitly.
pub const DEFAULT_MAX_PENDING: u32 = 256;
/// Default connect timeout when `timeouts.connect_ms` is absent.
pub const DEFAULT_CONNECT_TIMEOUT_MS: u64 = 5_000;
/// Maximum pool shard count (caps memory overhead of per-shard pools).
const MAX_POOL_SHARDS: usize = 16;

/// Number of hyper-util client pool shards per upstream. Each shard is an
/// independent `Client` with its own `Mutex<PoolInner>`, so the pool lock
/// contention that dominates the hot path under high concurrency is
/// distributed N-fold. The connector (and its connection-cap semaphore) is
/// shared across shards via `Arc`, so the TOTAL connection cap still
/// applies — the cap is NOT multiplied by the shard count.
///
/// `DWARA_POOL_SHARDS` env var overrides; default is 1 (the previous
/// single-pool behavior, preserving connection-reuse semantics). Set to
/// `available_parallelism()` (or the number of tokio workers) to reduce
/// pool-lock contention under high concurrency. The tradeoff: with N
/// shards, N connections to each upstream endpoint may be open (one per
/// shard) instead of 1, slightly increasing upstream connection count
/// (still bounded by `connection_cap`).
fn pool_shard_count() -> usize {
    let default = 1;
    std::env::var("DWARA_POOL_SHARDS")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .map(|n| n.clamp(1, MAX_POOL_SHARDS))
        .unwrap_or(default)
}

/// System millisecond clock for a freshly built breaker (the breaker's
/// own `system_now_ms` is private to its module).
fn system_now_ms_for_breaker() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// Error sending a request to an upstream.
#[derive(Debug)]
pub enum UpstreamError {
    /// `send` was called on an upstream with no endpoints. Only reachable
    /// via unvalidated direct construction; config validation rejects
    /// empty endpoint lists.
    NoEndpoints,
    /// A root certificate supplied to
    /// [`UpstreamRegistry::with_root_certificates`] could not be parsed
    /// or was otherwise unusable as a trust anchor.
    InvalidRootCertificate(String),
    /// The endpoint address is not usable as a TLS server name.
    InvalidHost(String),
    /// Dialing (TCP or TLS handshake) exceeded the connect timeout.
    ConnectTimeout { after: Duration },
    /// The per-attempt deadline (`timeouts.read_ms`) expired before the
    /// response headers resolved. Covers pool-queue wait + connect +
    /// request write + header read; see the module docs (DW-014).
    ReadTimeout { after: Duration },
    /// The per-upstream pending cap (`max_pending`, DW-015) is full: the
    /// request would have to WAIT for an outbound connection slot and the
    /// config chose immediate rejection over queueing. Classified 503
    /// ("upstream saturated") by the proxy; NOT retryable (the upstream is
    /// saturated by definition — a retry adds load).
    Saturated,
    /// Transport-level I/O failure while connecting.
    Io(std::io::Error),
    /// The hyper client failed to complete the request (broken pool
    /// connection, framing error, ...).
    Client(hyper_util::client::legacy::Error),
    /// DW-108: the upstream is configured `protocol: h3` but the H3
    /// transport is unavailable. When the `h3` cargo feature is off the
    /// H3 connector is not compiled, so an `h3` upstream is inert and
    /// every dispatch fails closed with this error rather than silently
    /// falling back to a wrong transport. Classified 502 by the proxy;
    /// not health-reportable (a configuration/feature gap, not an
    /// endpoint failure).
    H3Unavailable,
}

impl std::fmt::Display for UpstreamError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            UpstreamError::NoEndpoints => write!(f, "upstream has no endpoints"),
            UpstreamError::Saturated => {
                write!(f, "upstream pending queue is saturated")
            }
            UpstreamError::InvalidRootCertificate(e) => {
                write!(f, "unusable root certificate: {e}")
            }
            UpstreamError::InvalidHost(h) => {
                write!(f, "endpoint address '{h}' is not a valid TLS server name")
            }
            UpstreamError::ConnectTimeout { after } => {
                write!(f, "upstream connect timed out after {after:?}")
            }
            UpstreamError::ReadTimeout { after } => {
                write!(f, "upstream response headers timed out after {after:?}")
            }
            UpstreamError::Io(e) => write!(f, "upstream connect failed: {e}"),
            UpstreamError::Client(e) => write!(f, "upstream request failed: {e}"),
            UpstreamError::H3Unavailable => write!(
                f,
                "upstream protocol h3 requires the `h3` cargo feature (build with --features h3)"
            ),
        }
    }
}

impl std::error::Error for UpstreamError {}

/// Whether a send-path error reflects a genuine transport/exchange
/// failure of the picked endpoint (and therefore feeds passive health).
/// Client-side admission rejections (Saturated — the request never
/// contacted the endpoint) and configuration-class errors (NoEndpoints,
/// InvalidRootCertificate, InvalidHost) would eject the endpoint for
/// reasons unrelated to its health, so they are not reported.
fn health_reportable(err: &UpstreamError) -> bool {
    !matches!(
        err,
        UpstreamError::Saturated
            | UpstreamError::NoEndpoints
            | UpstreamError::InvalidRootCertificate(_)
            | UpstreamError::InvalidHost(_)
            | UpstreamError::H3Unavailable
    )
}

/// [`health_reportable`] over a legacy client error: our typed connector
/// errors (Saturated, InvalidHost, ...) ride in its source chain, so walk
/// it; anything else is a genuine transport failure of the exchange.
fn health_reportable_legacy(err: &hyper_util::client::legacy::Error) -> bool {
    let mut src: Option<&(dyn std::error::Error + 'static)> = Some(err);
    while let Some(s) = src {
        if let Some(u) = s.downcast_ref::<UpstreamError>() {
            return health_reportable(u);
        }
        src = s.source();
    }
    true
}

impl From<std::io::Error> for UpstreamError {
    fn from(e: std::io::Error) -> Self {
        UpstreamError::Io(e)
    }
}

impl From<hyper_util::client::legacy::Error> for UpstreamError {
    fn from(e: hyper_util::client::legacy::Error) -> Self {
        // The legacy client wraps connector errors (ours included) in its
        // Connect variant; lift our typed errors back out so callers can
        // match e.g. ConnectTimeout without digging through sources.
        let mut src: Option<&(dyn std::error::Error + 'static)> = std::error::Error::source(&e);
        while let Some(s) = src {
            if let Some(u) = s.downcast_ref::<UpstreamError>() {
                match u {
                    UpstreamError::ConnectTimeout { after } => {
                        return UpstreamError::ConnectTimeout { after: *after }
                    }
                    UpstreamError::InvalidHost(h) => return UpstreamError::InvalidHost(h.clone()),
                    UpstreamError::Saturated => return UpstreamError::Saturated,
                    _ => {}
                }
            }
            src = std::error::Error::source(s);
        }
        UpstreamError::Client(e)
    }
}

/// Error surfaced by a streamed upstream response body
/// ([`UpstreamBody`], DW-014).
#[derive(Debug)]
pub enum UpstreamBodyError {
    /// The underlying transport errored mid-stream (connection reset,
    /// framing error, ...).
    Upstream(hyper::Error),
    /// PERF-10 (#243): an H3/QUIC stream errored mid-body (h3 frame
    /// error, QPACK error, QUIC stream reset, ...). The h3 error is
    /// stringified because `h3::error::StreamError` is not `Clone` and
    /// the channel carries `Result<Bytes, UpstreamBodyError>` across
    /// an `await` boundary in the driver task.
    H3Stream(String),
    /// The gap between two body frames exceeded `timeouts.write_ms`
    /// (inactivity timeout; see the module docs).
    WriteTimeout { after: Duration },
    /// The response body crossed its ABSOLUTE deadline (DW-039: a
    /// gRPC request's `grpc-timeout` covers the whole RPC, so the
    /// deadline armed at dispatch keeps ticking through the body).
    DeadlineExceeded,
}

impl std::fmt::Display for UpstreamBodyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            UpstreamBodyError::Upstream(e) => write!(f, "upstream body failed: {e}"),
            UpstreamBodyError::H3Stream(msg) => {
                write!(f, "h3 upstream stream failed: {msg}")
            }
            UpstreamBodyError::WriteTimeout { after } => {
                write!(f, "upstream body stalled for more than {after:?}")
            }
            UpstreamBodyError::DeadlineExceeded => {
                write!(f, "response crossed its request deadline (grpc-timeout)")
            }
        }
    }
}

impl std::error::Error for UpstreamBodyError {}

/// Streaming upstream response body: the pooled `Incoming` wrapped with the
/// DW-014 response-side knobs. A thin frame passthrough when both knobs
/// are unset; otherwise:
///
/// - `idle` (from `timeouts.write_ms`): an inactivity timeout — if the gap
///   between two consecutive body frames exceeds the duration, the body
///   errors with [`UpstreamBodyError::WriteTimeout`]. Implemented by
///   arming a `tokio::time::Sleep` on every `Pending` poll and clearing it
///   on every frame; the timer is only alive while the stream is idle, so
///   an actively-trickling body never trips it (documented: it bounds
///   stalls, not total streaming time).
/// - `health`: when the dispatch carried passive health, a mid-stream
///   failure (transport error or idle timeout — either way the endpoint
///   died AFTER headers resolved) is reported as a health FAILURE to the
///   picked endpoint's tracker, closing the DW-012 gap where mid-body
///   aborts were invisible to ejection. A clean end-of-stream reports
///   nothing: the header-resolution report already classified the
///   exchange, and doubling successes would dilute failure ratios.
///
/// A channel-backed streaming body for H3 responses (PERF-10, #243).
/// Wraps a tokio mpsc receiver that receives `Result<Bytes,
/// UpstreamBodyError>` chunks from a driver task that reads h3
/// `recv_data`. Implements `hyper::body::Body` by polling the
/// receiver; the bounded channel (capacity 1) applies backpressure
/// from the consumer to the upstream.
struct H3ChannelBody {
    rx: tokio::sync::mpsc::Receiver<Result<Bytes, UpstreamBodyError>>,
}

impl hyper::body::Body for H3ChannelBody {
    type Data = Bytes;
    type Error = UpstreamBodyError;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Bytes>, UpstreamBodyError>>> {
        let this = self.get_mut();
        match this.rx.poll_recv(cx) {
            Poll::Ready(Some(Ok(bytes))) => Poll::Ready(Some(Ok(Frame::data(bytes)))),
            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

/// The error is terminal for the stream: frames already forwarded to the
/// client end abruptly (HTTP/1.1 truncation semantics); it is never
/// retried (an attempt is final once its headers resolved).
pub struct UpstreamBody {
    /// The wrapped upstream response body. A boxed body so the H3
    /// transport (DW-108) can supply a buffered body alongside the
    /// TCP/TLS path's streaming `Incoming`: the legacy path boxes an
    /// `Incoming` (mapping `hyper::Error` to [`UpstreamBodyError`]), the
    /// H3 path boxes a `Full<Bytes>` (the response is buffered; see
    /// `dataplane::upstream_h3`'s module docs). Both share the
    /// `Bytes`/`UpstreamBodyError` body shape, so the idle/deadline/
    /// health-reporting wrapper below is transport-agnostic.
    inner: http_body_util::combinators::BoxBody<Bytes, UpstreamBodyError>,
    idle: Option<Duration>,
    sleep: Option<Pin<Box<tokio::time::Sleep>>>,
    health: Option<(Arc<crate::dataplane::balance::UpstreamLb>, HealthDispatch)>,
    /// Gateway concurrency-cap permit (DW-015), attached by the proxy to
    /// STREAMING responses so the global slot is held until the body
    /// completes (or the response is dropped — client disconnect included).
    /// Dropped with the body; no polling logic needed.
    release: Option<OwnedSemaphorePermit>,
    /// ABSOLUTE deadline for the body (DW-039): a gRPC request's
    /// `grpc-timeout` covers the whole RPC, so the same deadline that
    /// bounded the forward keeps ticking here. Checked at the top of
    /// every poll; unlike `idle`, no timer is armed — the deadline is
    /// compared against the clock each poll the frame path runs, and
    /// the body's own activity drives polling (a body mid-flight
    /// polls constantly; a stalled one is bounded by the idle timer
    /// first). `None` = no deadline.
    deadline: Option<std::time::Instant>,
    /// REL-06 (#217): optional in-flight guard held until the body
    /// completes (or is dropped). When present, the in-flight counter
    /// is decremented when the body stream ends (Poll::Ready(None)) or
    /// when the body is dropped — NOT when headers resolve. This gives
    /// `least_requests`, `random`, and `peak_ewma` a more accurate
    /// view of actual endpoint load. Default behavior (when `None`)
    /// remains header-resolution release (v1).
    inflight_guard: Option<crate::dataplane::balance::InflightGuard>,
}

impl hyper::body::Body for UpstreamBody {
    type Data = Bytes;
    type Error = UpstreamBodyError;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Bytes>, UpstreamBodyError>>> {
        let this = self.get_mut();
        loop {
            // Absolute deadline (DW-039): fires whether or not the inner
            // stream is active — the RPC is over when it is over.
            if let Some(deadline) = this.deadline {
                if std::time::Instant::now() >= deadline {
                    this.report_health_failure();
                    return Poll::Ready(Some(Err(UpstreamBodyError::DeadlineExceeded)));
                }
            }
            // Idle deadline armed: check it before (and after) polling the
            // inner stream so an elapsed stall errors even when the inner
            // stream is still Pending.
            if let Some(sleep) = &mut this.sleep {
                if sleep.as_mut().poll(cx).is_ready() {
                    this.report_health_failure();
                    return Poll::Ready(Some(Err(UpstreamBodyError::WriteTimeout {
                        after: this.idle.unwrap_or_default(),
                    })));
                }
            }
            match Pin::new(&mut this.inner).poll_frame(cx) {
                Poll::Ready(Some(Ok(frame))) => {
                    // Activity: clear any armed deadline.
                    this.sleep = None;
                    return Poll::Ready(Some(Ok(frame)));
                }
                Poll::Ready(Some(Err(e))) => {
                    this.report_health_failure();
                    return Poll::Ready(Some(Err(e)));
                }
                Poll::Ready(None) => return Poll::Ready(None),
                Poll::Pending => {
                    if this.idle.is_some() && this.sleep.is_none() {
                        this.sleep =
                            Some(Box::pin(tokio::time::sleep(this.idle.unwrap_or_default())));
                        // Loop once more so the fresh timer registers its
                        // waker with the executor before we return Pending.
                        continue;
                    }
                    return Poll::Pending;
                }
            }
        }
    }

    fn size_hint(&self) -> hyper::body::SizeHint {
        self.inner.size_hint()
    }

    fn is_end_stream(&self) -> bool {
        self.inner.is_end_stream()
    }
}

impl std::fmt::Debug for UpstreamBody {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("UpstreamBody")
            .field("idle_timeout", &self.idle)
            .field("health_reporting", &self.health.is_some())
            .finish_non_exhaustive()
    }
}

impl UpstreamBody {
    /// Wrap a streaming TCP/TLS upstream body (`Incoming`) with the
    /// DW-014 idle/deadline/health knobs. `idle` is the response-body
    /// inactivity timeout (`timeouts.write_ms`); `health` carries the
    /// dispatch's passive-health tracker for mid-body failure reporting.
    /// The legacy pooled-client path uses this constructor.
    pub fn from_incoming(
        inner: Incoming,
        idle: Option<Duration>,
        health: Option<(Arc<crate::dataplane::balance::UpstreamLb>, HealthDispatch)>,
    ) -> Self {
        UpstreamBody {
            inner: inner.map_err(UpstreamBodyError::Upstream).boxed(),
            idle,
            sleep: None,
            health,
            release: None,
            deadline: None,
            inflight_guard: None,
        }
    }

    /// Wrap a fully-buffered H3 response body (DW-108). The H3 transport
    /// collects the response into one `Bytes` (streaming H3 bodies are a
    /// follow-up; see `dataplane::upstream_h3`); the wrapper still
    /// applies the deadline knob (the body is already complete, so the
    /// idle timer never arms). `health` is `None` for H3 today (a buffered
    /// body cannot mid-stream abort, so there is no mid-body failure to
    /// report).
    pub fn from_buffered(body: Bytes) -> Self {
        UpstreamBody {
            inner: http_body_util::Full::new(body)
                .map_err(|e: std::convert::Infallible| match e {})
                .boxed(),
            idle: None,
            sleep: None,
            health: None,
            release: None,
            deadline: None,
            inflight_guard: None,
        }
    }

    /// Wrap a streaming H3 response body (PERF-10, #243): a
    /// channel-backed body that forwards h3 `recv_data` chunks as they
    /// arrive, matching the h1/h2 zero-buffer guarantee. The H3
    /// transport spawns a driver task that reads `recv_data` from the
    /// h3 stream and sends each chunk through the channel; this
    /// constructor wraps the receiver as a `BoxBody` so the
    /// idle/deadline/health knobs apply uniformly across transports.
    pub fn from_h3_streaming(
        rx: tokio::sync::mpsc::Receiver<Result<Bytes, UpstreamBodyError>>,
        idle: Option<Duration>,
        health: Option<(Arc<crate::dataplane::balance::UpstreamLb>, HealthDispatch)>,
    ) -> Self {
        let body = H3ChannelBody { rx };
        UpstreamBody {
            inner: body.boxed(),
            idle,
            sleep: None,
            health,
            release: None,
            deadline: None,
            inflight_guard: None,
        }
    }

    /// Report a mid-stream failure to the dispatch's health tracker, if any.
    fn report_health_failure(&mut self) {
        if let Some((lb, hd)) = &self.health.take() {
            hd.report(lb.now_ms(), true);
        }
    }

    /// Attach the gateway concurrency-cap permit (DW-015): it releases
    /// when this body completes or is dropped, which is exactly the
    /// "release at body completion" contract of the global cap.
    pub fn set_release_permit(&mut self, permit: OwnedSemaphorePermit) {
        self.release = Some(permit);
    }

    /// Arm the ABSOLUTE body deadline (DW-039): set by the proxy from a
    /// gRPC request's `grpc-timeout` (the deadline that bounded the
    /// forward, continuing through the body — the RPC's total budget).
    pub fn set_deadline(&mut self, deadline: std::time::Instant) {
        self.deadline = Some(deadline);
    }

    /// REL-06 (#217): attach the in-flight guard so the endpoint's
    /// in-flight counter is held until the body stream completes (or
    /// the body is dropped). Used when `body_completion_inflight` is
    /// enabled in the upstream's health config.
    pub fn set_inflight_guard(&mut self, guard: crate::dataplane::balance::InflightGuard) {
        self.inflight_guard = Some(guard);
    }

    /// REL-05 (#216): create an empty body (used as a placeholder while
    /// the first-frame buffer drains the real body).
    pub fn empty() -> Self {
        UpstreamBody {
            inner: http_body_util::Empty::new()
                .map_err(|e: std::convert::Infallible| match e {})
                .boxed(),
            idle: None,
            sleep: None,
            health: None,
            release: None,
            deadline: None,
            inflight_guard: None,
        }
    }

    /// REL-05 (#216): replace the inner body with a pinned boxed body
    /// (the remaining stream after the first-frame buffer drained the
    /// prefix). The idle/deadline/health knobs are preserved.
    pub fn set_inner(
        &mut self,
        inner: http_body_util::combinators::BoxBody<Bytes, UpstreamBodyError>,
    ) {
        self.inner = inner;
    }

    /// REL-05 (#216): replace the inner body with a fully-buffered
    /// `Bytes` (the body completed within the first-frame buffer).
    pub fn set_full(&mut self, body: Bytes) {
        self.inner = http_body_util::Full::new(body)
            .map_err(|e: std::convert::Infallible| match e {})
            .boxed();
    }

    /// REL-05 (#216): create a body that emits a `prefix` then continues
    /// with `rest`. Used when the first-frame buffer filled: the prefix
    /// is flushed to the client, then the remaining upstream body
    /// streams frame-by-frame.
    pub fn with_prefix(prefix: Bytes, mut rest: UpstreamBody) -> Self {
        // We extract the inner body from `rest` and build a chained
        // body that emits the prefix first, then the rest's inner.
        // The idle/deadline/health knobs are preserved on the new body.
        let idle = rest.idle;
        let health = rest.health.take();
        let release = rest.release.take();
        let deadline = rest.deadline;
        let inflight_guard = rest.inflight_guard.take();
        let rest_inner = std::mem::replace(
            &mut rest.inner,
            http_body_util::Empty::new()
                .map_err(|e: std::convert::Infallible| match e {})
                .boxed(),
        );
        let chained = http_body_util::combinators::BoxBody::new(PrefixBody {
            prefix: Some(prefix),
            rest: rest_inner,
        });
        UpstreamBody {
            inner: chained,
            idle,
            sleep: None,
            health,
            release,
            deadline,
            inflight_guard,
        }
    }
}

/// REL-05 (#216): a body that emits a prefix `Bytes` frame first, then
/// delegates to the rest of the upstream body. Used by
/// [`UpstreamBody::with_prefix`] when the first-frame buffer filled.
struct PrefixBody {
    prefix: Option<Bytes>,
    rest: http_body_util::combinators::BoxBody<Bytes, UpstreamBodyError>,
}

impl hyper::body::Body for PrefixBody {
    type Data = Bytes;
    type Error = UpstreamBodyError;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Bytes>, UpstreamBodyError>>> {
        if let Some(prefix) = self.prefix.take() {
            return Poll::Ready(Some(Ok(Frame::data(prefix))));
        }
        Pin::new(&mut self.rest).poll_frame(cx)
    }
}

/// Live connection counters for one upstream; observability for pooling
/// behavior (used by tests and later by the admin surface).
#[derive(Debug, Default)]
pub struct UpstreamStats {
    /// Total TCP/TLS connections established to this upstream since the
    /// handle was built. Stays flat when the pool reuses connections.
    pub connections_opened: AtomicU64,
    /// Total requests handed to the pool.
    pub requests_sent: AtomicU64,
}

/// The transport under a pooled connection: plaintext TCP or TLS, wrapped
/// in `TokioIo` to bridge tokio traits to hyper's runtime traits.
enum Transport {
    Plain(TokioIo<TcpStream>),
    Tls(Box<TokioIo<tokio_rustls::client::TlsStream<TcpStream>>>),
}

impl hyper::rt::Read for Transport {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: hyper::rt::ReadBufCursor<'_>,
    ) -> Poll<std::io::Result<()>> {
        match &mut *self {
            Transport::Plain(s) => Pin::new(s).poll_read(cx, buf),
            Transport::Tls(s) => Pin::new(s.as_mut()).poll_read(cx, buf),
        }
    }
}

impl hyper::rt::Write for Transport {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        match &mut *self {
            Transport::Plain(s) => Pin::new(s).poll_write(cx, buf),
            Transport::Tls(s) => Pin::new(s.as_mut()).poll_write(cx, buf),
        }
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        match &mut *self {
            Transport::Plain(s) => Pin::new(s).poll_flush(cx),
            Transport::Tls(s) => Pin::new(s.as_mut()).poll_flush(cx),
        }
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        match &mut *self {
            Transport::Plain(s) => Pin::new(s).poll_shutdown(cx),
            Transport::Tls(s) => Pin::new(s.as_mut()).poll_shutdown(cx),
        }
    }
}

/// Pooled connection IO: the transport plus the connection-cap permit.
/// The permit lives exactly as long as the connection: when the pool drops
/// a closed/evicted connection, the permit is released and the next queued
/// dial may proceed.
struct CappedStream {
    transport: Transport,
    _permit: OwnedSemaphorePermit,
}

impl hyper::rt::Read for CappedStream {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: hyper::rt::ReadBufCursor<'_>,
    ) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.transport).poll_read(cx, buf)
    }
}

impl hyper::rt::Write for CappedStream {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        Pin::new(&mut self.transport).poll_write(cx, buf)
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.transport).poll_flush(cx)
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.transport).poll_shutdown(cx)
    }
}

impl HyperConnection for CappedStream {
    fn connected(&self) -> Connected {
        Connected::new()
    }
}

/// Default happy-eyeballs inter-connection delay when
/// `timeouts.happy_eyeballs_ms` is absent (the RFC 8305 recommended
/// value; DW-030).
pub const DEFAULT_HAPPY_EYEBALLS_MS: u64 = 250;

/// `timeouts.happy_eyeballs_ms`, resolved to the effective racing delay:
/// absent -> the RFC 8305 default; `0` -> None (racing disabled, strict
/// resolver order); otherwise the configured value (clamped to the
/// ceiling in `config::limits`; validation has already rejected
/// over-cap values).
fn effective_happy_eyeballs(u: &Upstream) -> Option<Duration> {
    match u.timeouts.as_ref().and_then(|t| t.happy_eyeballs_ms) {
        None => Some(Duration::from_millis(DEFAULT_HAPPY_EYEBALLS_MS)),
        Some(0) => None,
        Some(ms) => Some(Duration::from_millis(
            ms.min(crate::config::limits::MAX_HAPPY_EYEBALLS_MS),
        )),
    }
}

/// RFC 8305 address ordering (DW-030): the resolver's FIRST address keeps
/// its place and defines the preferred family; the remainder follow with
/// the OTHER family second, alternating onward — so the gateway reaches
/// the second family after one inter-connection delay no matter how long
/// the preferred family's list is.
///
/// `#[doc(hidden)]` public for the unit tests in `tests/unit/upstream.rs`
/// (pure ordering function; exercising it through real dials would need
/// controllable DNS, which the test environment cannot provide).
#[doc(hidden)]
pub fn interleave_order(addrs: &[SocketAddr]) -> Vec<SocketAddr> {
    if addrs.len() < 2 {
        return addrs.to_vec();
    }
    let preferred_v6 = addrs[0].is_ipv6();
    let mut preferred: VecDeque<SocketAddr> = VecDeque::new();
    let mut other: VecDeque<SocketAddr> = VecDeque::new();
    for a in addrs {
        if a.is_ipv6() == preferred_v6 {
            preferred.push_back(*a);
        } else {
            other.push_back(*a);
        }
    }
    let mut out = Vec::with_capacity(addrs.len());
    while let (Some(a), maybe_b) = (preferred.pop_front(), other.front()) {
        out.push(a);
        if let Some(b) = maybe_b {
            out.push(*b);
            other.pop_front();
        }
    }
    out
}

/// Race connection attempts across addresses, RFC 8305 shape (DW-030).
///
/// Starts an attempt on `seq`'s first address immediately; each further
/// attempt starts `delay` after the previous START — or immediately when
/// an attempt FAILS and nothing else is in flight (RFC 8305 5.2's
/// failure fast-forward). The first success wins and cancelling the
/// losers is dropping the [`JoinSet`]; an all-failed race surfaces the
/// LAST error. Generic over the dial so the unit tests can drive it with
/// controllable futures (real dials cannot make "the first address hangs"
/// deterministic on loopback); `#[doc(hidden)]` for that test seam.
#[doc(hidden)]
pub async fn happy_race<F, Fut, T>(
    seq: Vec<SocketAddr>,
    delay: Option<Duration>,
    dial: F,
) -> std::io::Result<T>
where
    F: Fn(SocketAddr) -> Fut,
    Fut: Future<Output = std::io::Result<T>> + Send + 'static,
    T: Send + 'static,
{
    // Racing disabled (or a single address): strict order, one at a
    // time — the pre-DW-030 dial semantics.
    let Some(delay) = delay.filter(|d| !d.is_zero()) else {
        let mut last = None;
        for a in seq {
            match dial(a).await {
                Ok(t) => return Ok(t),
                Err(e) => last = Some(e),
            }
        }
        return Err(last.unwrap_or_else(no_addresses));
    };
    let mut set: JoinSet<std::io::Result<T>> = JoinSet::new();
    let mut it = seq.into_iter().peekable();
    let mut last_err: Option<std::io::Error> = None;
    let Some(first) = it.next() else {
        return Err(no_addresses());
    };
    set.spawn(dial(first));
    let next_start = tokio::time::sleep(delay);
    tokio::pin!(next_start);
    loop {
        tokio::select! {
            joined = set.join_next() => {
                match joined.expect("join_next polled on a non-empty JoinSet") {
                    // First success wins; returning drops `set`, which
                    // cancels every losing arm mid-connect.
                    Ok(Ok(t)) => return Ok(t),
                    Ok(Err(e)) => last_err = Some(e),
                    // A panicking dial task is a bug, not an address
                    // being unreachable — surface it loudly.
                    Err(join_err) => {
                        return Err(std::io::Error::other(format!("dial task failed: {join_err}")))
                    }
                }
            }
            _ = &mut next_start, if it.peek().is_some() => {
                let a = it.next().expect("peek guaranteed an address");
                set.spawn(dial(a));
                next_start.as_mut().reset(tokio::time::Instant::now() + delay);
            }
        }
        // RFC 8305 failure fast-forward: with nothing in flight, the
        // next attempt starts NOW instead of waiting out the delay (the
        // peer already refused; a delay would only add latency).
        if set.is_empty() {
            match it.next() {
                Some(a) => {
                    set.spawn(dial(a));
                    next_start
                        .as_mut()
                        .reset(tokio::time::Instant::now() + delay);
                }
                None => {
                    return Err(last_err.take().unwrap_or_else(no_addresses));
                }
            }
        }
    }
}

/// The empty-address-list error (both racing modes share it).
fn no_addresses() -> std::io::Error {
    std::io::Error::new(std::io::ErrorKind::InvalidInput, "no addresses to dial")
}

/// Resolve one endpoint authority and dial it with the RFC 8305 racing
/// (DW-030). IP-literal authorities skip DNS entirely (the resolver's
/// short-circuit, the same path `getaddrinfo` takes); a multi-address
/// resolution races per `delay`. One dial, one outcome: the caller's
/// health/breaker accounting sees exactly this Result, never the losing
/// arms' failures. Shared by the pooled connector and the active health
/// probes (one dialing discipline per upstream).
pub async fn happy_dial(
    host: &str,
    port: u16,
    delay: Option<Duration>,
) -> std::io::Result<TcpStream> {
    let addrs: Vec<SocketAddr> = tokio::net::lookup_host((host, port)).await?.collect();
    if addrs.is_empty() {
        return Err(std::io::Error::new(
            std::io::ErrorKind::AddrNotAvailable,
            format!("'{host}:{port}' resolved to no addresses"),
        ));
    }
    let seq = interleave_order(&addrs);
    happy_race(seq, delay, |addr| async move {
        let stream = TcpStream::connect(addr).await?;
        // Preserve the pre-DW-030 hyper-util connector behavior: NODELAY
        // on the proxy hop (latency over throughput on stream request/
        // response exchanges).
        let _ = stream.set_nodelay(true);
        Ok(stream)
    })
    .await
}

/// The cached variant of [`happy_dial`] for the pooled connector (PERF-05,
/// #170). Resolution runs through the shared [`DnsCache`] (async hickory
/// with TTL + negative caching) instead of the blocking-pool
/// `getaddrinfo` of `tokio::net::lookup_host`; the RFC 8305 racing and
/// NODELAY behavior are identical. IP-literal hosts skip DNS via the
/// cache's short-circuit. The active health probes keep using
/// [`happy_dial`] (system resolver) — they are a background, infrequent
/// path and do not share the pooled connector's cache.
pub async fn happy_dial_via(
    dns: &DnsCache,
    host: &str,
    port: u16,
    delay: Option<Duration>,
) -> std::io::Result<TcpStream> {
    let addrs: Vec<SocketAddr> = dns.resolve(host, port).await?;
    if addrs.is_empty() {
        return Err(std::io::Error::new(
            std::io::ErrorKind::AddrNotAvailable,
            format!("'{host}:{port}' resolved to no addresses"),
        ));
    }
    let seq = interleave_order(&addrs);
    happy_race(seq, delay, |addr| async move {
        let stream = TcpStream::connect(addr).await?;
        let _ = stream.set_nodelay(true);
        Ok(stream)
    })
    .await
}

/// Per-upstream connector: RFC 8305 resolve + dial (DW-030), optional
/// rustls TLS with baked-in ALPN, the connect timeout, and the
/// connection-cap semaphore.
///
/// Cloned per pool shard (all fields are `Arc`-shared or `Copy`: the
/// connection-cap semaphore, DNS cache, TLS config, and stats are
/// shared across shards so the total connection cap and observability
/// are preserved).
#[derive(Clone)]
struct UpstreamConnector {
    /// Happy-eyeballs inter-connection delay; None = racing disabled
    /// (`timeouts.happy_eyeballs_ms: 0`, strict resolver order).
    happy_eyeballs: Option<Duration>,
    /// TLS client config (ALPN already set) for https/http2 upstreams.
    tls: Option<Arc<rustls::ClientConfig>>,
    cap: Arc<Semaphore>,
    /// Pending-request cap (`max_pending`, DW-015). Absent = bounded queue
    /// of 256 (the #270 default). Some = at most this many requests may
    /// WAIT for a connection-cap slot; further dials fail fast with
    /// [`UpstreamError::Saturated`].
    pending_cap: Option<Arc<Semaphore>>,
    connect_timeout: Duration,
    stats: Arc<UpstreamStats>,
    /// Shared async DNS cache (PERF-05, #170). One cache is shared
    /// across every upstream in a registry so repeated dials to the same
    /// host hit the TTL/negative cache. Replaces the blocking-pool
    /// `getaddrinfo` of `tokio::net::lookup_host` on the pooled dial
    /// path.
    dns: Arc<DnsCache>,
}

impl Service<Uri> for UpstreamConnector {
    type Response = CappedStream;
    type Error = UpstreamError;
    type Future =
        Pin<Box<dyn Future<Output = Result<CappedStream, UpstreamError>> + Send + 'static>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // Resolution happens inside `call` via the async hickory cache
        // (PERF-05); no readiness gating needed.
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, uri: Uri) -> Self::Future {
        let happy_eyeballs = self.happy_eyeballs;
        let tls = self.tls.clone();
        let cap = Arc::clone(&self.cap);
        let pending_cap = self.pending_cap.clone();
        let connect_timeout = self.connect_timeout;
        let stats = Arc::clone(&self.stats);
        let dns = Arc::clone(&self.dns);
        // Pending admission (DW-015) happens OUTSIDE the async block so a
        // saturated upstream rejects immediately: a request that would
        // have to queue behind more than `max_pending` waiters never even
        // arms its dial. The pending permit is held only while the request
        // is WAITING for a connection-cap slot and is dropped the moment
        // the cap permit is acquired (the request is then connecting, no
        // longer pending).
        let _pending = match pending_cap {
            Some(pc) => match pc.try_acquire_owned() {
                Ok(permit) => Some(permit),
                Err(_) => {
                    return Box::pin(async { Err(UpstreamError::Saturated) });
                }
            },
            None => None,
        };
        Box::pin(async move {
            // Acquire a cap slot BEFORE dialing. The semaphore is never
            // closed, so acquire cannot fail; waiting here is the cap's
            // documented backpressure (queue, never fail).
            let permit = cap
                .acquire_owned()
                .await
                .expect("connection-cap semaphore is never closed");
            // Connection slot acquired: no longer pending. Dropping the
            // permit here (before the dial) frees the pending slot for
            // the next request.
            drop(_pending);
            // `Uri::host()` keeps IPv6 brackets ("[::1]"); the resolver
            // and rustls `ServerName` want the bare address ("::1").
            // hyper-util's GaiResolver stripped these internally — the
            // DW-030 dial is ours, so the strip is ours too.
            let uri_host = uri.host().unwrap_or_default();
            let host = uri_host
                .strip_prefix('[')
                .and_then(|h| h.strip_suffix(']'))
                .unwrap_or(uri_host)
                .to_string();
            // The authority always carries the configured endpoint port
            // (endpoint_authority); the defaults are the inert fallback
            // for a hand-built URI.
            let port = uri
                .port_u16()
                .unwrap_or(if tls.is_some() { 443 } else { 80 });
            let dial = async {
                let tcp = happy_dial_via(&dns, &host, port, happy_eyeballs)
                    .await
                    .map_err(UpstreamError::Io)?;
                let transport = match tls {
                    Some(config) => {
                        let name = ServerName::try_from(host.clone())
                            .map_err(|_| UpstreamError::InvalidHost(host.clone()))?;
                        let connector = tokio_rustls::TlsConnector::from(config);
                        let tls_stream = connector
                            .connect(name, tcp)
                            .await
                            .map_err(|e| std::io::Error::other(e.to_string()))?;
                        Transport::Tls(Box::new(TokioIo::new(tls_stream)))
                    }
                    None => Transport::Plain(TokioIo::new(tcp)),
                };
                Ok::<Transport, UpstreamError>(transport)
            };
            let transport = tokio::time::timeout(connect_timeout, dial)
                .await
                .map_err(|_| UpstreamError::ConnectTimeout {
                    after: connect_timeout,
                })??;
            stats.connections_opened.fetch_add(1, Ordering::Relaxed);
            Ok(CappedStream {
                transport,
                _permit: permit,
            })
        })
    }
}

/// A send-capable handle to one configured upstream: a dedicated pooled
/// hyper client plus observability. Cheap to share via `Arc` from the
/// [`UpstreamRegistry`].
pub struct UpstreamHandle {
    name: String,
    cap: u32,
    connect_timeout: Duration,
    /// Per-attempt deadline over pool wait + connect + request write +
    /// response headers (`timeouts.read_ms`; DW-014). None = unbounded.
    read_timeout: Option<Duration>,
    /// Response-body inactivity timeout (`timeouts.write_ms`; DW-014).
    /// None = unbounded.
    write_timeout: Option<Duration>,
    /// Resolved retry parameters (`upstreams[].retries`; DW-014).
    /// `attempts == 0` means retries off.
    retries: RetryParams,
    /// Resolved hedge parameters (DW-063); extracted from `retries.hedge`
    /// for convenience. `hedge_after == ZERO` means hedging is disabled.
    hedge: HedgeParams,
    /// Rolling-window retry budget, carried across reloads like the
    /// balancer.
    retry_budget: Arc<RetryBudget>,
    /// Effective pending cap (`max_pending`, DW-015); defaults to 256
    /// when absent (#270). 0 = explicitly unbounded (legacy behavior).
    max_pending: u32,
    /// Per-upstream circuit breaker state (DW-015), carried across reloads
    /// like the retry budget. Always present (state-only object); whether
    /// the breaker is ENABLED is `breaker_params`.
    breaker: Arc<Breaker>,
    /// Resolved breaker parameters; None disables the breaker (the proxy
    /// then never consults it — behavior identical to pre-DW-015).
    breaker_params: Option<BreakerParams>,
    /// Resolved happy-eyeballs delay (DW-030); None = racing disabled.
    /// Kept on the handle so the active health probes dial with the same
    /// discipline as the pooled connector.
    happy_eyeballs: Option<Duration>,
    stats: Arc<UpstreamStats>,
    /// Sharded hyper-util client pool. hyper-util's connection pool uses a
    /// single `std::sync::Mutex<PoolInner>` for ALL operations (checkout +
    /// return), which becomes the serialization bottleneck under high
    /// concurrency (profiled at ~21% of CPU in `__psynch_mutexwait` at
    /// 80k RPS with 256 connections). Sharding into N independent clients
    /// gives N independent pool mutexes, reducing contention N-fold.
    ///
    /// Requests are distributed round-robin via `shard_counter` (a relaxed
    /// `AtomicUsize`). The connection-cap semaphore lives on the shared
    /// `UpstreamConnector`, so the TOTAL connection cap still applies
    /// across all shards (the cap is not multiplied by the shard count).
    ///
    /// Shard count is `DWARA_POOL_SHARDS` (default: available parallelism,
    /// capped at 16). 1 shard = the previous single-pool behavior.
    clients: Vec<UpstreamClient>,
    /// Round-robin shard selector (relaxed: a slight skew under extreme
    /// contention is harmless — the goal is distribution, not exactness).
    shard_counter: std::sync::atomic::AtomicUsize,
    /// Load balancer over this upstream's endpoint set (DW-011); picks
    /// the endpoint per dispatch and tracks in-flight counts.
    lb: Arc<crate::dataplane::balance::UpstreamLb>,
    scheme: &'static str,
    http2_only: bool,
    /// The trust roots this upstream's TLS connections verify against
    /// (#121): None for plaintext `http1` upstreams, the configured
    /// `trusted_ca_file` bundle when set, else the webpki public set.
    /// Kept on the handle so the active https health probes use the
    /// SAME trust as the pooled connector.
    tls_roots: Option<rustls::RootCertStore>,
    /// DW-108: the configured protocol. `H3` selects the QUIC transport
    /// (`h3` field) when the `h3` cargo feature is on, and fails closed
    /// (`UpstreamError::H3Unavailable`) when the feature is off (an `h3`
    /// upstream is accepted at validation but inert). The other variants
    /// use the TCP/TLS `client` below.
    protocol: UpstreamProtocol,
    /// DP-07 (#236): what to hash for consistent-hash load balancers
    /// (`ip_hash` and `maglev`). `None` = client IP (the default).
    hash_on: Option<crate::config::HashOn>,
    /// DW-108: the H3/QUIC upstream transport, present iff
    /// `protocol == H3` AND the `h3` cargo feature is enabled. When
    /// `protocol == H3` and this is `None`, the feature is off and every
    /// dispatch fails closed with [`UpstreamError::H3Unavailable`].
    h3: Option<Arc<crate::dataplane::upstream_h3::H3UpstreamHandle>>,
    /// SCALE-10 (#189): number of connections to pre-establish per
    /// endpoint on startup/reload. 0 = no pre-warming.
    pre_warm: u32,
    /// REL-06 (#217): whether the in-flight counter is held until the
    /// response body completes (true) or released at header resolution
    /// (false, the v1 default).
    body_completion_inflight: bool,
}

impl UpstreamHandle {
    /// Upstream name this handle serves.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Effective connection cap (explicit or default, minimum 1).
    pub fn cap(&self) -> u32 {
        self.cap
    }

    /// Effective connect timeout (explicit or default).
    pub fn connect_timeout(&self) -> Duration {
        self.connect_timeout
    }

    /// Effective happy-eyeballs delay (DW-030); None = racing disabled.
    pub fn happy_eyeballs(&self) -> Option<Duration> {
        self.happy_eyeballs
    }

    /// Per-attempt read timeout (`timeouts.read_ms`), if configured.
    pub fn read_timeout(&self) -> Option<Duration> {
        self.read_timeout
    }

    /// Response-body inactivity timeout (`timeouts.write_ms`), if
    /// configured.
    pub fn write_timeout(&self) -> Option<Duration> {
        self.write_timeout
    }

    /// Resolved retry parameters (DW-014). `attempts == 0` = retries off.
    pub fn retry_params(&self) -> &RetryParams {
        &self.retries
    }

    /// Resolved hedge parameters (DW-063).
    pub fn hedge_params(&self) -> &HedgeParams {
        &self.hedge
    }

    /// This upstream's rolling-window retry budget (DW-014).
    pub fn retry_budget(&self) -> &Arc<RetryBudget> {
        &self.retry_budget
    }

    /// Effective pending cap (DW-015); 0 = unbounded queueing.
    pub fn max_pending(&self) -> u32 {
        self.max_pending
    }

    /// This upstream's circuit breaker state (DW-015). Always present;
    /// consult [`UpstreamHandle::breaker_params`] first — a None there
    /// means the breaker is disabled and the state object is dormant.
    pub fn breaker(&self) -> &Arc<Breaker> {
        &self.breaker
    }

    /// Resolved breaker parameters (DW-015); None = breaker disabled.
    pub fn breaker_params(&self) -> Option<&BreakerParams> {
        self.breaker_params.as_ref()
    }

    /// Total connections established since the handle was built.
    pub fn connections_opened(&self) -> u64 {
        self.stats.connections_opened.load(Ordering::Relaxed)
    }

    /// Total requests sent through the pool.
    pub fn requests_sent(&self) -> u64 {
        self.stats.requests_sent.load(Ordering::Relaxed)
    }

    /// Effective scheme this upstream dials ("http" or "https").
    pub fn scheme(&self) -> &str {
        self.scheme
    }

    /// This upstream's load balancer (endpoint set, algorithm, in-flight
    /// counters). Exposed for the TLS-passthrough path (which picks an
    /// endpoint the same way) and for tests.
    pub fn lb(&self) -> &Arc<crate::dataplane::balance::UpstreamLb> {
        &self.lb
    }

    /// DP-07 (#236): the hash-key source for this upstream's
    /// consistent-hash load balancer (`ip_hash` or `maglev`). `None`
    /// means the client IP is used (the default).
    pub fn hash_on(&self) -> Option<&crate::config::HashOn> {
        self.hash_on.as_ref()
    }

    /// DW-094 (Ent): set the edge's locality context on this upstream's
    /// load balancer. Called by the dataplane at startup to propagate
    /// the edge's region/zone (from env vars or edge labels) to every
    /// upstream's balancer.
    pub fn set_locality_context(&self, ctx: crate::config::LocalityContext) {
        self.lb.set_locality_context(ctx);
    }

    /// The trust roots this upstream's TLS connections verify against
    /// (#121): None for plaintext `http1` upstreams, the configured
    /// `trusted_ca_file` bundle when set, else the webpki public set.
    /// Active https health probes clone this so a probe and a proxied
    /// request can never disagree about who to trust.
    pub fn tls_roots(&self) -> Option<&rustls::RootCertStore> {
        self.tls_roots.as_ref()
    }

    /// DW-108: the H3/QUIC upstream transport, when this upstream is
    /// `protocol: h3` AND the `h3` cargo feature is enabled. The QUIC
    /// active health probe reuses its TLS client config (trust roots +
    /// `h3` ALPN) so a probe and a proxied request trust the same roots.
    /// None for non-H3 upstreams or when the feature is off.
    pub fn h3_handle(&self) -> Option<&Arc<crate::dataplane::upstream_h3::H3UpstreamHandle>> {
        self.h3.as_ref()
    }

    /// The configured upstream protocol (DW-108). `H3` selects the QUIC
    /// transport; the other variants use the TCP/TLS pooled client.
    pub fn protocol(&self) -> UpstreamProtocol {
        self.protocol
    }

    /// Send a request through this upstream's pool without a hash key
    /// (algorithms other than `ip_hash` ignore the key anyway).
    pub async fn send<B>(&self, req: Request<B>) -> Result<Response<UpstreamBody>, UpstreamError>
    where
        B: hyper::body::Body<Data = bytes::Bytes> + Send + 'static,
        B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
    {
        self.send_with_hash_key(req, None).await
    }

    /// Send a request through this upstream's pool, dispatching via the
    /// upstream's load balancer (DW-011). `hash_key` is the client IP as a
    /// string (used by `ip_hash`; ignored by the other algorithms). The
    /// picked endpoint's `address:port` becomes both the dialed URI's
    /// authority and the outbound `Host` header (replacing whatever the
    /// caller set); the path and query are preserved verbatim. The picked
    /// endpoint's in-flight counter is held for the request/response-header
    /// exchange (documented approximation: released when headers resolve,
    /// not when the streaming body completes). The response body streams
    /// (wrapped in [`UpstreamBody`] for the DW-014 write-timeout /
    /// mid-body health reporting knobs; the wrapper is a frame-for-frame
    /// passthrough when both are unset), so proxying (DW-009) can forward
    /// it without buffering.
    pub async fn send_with_hash_key<B>(
        &self,
        req: Request<B>,
        hash_key: Option<&str>,
    ) -> Result<Response<UpstreamBody>, UpstreamError>
    where
        B: hyper::body::Body<Data = bytes::Bytes> + Send + 'static,
        B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
    {
        let mut picked = None;
        self.send_inner(req, hash_key, &mut picked)
            .await
            .map(|(resp, ())| resp)
    }

    /// [`UpstreamHandle::send_with_hash_key`] with observability
    /// (DW-021): the load-balancer pick runs under an `upstream_pick`
    /// span (child of the caller's `upstream_attempt` span) and the
    /// picked endpoint's `address:port` is written to `picked` when a
    /// dispatch resolved (left None on the NoEndpoints guard), so the
    /// proxy can attribute the attempt in its access log and the
    /// `upstream_attempts_total` metric.
    pub async fn send_with_hash_key_observed<B>(
        &self,
        req: Request<B>,
        hash_key: Option<&str>,
        picked: &mut Option<String>,
    ) -> Result<Response<UpstreamBody>, UpstreamError>
    where
        B: hyper::body::Body<Data = bytes::Bytes> + Send + 'static,
        B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
    {
        self.send_inner(req, hash_key, picked)
            .await
            .map(|(resp, ())| resp)
    }

    async fn send_inner<B>(
        &self,
        mut req: Request<B>,
        hash_key: Option<&str>,
        picked: &mut Option<String>,
    ) -> Result<(Response<UpstreamBody>, ()), UpstreamError>
    where
        B: hyper::body::Body<Data = bytes::Bytes> + Send + 'static,
        B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
    {
        // PERF: clone the PathAndQuery directly from the request URI
        // instead of allocating a String and re-parsing it. PathAndQuery
        // is internally an Arc<str>-like type, so clone is a refcount
        // bump. The String form is derived via as_str() only where needed
        // (the H3 branch's format!()).
        let path = req
            .uri()
            .path_and_query()
            .cloned()
            .unwrap_or_else(|| PathAndQuery::from_static("/"));
        // DW-157: a plugin's `:authority` rewrite, staged by the proxy's
        // forward build as the Host value the forwarded request should
        // carry. The dial target and the URI authority stay the picked
        // endpoint's (a plugin shapes the Host header, never which
        // origin the gateway dials); applied per dispatch below, pooled
        // and H3 alike.
        let plugin_host = req
            .extensions()
            .get::<crate::dataplane::plugin_dispatch::PluginForwardHost>()
            .map(|host| host.0.clone());
        // Guard rather than dialing a fabricated address: empty endpoint
        // lists are only possible via unvalidated construction. The pick,
        // endpoint resolution, and in-flight acquisition all run against
        // ONE state snapshot (pick_for_dispatch), so a concurrent reload
        // cannot detach the guard from the picked endpoint.
        let mut dispatch = {
            // DW-021: the pick phase is its own span so a full trace
            // shows pick separately from the attempt that contains it.
            let span = tracing::info_span!(
                "upstream_pick",
                upstream = self.name,
                endpoint = tracing::field::Empty
            );
            let _guard = span.enter();
            let dispatch = self
                .lb
                .pick_for_dispatch(hash_key)
                .ok_or(UpstreamError::NoEndpoints)?;
            // PERF: the authority (and its parsed forms) are precomputed
            // per endpoint at LB-state build — no per-attempt format or
            // parse. One owned String is retained for the caller's
            // picked-endpoint label.
            span.record("endpoint", &*dispatch.authority);
            *picked = Some(dispatch.authority.to_string());
            dispatch
        };
        // DW-108: H3 upstreams dispatch over QUIC, not the TCP/TLS pooled
        // client. The LB pick, in-flight guard, health reporting, and
        // latency recording are SHARED with the legacy path (they operate
        // on endpoint addresses, not the transport); only the dial +
        // request/response exchange differs. When the `h3` cargo feature
        // is off (or the H3 handle failed to build), the upstream is
        // inert and every dispatch fails closed with
        // [`UpstreamError::H3Unavailable`] — never a silent fallback to a
        // wrong transport.
        if matches!(self.protocol, UpstreamProtocol::H3) {
            let issued = std::time::Instant::now();
            let outcome: Result<hyper::Response<Bytes>, UpstreamError> = {
                {
                    if let Some(h3) = &self.h3 {
                        // Buffer the request body (the H3 path sends it as
                        // one DATA frame; see `dataplane::upstream_h3`).
                        let method = req.method().clone();
                        let mut headers = req.headers().clone();
                        // The plugin's `:authority` rewrite applies to the
                        // Host header here too (the H3 request's own
                        // authority pseudo-header is the dialed endpoint,
                        // per the note above).
                        if let Some(host) = plugin_host.clone() {
                            headers.insert(hyper::header::HOST, host);
                        }
                        let body_bytes = match req.into_body().collect().await {
                            Ok(collected) => collected.to_bytes(),
                            Err(e) => {
                                let err = UpstreamError::Io(std::io::Error::other(e));
                                if let Some(health) = &dispatch.health {
                                    health.report(self.lb.now_ms(), true);
                                }
                                self.lb.record_latency(dispatch.idx, issued.elapsed());
                                dispatch.release();
                                return Err(err);
                            }
                        };
                        let mut h3_req = hyper::Request::builder().method(method).uri(format!(
                            "https://{}{}",
                            dispatch.authority,
                            path.as_str()
                        ));
                        if let Some(h) = h3_req.headers_mut() {
                            *h = headers;
                        }
                        // H3 speaks HTTP/3; stamp the version so the
                        // upstream sees its own dialect.
                        h3_req = h3_req.version(hyper::Version::HTTP_3);
                        let h3_req = h3_req.body(body_bytes).expect("static builder");
                        h3.send(&dispatch.address, dispatch.port, &dispatch.address, h3_req)
                            .await
                    } else {
                        Err(UpstreamError::H3Unavailable)
                    }
                }
            };
            // Same health classification as the legacy path: transport
            // errors and statuses >= 5xx are failures; admission/config
            // errors (H3Unavailable) are not health-reportable.
            let report = match &outcome {
                Ok(resp) => Some(resp.status().as_u16() >= 500),
                Err(err) if health_reportable(err) => Some(true),
                Err(_) => None,
            };
            if let (Some(health), Some(is_failure)) = (&dispatch.health, report) {
                health.report(self.lb.now_ms(), is_failure);
            }
            self.lb.record_latency(dispatch.idx, issued.elapsed());
            dispatch.release();
            // The H3 path returns a buffered body (UpstreamBody::from_buffered);
            // the non-H3 path returns H3Unavailable (an Err), so the map
            // only runs when the feature is on and the request succeeded.
            return outcome.map(|resp| (resp.map(UpstreamBody::from_buffered), ()));
        }
        // Held (inside `dispatch`) until the response (headers) resolves;
        // see the doc comment.
        //
        // PERF: construct the Uri from parts instead of format!() + parse.
        // The previous code allocated a String for the full URI
        // ("scheme://authority/path") and then parsed it; from_parts
        // reuses the precomputed authority (a refcount clone from the
        // LB endpoint row — no per-attempt parse) and the cloned
        // PathAndQuery (an Arc-bump clone from the inbound URI), avoiding
        // the intermediate full-URI String allocation and the full-URI
        // parse pass.
        let uri = {
            let scheme = Scheme::try_from(self.scheme)
                .map_err(|e| UpstreamError::Io(std::io::Error::other(e.to_string())))?;
            // The precomputed parse fails only for a mis-configured
            // endpoint address that validation already rejects; fall back
            // to parsing the authority string (the pre-optimization
            // behavior) rather than inventing a new failure mode.
            let auth = match dispatch.http_authority.clone() {
                Some(auth) => auth,
                None => Authority::from_str(&dispatch.authority)
                    .map_err(|e| UpstreamError::Io(std::io::Error::other(e.to_string())))?,
            };
            let mut parts = http::uri::Parts::default();
            parts.scheme = Some(scheme);
            parts.authority = Some(auth);
            parts.path_and_query = Some(path);
            Uri::from_parts(parts)
                .map_err(|e| UpstreamError::Io(std::io::Error::other(e.to_string())))
        }?;
        *req.uri_mut() = uri;
        // The gateway, not the client, names the origin it dials: the
        // picked endpoint's authority replaces any Host the caller set —
        // EXCEPT a plugin's `:authority` rewrite (DW-157), which names
        // the Host header the upstream should see while the dial target
        // stays the balancer's pick. PERF: the HeaderValue is
        // precomputed with the authority; the fallback (and the
        // skip-on-failure shape) mirrors the Uri note above —
        // unreachable for validated configs.
        let host = plugin_host.or_else(|| {
            dispatch
                .host_value
                .clone()
                .or_else(|| hyper::header::HeaderValue::from_str(&dispatch.authority).ok())
        });
        if let Some(v) = host {
            req.headers_mut().insert(hyper::header::HOST, v);
        }
        // Normalize the HTTP version for the pool's protocol: an inbound h2
        // request proxied to an h1 upstream must be downgraded to 1.1 (and
        // vice versa); the pooled client speaks exactly one dialect.
        *req.version_mut() = if self.http2_only {
            Version::HTTP_2
        } else {
            Version::HTTP_11
        };
        self.stats.requests_sent.fetch_add(1, Ordering::Relaxed);
        let req =
            req.map(|b| http_body_util::combinators::UnsyncBoxBody::new(b.map_err(Into::into)));
        // Observation wire (DW-012): the outcome is reported to the picked
        // endpoint's health tracker when the response headers resolve —
        // the same point the in-flight guard releases. Classification:
        // transport errors (the Err arm: connect timeout, refusal, reset,
        // framing, and the DW-014 read timeout) and statuses >= 500 are
        // failures; 1xx-4xx (including 429/408; documented choice) are
        // successes. Mid-BODY aborts are reported later, by `UpstreamBody`.
        // DW-090: the issued instant is captured here so the peak-EWMA
        // latency tracker records the FULL attempt duration (pool-queue
        // wait + connect + request write + response headers) for every
        // outcome — success, error, and timeout alike (a timeout is a
        // latency signal too).
        let issued = std::time::Instant::now();
        // Round-robin shard selection: fetch_add gives each request the
        // next shard index. Relaxed ordering is sufficient — the goal is
        // distribution, not a strict ordering guarantee.
        let shard_idx = self.shard_counter.fetch_add(1, Ordering::Relaxed) % self.clients.len();
        let request = self.clients[shard_idx].request(req);
        // Captured before release(): the body wrapper reports mid-stream
        // failures into the same tracker (DW-014 closing the DW-012 gap).
        let body_health = dispatch.health.clone().map(|hd| (Arc::clone(&self.lb), hd));
        let outcome = match self.read_timeout {
            // Per-attempt deadline (DW-014): pool-queue wait + connect +
            // request write + response headers, one bound. Dropping the
            // request future cancels the attempt (the pooled connection,
            // if one was involved, is discarded by the pool).
            Some(after) => match tokio::time::timeout(after, request).await {
                Ok(outcome) => outcome,
                Err(_elapsed) => {
                    if let Some(health) = &dispatch.health {
                        health.report(self.lb.now_ms(), true);
                    }
                    // DW-090: record the latency (the full timeout
                    // duration) before releasing the in-flight guard.
                    self.lb.record_latency(dispatch.idx, issued.elapsed());
                    dispatch.release();
                    return Err(UpstreamError::ReadTimeout { after });
                }
            },
            None => request.await,
        };
        // Client-side admission rejections (Saturated) and configuration-
        // class errors say nothing about the PICKED endpoint's health and
        // must not eject it; only genuine transport/exchange outcomes
        // report (same classification as the breaker wire in the proxy).
        let report = match &outcome {
            Ok(resp) => Some(resp.status().as_u16() >= 500),
            Err(err) if health_reportable_legacy(err) => Some(true),
            Err(_) => None,
        };
        if let (Some(health), Some(is_failure)) = (&dispatch.health, report) {
            health.report(self.lb.now_ms(), is_failure);
        }
        // DW-090: record the observed latency for the peak-EWMA tracker.
        // Recorded for ALL outcomes (success, error, timeout) since even
        // a timeout is a latency signal. No-op when the algorithm is not
        // peak_ewma (the tracker is absent).
        self.lb.record_latency(dispatch.idx, issued.elapsed());
        // REL-06 (#217): when body-completion in-flight accounting is
        // enabled, move the in-flight guard to the response body so
        // the counter is held until the body stream completes. On
        // error paths, the guard is released normally (the body never
        // starts).
        let body_guard = if self.body_completion_inflight {
            dispatch.take_guard()
        } else {
            None
        };
        dispatch.release();
        outcome.map_err(Into::into).map(|resp| {
            (
                resp.map(|inner| {
                    let mut body =
                        UpstreamBody::from_incoming(inner, self.write_timeout, body_health);
                    if let Some(guard) = body_guard {
                        body.set_inflight_guard(guard);
                    }
                    body
                }),
                (),
            )
        })
    }
}

impl std::fmt::Debug for UpstreamHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("UpstreamHandle")
            .field("name", &self.name)
            .field("cap", &self.cap)
            .field("connect_timeout", &self.connect_timeout)
            .field("scheme", &self.scheme)
            .finish_non_exhaustive()
    }
}

/// Public for testing the connection-cap clamping contract.
pub fn effective_cap(u: &Upstream) -> u32 {
    // Validated configs always carry cap >= 1, but a directly constructed
    // (unvalidated) Gateway can bypass that; a zero-permit semaphore would
    // hang every dial forever. Clamp so misuse degrades to serial.
    u.connection_cap.unwrap_or(DEFAULT_CONNECTION_CAP).max(1)
}

fn effective_connect_timeout(u: &Upstream) -> Duration {
    Duration::from_millis(
        u.timeouts
            .as_ref()
            .and_then(|t: &Timeouts| t.connect_ms)
            .unwrap_or(DEFAULT_CONNECT_TIMEOUT_MS),
    )
}

/// `timeouts.read_ms`, resolved to a Duration. None leaves the
/// request/headers exchange unbounded (DW-014 enforcement is opt-in by
/// configuring the knob).
fn effective_read_timeout(u: &Upstream) -> Option<Duration> {
    u.timeouts
        .as_ref()
        .and_then(|t| t.read_ms)
        .map(Duration::from_millis)
}

/// `timeouts.write_ms`, resolved to a Duration (response-body inactivity
/// bound; DW-014).
fn effective_write_timeout(u: &Upstream) -> Option<Duration> {
    u.timeouts
        .as_ref()
        .and_then(|t| t.write_ms)
        .map(Duration::from_millis)
}

fn effective_slow_start(u: &Upstream) -> Duration {
    Duration::from_millis(u.slow_start_ms.unwrap_or(0).min(MAX_SLOW_START_MS))
}

fn build_handle(
    u: &Upstream,
    root_store: rustls::RootCertStore,
    previous: Option<&Arc<UpstreamHandle>>,
    events: Option<&crate::events::Emitter>,
    dns: Arc<DnsCache>,
) -> Arc<UpstreamHandle> {
    let cap = effective_cap(u);
    let connect_timeout = effective_connect_timeout(u);
    let stats = Arc::new(UpstreamStats::default());
    // DW-044: one upstream-labeled emitter for this handle's state
    // machines — the breaker's transitions and (via the balancer) the
    // endpoint trackers' ejection/recovery events.
    let upstream_events = events.map(|em| em.for_upstream(&u.name));

    // DW-030: the connector resolves + dials itself (RFC 8305 happy
    // eyeballs, `timeouts.happy_eyeballs_ms`); there is no hyper-util
    // HttpConnector to configure anymore.

    // `root_store` is the trust for THIS upstream (#121): the configured
    // trusted_ca_file bundle when set, else webpki (+ any programmatic
    // extras). Kept on the handle so probes share it (tls_roots).
    let (scheme, tls, http2_only, tls_roots): (&'static str, Option<_>, bool, Option<_>) = match u
        .protocol
    {
        UpstreamProtocol::Http1 => ("http", None, false, None),
        UpstreamProtocol::H2c => {
            // DP-02 (#235): cleartext HTTP/2 with prior knowledge. No
            // TLS; the client speaks HTTP/2 directly over TCP. Common
            // for internal east-west gRPC traffic without TLS.
            ("http", None, true, None)
        }
        UpstreamProtocol::Https => {
            // SEC-04 / DW-109: when cert_pinning is configured AND
            // the `cert_pinning` cargo feature is ON, install the
            // SPKI pin verifier (fail-closed, no CA fallback).
            // Otherwise use the normal CA-based config.
            if let Some(verifier) = crate::security::cert_pinning::CertPinVerifier::from_upstream(u)
            {
                let cfg = crate::security::cert_pinning::client_config_with_pinning(
                    Arc::new(verifier),
                    b"http/1.1",
                );
                ("https", Some(Arc::new(cfg)), false, Some(root_store))
            } else {
                // SEC-03: mTLS client cert when configured.
                let cfg = match &u.mtls {
                        Some(mtls) => crate::security::tls::https_h1_client_config_with_auth(
                            root_store.clone(),
                            &mtls.client_cert_file,
                            &mtls.client_key_file,
                        )
                        .unwrap_or_else(|e| {
                            tracing::error!(
                                code = "upstream_mtls_load_failed",
                                upstream = %u.name,
                                "upstream mtls cert/key load failed: {e}; falling back to no client auth"
                            );
                            crate::security::tls::https_h1_client_config_pq(
                                root_store.clone(),
                                u.pq,
                            )
                        }),
                        None => crate::security::tls::https_h1_client_config_pq(
                            root_store.clone(),
                            u.pq,
                        ),
                    };
                ("https", Some(Arc::new(cfg)), false, Some(root_store))
            }
        }
        UpstreamProtocol::Http2 => {
            // Same roots as https, but ALPN h2 and a client locked to
            // HTTP/2 (see module docs). DW-105: when the upstream opts
            // in to PQ hybrid key exchange (`pq: true`), prepend the
            // hybrid kx group before building the config (experimental
            // no-op when the rustls PQ API is not reachable).
            if let Some(verifier) = crate::security::cert_pinning::CertPinVerifier::from_upstream(u)
            {
                let cfg = crate::security::cert_pinning::client_config_with_pinning(
                    Arc::new(verifier),
                    b"h2",
                );
                ("https", Some(Arc::new(cfg)), true, Some(root_store))
            } else {
                if u.pq {
                    let _ = crate::security::pq::install_pq_kx_group();
                }
                // SEC-03: mTLS client cert when configured.
                let cfg = match &u.mtls {
                        Some(mtls) => crate::security::tls::https_h2_client_config_with_auth(
                            root_store.clone(),
                            &mtls.client_cert_file,
                            &mtls.client_key_file,
                        )
                        .unwrap_or_else(|e| {
                            tracing::error!(
                                code = "upstream_mtls_load_failed",
                                upstream = %u.name,
                                "upstream mtls cert/key load failed: {e}; falling back to no client auth"
                            );
                            let mut c = rustls::ClientConfig::builder()
                                .with_root_certificates(root_store.clone())
                                .with_no_client_auth();
                            c.alpn_protocols = vec![b"h2".to_vec()];
                            c
                        }),
                        None => {
                            let mut c = rustls::ClientConfig::builder()
                                .with_root_certificates(root_store.clone())
                                .with_no_client_auth();
                            c.alpn_protocols = vec![b"h2".to_vec()];
                            c
                        }
                    };
                ("https", Some(Arc::new(cfg)), true, Some(root_store))
            }
        }
        // DW-108: H3 dials QUIC, not the TCP/TLS pooled client. The
        // legacy `client` below is built but never used for an H3
        // upstream (send_inner dispatches to the `h3` handle). scheme
        // is "https" so URI/authority construction is shared; the
        // trust roots are kept so the QUIC active health probe shares
        // trust with the H3 connector (#121).
        UpstreamProtocol::H3 => ("https", None, false, Some(root_store)),
    };

    // DW-108: build the H3/QUIC transport for an `h3` upstream (feature
    // on). A build failure (the quinn client endpoint cannot bind an
    // ephemeral port, or the rustls config is rejected by QUIC) is a
    // torn-state the operator must see: log loudly and leave the handle
    // inert (h3 = None) so every dispatch fails closed with
    // `UpstreamError::H3Unavailable` rather than half-working.
    let h3_handle: Option<Arc<crate::dataplane::upstream_h3::H3UpstreamHandle>> =
        if matches!(u.protocol, UpstreamProtocol::H3) {
            match crate::dataplane::upstream_h3::H3UpstreamHandle::new(
                u.name.clone(),
                tls_roots
                    .clone()
                    .unwrap_or_else(rustls::RootCertStore::empty),
                connect_timeout,
                cap,
                effective_read_timeout(u),
                Arc::clone(&stats),
            ) {
                Ok(h) => Some(Arc::new(h)),
                Err(err) => {
                    tracing::error!(
                        code = "upstream_h3_build_failed",
                        upstream = %u.name,
                        "h3 upstream transport build failed; failing closed (every dispatch will \
                         return H3Unavailable): {err}"
                    );
                    None
                }
            }
        } else {
            None
        };

    let connector = UpstreamConnector {
        happy_eyeballs: effective_happy_eyeballs(u),
        tls,
        cap: Arc::new(Semaphore::new(cap as usize)),
        pending_cap: u
            .max_pending
            .filter(|p| *p > 0)
            .or(Some(DEFAULT_MAX_PENDING))
            .map(|p| Arc::new(Semaphore::new(p as usize))),
        connect_timeout,
        stats: Arc::clone(&stats),
        dns,
    };

    // Pool sharding: build N independent hyper-util clients, each with its
    // own `Mutex<PoolInner>`. hyper-util's pool lock is the dominant
    // serialization bottleneck under high concurrency (profiled at ~21% of
    // CPU in mutex waits at 80k RPS). Sharding distributes the lock N ways.
    // The connector is cloned per shard — all shared state (connection-cap
    // semaphore, DNS cache, TLS config, stats) is `Arc`-shared, so the
    // TOTAL connection cap still applies across all shards.
    let pool_shards = pool_shard_count();
    let build_one_client = || {
        let mut builder = Client::builder(TokioExecutor::new());
        if http2_only {
            builder.http2_only(true);
        }
        // The pool timer is ALWAYS installed: it is required for
        // `pool_idle_timeout` eviction to fire (hyper-util silently no-ops
        // idle-timeout without a timer). DP-03: the optional `pool` block
        // layers operator-tuned knobs on top of this baseline; omitted knobs
        // keep hyper-util's built-in defaults (the prior behavior).
        builder.pool_timer(TokioTimer::new());
        // The h2 timer is likewise ALWAYS installed: hyper requires it to
        // schedule HTTP/2 keep-alive PINGs, and panics ("You must supply a
        // timer") on the first h2 connection when
        // `http2_keep_alive_interval` is set without one. Installing it
        // unconditionally is harmless (it is inert when no PING interval is
        // configured) and matches the pool-timer pattern.
        builder.timer(TokioTimer::new());
        if let Some(p) = &u.pool {
            // Pool knobs apply to every protocol (h1 and h2 reuse the same
            // hyper-util connection pool).
            // SCALE-10 (#189): max_connection_age_ms sets the pool idle
            // timeout to the minimum of itself and pool_idle_timeout_ms,
            // so connections are evicted before they become stale.
            let idle_timeout = match (p.pool_idle_timeout_ms, p.max_connection_age_ms) {
                (Some(a), Some(b)) => Some(Duration::from_millis(a.min(b))),
                (Some(a), None) => Some(Duration::from_millis(a)),
                (None, Some(b)) => Some(Duration::from_millis(b)),
                (None, None) => None,
            };
            if let Some(d) = idle_timeout {
                builder.pool_idle_timeout(Some(d));
            }
            if let Some(n) = p.pool_max_idle_per_host {
                builder.pool_max_idle_per_host(n as usize);
            }
            // HTTP/2 knobs are only effective on an h2 client, but hyper-util
            // accepts them on the builder regardless (they are inert when no
            // h2 connection is negotiated), so they are wired unconditionally
            // rather than gated on `http2_only`.
            if let Some(ms) = p.http2_keep_alive_interval_ms {
                builder.http2_keep_alive_interval(Some(Duration::from_millis(ms)));
            }
            if let Some(ms) = p.http2_keep_alive_timeout_ms {
                builder.http2_keep_alive_timeout(Duration::from_millis(ms));
            }
            if p.http2_adaptive_window {
                builder.http2_adaptive_window(true);
            }
            if let Some(n) = p.max_concurrent_streams {
                // hyper-util exposes the client-side send-concurrency cap as
                // `http2_initial_max_send_streams` (the h2 client's own
                // stream limiter, distinct from the SETTINGS value the client
                // advertises to the server). This is the knob operators mean
                // when they ask to "tune max_concurrent_streams" for p99.
                builder.http2_initial_max_send_streams(n as usize);
            }
            // PERF-11 (#244): per-upstream h2 flow-control window sizes.
            if let Some(sz) = p.http2_initial_stream_window_size {
                builder.http2_initial_stream_window_size(sz);
            }
            if let Some(sz) = p.http2_initial_connection_window_size {
                builder.http2_initial_connection_window_size(sz);
            }
        }
        builder.build(connector.clone())
    };
    let clients = (0..pool_shards)
        .map(|_| build_one_client())
        .collect::<Vec<_>>();

    // Hot-swap: an existing balancer for this upstream name keeps its
    // live state (in-flight counters, WRR phase, slow-start clocks,
    // passive-health trackers) for unchanged endpoint addresses; a fresh
    // one starts clean. The retry budget is carried the same way (the
    // rolling window survives a reload), while the retry PARAMETERS apply
    // from the new config.
    let previous_lb = previous.map(|h| h.lb());
    let lb = match previous_lb {
        Some(prev) => {
            prev.rebuild_with_health_and_events(
                &u.endpoints,
                u.load_balancer,
                effective_slow_start(u),
                u.health.as_ref(),
                upstream_events.as_ref(),
                u.peak_ewma.as_ref(),
                u.locality.as_ref(),
            );
            Arc::clone(prev)
        }
        None => crate::dataplane::balance::UpstreamLb::new_with_health_and_events(
            &u.endpoints,
            u.load_balancer,
            effective_slow_start(u),
            u.health.as_ref(),
            upstream_events.as_ref(),
            u.peak_ewma.as_ref(),
            u.locality.as_ref(),
        ),
    };
    let retry_budget = previous
        .map(|h| Arc::clone(h.retry_budget()))
        .unwrap_or_else(|| Arc::new(RetryBudget::new()));
    // Breaker state carries across reloads; PARAMETERS apply from the new
    // config (the RetryBudget/RetryParams split, verbatim). A carried
    // breaker keeps its emitter binding (state-only object; the emitter
    // is per-dataplane and stable); a fresh one binds it here (DW-044).
    let breaker =
        previous
            .map(|h| Arc::clone(h.breaker()))
            .unwrap_or_else(|| match &upstream_events {
                Some(events) => Arc::new(Breaker::with_clock_and_events(
                    system_now_ms_for_breaker,
                    Some(events.clone()),
                )),
                None => Arc::new(Breaker::new()),
            });

    Arc::new(UpstreamHandle {
        name: u.name.clone(),
        cap,
        connect_timeout,
        read_timeout: effective_read_timeout(u),
        write_timeout: effective_write_timeout(u),
        retries: RetryParams::from_config(u.retries.as_ref()),
        hedge: HedgeParams::from_config(u.retries.as_ref().and_then(|r| r.hedge.as_ref())),
        retry_budget,
        max_pending: u
            .max_pending
            .filter(|p| *p > 0)
            .unwrap_or(DEFAULT_MAX_PENDING),
        breaker,
        breaker_params: u.breaker.as_ref().map(BreakerParams::from_config),
        happy_eyeballs: effective_happy_eyeballs(u),
        stats,
        clients,
        shard_counter: std::sync::atomic::AtomicUsize::new(0),
        lb,
        scheme,
        http2_only,
        tls_roots,
        protocol: u.protocol,
        hash_on: u.hash_on.clone(),
        h3: h3_handle,
        pre_warm: u.pool.as_ref().and_then(|p| p.pre_warm).unwrap_or(0),
        body_completion_inflight: u
            .health
            .as_ref()
            .is_some_and(|h| h.body_completion_inflight),
    })
}

/// Registry of per-upstream pooled clients, built from one published
/// snapshot. Rebuild (and drop the old registry) on snapshot swap; in-flight
/// requests keep their old pools until their handles are dropped.
#[derive(Default)]
pub struct UpstreamRegistry {
    handles: BTreeMap<String, Arc<UpstreamHandle>>,
    /// Compiled service splits (DW-040), keyed by service name. Empty
    /// unless a service carries a `split` block; single-target
    /// services resolve through `handles` as ever.
    splits: BTreeMap<String, Arc<crate::dataplane::split::ServiceSplit>>,
}

impl UpstreamRegistry {
    /// Build from a snapshot, verifying upstream TLS certificates against
    /// the Mozilla webpki root set — except upstreams that configure
    /// `trusted_ca_file`, which verify against their own bundle (#121).
    pub fn from_snapshot(snapshot: &Snapshot) -> Self {
        Self::with_root_certificates(snapshot, &[])
            .expect("registry build without extra roots cannot fail")
    }

    /// Build from a snapshot, carrying over per-upstream load-balancer
    /// state from `previous` (in-flight counters, WRR phase, slow-start
    /// clocks for unchanged endpoint addresses). This is the reload path:
    /// weight and endpoint changes take effect without a restart, and
    /// in-flight requests holding old handles are unaffected.
    pub fn from_snapshot_with_previous(snapshot: &Snapshot, previous: &UpstreamRegistry) -> Self {
        Self::with_root_certificates_and_previous(snapshot, &[], Some(previous))
            .expect("registry build without extra roots cannot fail")
    }

    /// `from_snapshot` with the dataplane's event emitter attached
    /// (DW-044): breaker transitions and endpoint ejection/recovery in
    /// this registry's state machines emit onto the bus. Distinct
    /// constructors rather than a changed signature so the many
    /// event-agnostic call sites (tests, tooling) stay untouched.
    pub fn from_snapshot_with_events(
        snapshot: &Snapshot,
        events: Option<&crate::events::Emitter>,
    ) -> Self {
        Self::with_root_certificates_previous_and_events(snapshot, &[], None, events)
            .expect("registry build without extra roots cannot fail")
    }

    /// `from_snapshot_with_previous` with the dataplane's event emitter
    /// attached (DW-044); the dataplane's reload path.
    pub fn from_snapshot_with_previous_and_events(
        snapshot: &Snapshot,
        previous: &UpstreamRegistry,
        events: Option<&crate::events::Emitter>,
    ) -> Self {
        Self::with_root_certificates_previous_and_events(snapshot, &[], Some(previous), events)
            .expect("registry build without extra roots cannot fail")
    }

    /// Build from a snapshot with EXTRA trusted root certificates (e.g. a
    /// private CA signing upstream certificates), added on top of the
    /// webpki roots. Fails if any extra root is malformed, so operators
    /// see the bad certificate at registry build time rather than as a
    /// mysterious handshake failure later.
    pub fn with_root_certificates(
        snapshot: &Snapshot,
        extra_roots: &[CertificateDer<'_>],
    ) -> Result<Self, UpstreamError> {
        Self::with_root_certificates_and_previous(snapshot, extra_roots, None)
    }

    /// `with_root_certificates` with optional balancer-state carry-over
    /// from a previous registry build (see
    /// [`UpstreamRegistry::from_snapshot_with_previous`]).
    ///
    /// Per-upstream `trusted_ca_file` (#121): when set, that upstream's
    /// connector trusts the file's bundle INSTEAD of the webpki+extras
    /// store — a private-CA upstream must not implicitly keep public-root
    /// trust. Config-driven bundles that fail to load at build time do
    /// NOT fail the build (the config was file-validated at publish, so
    /// this is a torn state — the bundle was replaced/unlinked
    /// underneath the gateway); the upstream FAILS CLOSED with an empty
    /// root store and a loud error log, so no traffic is ever sent under
    /// trust the operator did not configure. Programmatic `extra_roots`
    /// keep the immediate-Err contract above.
    pub fn with_root_certificates_and_previous(
        snapshot: &Snapshot,
        extra_roots: &[CertificateDer<'_>],
        previous: Option<&UpstreamRegistry>,
    ) -> Result<Self, UpstreamError> {
        Self::with_root_certificates_previous_and_events(snapshot, extra_roots, previous, None)
    }

    /// `with_root_certificates_and_previous` with the dataplane's event
    /// emitter (DW-044); the single build path behind every constructor.
    pub fn with_root_certificates_previous_and_events(
        snapshot: &Snapshot,
        extra_roots: &[CertificateDer<'_>],
        previous: Option<&UpstreamRegistry>,
        events: Option<&crate::events::Emitter>,
    ) -> Result<Self, UpstreamError> {
        let mut default_roots = crate::security::tls::webpki_root_store();
        for c in extra_roots {
            default_roots
                .add(c.clone())
                .map_err(|e| UpstreamError::InvalidRootCertificate(e.to_string()))?;
        }
        // PERF-05 (#170): one shared async DNS cache for every upstream
        // in this registry so repeated dials to the same host hit the
        // TTL/negative cache. Built once per registry generation; a
        // reload builds a fresh cache (acceptable: reloads are
        // infrequent, and the cache warms on the first dial to each
        // host). Default name servers (the public resolvers) plus the
        // system hosts file cover IP-literal and `/etc/hosts` endpoints
        // without a network round-trip.
        let dns = Arc::new(DnsCache::default());
        let handles: BTreeMap<String, Arc<UpstreamHandle>> = snapshot
            .gateway()
            .upstreams
            .iter()
            .map(|u| {
                let roots = match &u.trusted_ca_file {
                    Some(path) => match crate::security::tls::root_store_from_pem_file(path) {
                        Ok(roots) => roots,
                        Err(err) => {
                            // See the method docs: fail closed, never
                            // fall back to the public roots (that would
                            // silently change who this upstream trusts).
                            tracing::error!(
                                code = "upstream_ca_unloadable",
                                upstream = %u.name,
                                path = %path,
                                "trusted_ca_file unusable; failing closed (every TLS dial to \
                                 this upstream will be refused): {err}"
                            );
                            rustls::RootCertStore::empty()
                        }
                    },
                    None if u.use_system_roots => {
                        // REL-08 (#219): the upstream opted into the
                        // OS-native root store instead of the bundled
                        // webpki set. Loaded per-upstream (not shared)
                        // because the OS store is process-wide and a
                        // reload may pick up newly-installed roots.
                        crate::security::tls::system_root_store()
                    }
                    None => default_roots.clone(),
                };
                let prev = previous.and_then(|p| p.handles.get(&u.name));
                (
                    u.name.clone(),
                    build_handle(u, roots, prev, events, Arc::clone(&dns)),
                )
            })
            .collect();
        // DW-040: compile each split service's weighted targets.
        // Validation guarantees every named upstream exists and the
        // total weight is positive; a handle miss here (impossible
        // through validation) skips the split loudly — the service
        // answers 502 unknown_upstream rather than dispatching by a
        // half-built split.
        let mut splits = BTreeMap::new();
        for service in &snapshot.gateway().services {
            let Some(split_cfg) = &service.split else {
                continue;
            };
            let mut targets = Vec::with_capacity(split_cfg.targets.len());
            let mut ok = true;
            for t in &split_cfg.targets {
                match handles.get(&t.upstream) {
                    Some(handle) => targets.push((Arc::clone(handle), u64::from(t.weight))),
                    None => {
                        tracing::error!(
                            code = "service_split_target_missing",
                            service = %service.name,
                            upstream = %t.upstream,
                            "split target has no compiled upstream; split skipped (fail closed)"
                        );
                        ok = false;
                        break;
                    }
                }
            }
            if ok {
                splits.insert(
                    service.name.clone(),
                    Arc::new(crate::dataplane::split::ServiceSplit::new(&targets)),
                );
            }
        }
        Ok(UpstreamRegistry { handles, splits })
    }

    /// The compiled split for a service (DW-040), when it has one.
    pub fn split_for(&self, service: &str) -> Option<Arc<crate::dataplane::split::ServiceSplit>> {
        self.splits.get(service).cloned()
    }

    /// Handle for the named upstream, or None if the snapshot has no such
    /// upstream.
    pub fn get(&self, name: &str) -> Option<Arc<UpstreamHandle>> {
        self.handles.get(name).cloned()
    }

    /// Names of all upstreams in this registry.
    pub fn names(&self) -> Vec<&str> {
        self.handles.keys().map(String::as_str).collect()
    }

    /// DW-094 (Ent): set the edge's locality context on every upstream's
    /// load balancer in this registry. Called by the dataplane at startup
    /// and on refresh to propagate the edge's region/zone.
    pub fn set_locality_context(&self, ctx: crate::config::LocalityContext) {
        for handle in self.handles.values() {
            handle.set_locality_context(ctx.clone());
        }
    }

    /// DW-091: rebuild the split for `service` from explicit
    /// `(upstream_name, weight)` pairs. Returns a new registry with
    /// the replacement split; the handles are shared (Arc bump). The
    /// total weight MUST stay constant (the caller enforces this);
    /// changing the total would reshuffle every session's branch
    /// (the DW-040 invariant).
    pub fn with_rebuilt_split_from_pairs(
        &self,
        service: &str,
        pairs: &[(String, u64)],
    ) -> Option<UpstreamRegistry> {
        let mut new_splits = self.splits.clone();
        let mut targets = Vec::with_capacity(pairs.len());
        let mut ok = true;
        for (name, weight) in pairs {
            let Some(handle) = self.handles.get(name) else {
                ok = false;
                break;
            };
            targets.push((Arc::clone(handle), *weight));
        }
        if !ok {
            return None;
        }
        new_splits.insert(
            service.to_string(),
            Arc::new(crate::dataplane::split::ServiceSplit::new(&targets)),
        );
        Some(UpstreamRegistry {
            handles: self.handles.clone(),
            splits: new_splits,
        })
    }

    /// SCALE-10 (#189): pre-warm connection pools for upstreams that
    /// configure `pool.pre_warm`. For each upstream with a non-zero
    /// `pre_warm` value, spawns a background task that sends lightweight
    /// HEAD requests to each endpoint to establish `pre_warm`
    /// connections in the pool. This removes cold-start latency spikes
    /// after reloads and upgrade hand-offs. Failures are logged and
    /// silently ignored (pre-warming is best-effort; a failed
    /// pre-warm does not affect the upstream's availability).
    pub fn pre_warm(&self) {
        for (name, handle) in &self.handles {
            let n = handle.pre_warm;
            if n == 0 {
                continue;
            }
            let handle = Arc::clone(handle);
            let name = name.clone();
            tokio::spawn(async move {
                let endpoints = handle.lb().len();
                if endpoints == 0 {
                    return;
                }
                let per_endpoint = (n as usize).min(64);
                tracing::info!(
                    code = "upstream_pre_warm_start",
                    upstream = %name,
                    endpoints,
                    per_endpoint,
                    "pre-warming {} connections per endpoint for upstream {name}",
                    per_endpoint,
                );
                for _ in 0..per_endpoint {
                    for _ in 0..endpoints {
                        // Send a lightweight HEAD request to establish
                        // a connection. The request targets the
                        // upstream's origin with a minimal path; the
                        // response is discarded. Failures are silently
                        // ignored (best-effort pre-warming).
                        let req = http::Request::builder()
                            .method("HEAD")
                            .uri("/")
                            .body(http_body_util::Empty::<bytes::Bytes>::new().boxed())
                            .unwrap();
                        let _ = handle.send(req).await;
                    }
                }
                tracing::info!(
                    code = "upstream_pre_warm_done",
                    upstream = %name,
                    "pre-warming complete for upstream {name}",
                );
            });
        }
    }
}

/// Refresh the state-derived observation gauges (breaker state, endpoint
/// health, fail-open picks) from `registry` into `obs` (DW-021).
///
/// Called at scrape time: `/metrics` reflects a point-in-time snapshot,
/// and the hot paths (pick, report) stay pure atomics with zero metrics
/// coupling. The walk lives here — next to the registry state it reads —
/// and the observability side only exposes plain setters, so metrics
/// depend on nothing. Series for endpoints/upstreams removed by a reload
/// linger until process restart (documented Prometheus caveat).
pub fn refresh_observation_gauges(registry: &UpstreamRegistry, obs: &Observability) {
    for name in registry.names() {
        let Some(handle) = registry.get(name) else {
            continue;
        };
        let state = match handle.breaker().state() {
            BreakerState::Closed { .. } => 0,
            BreakerState::Open { .. } => 1,
            BreakerState::HalfOpen { .. } => 2,
        };
        obs.set_breaker_state(name, state);
        obs.set_fail_open_picks(name, handle.lb().fail_open_picks() as i64);
        let lb_now = handle.lb().now_ms();
        for (address, port, health) in handle.lb().health_targets() {
            let label = format!("{address}:{port}");
            let up = health.map(|h| h.is_available(lb_now)).unwrap_or(true);
            obs.set_endpoint_health(name, &label, up);
        }
    }
}

// White-box test staying in src/ per AGENTS.md: exercises private
// upstream client internals not reachable through the public API.
#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::{LoadBalancer, Upstream as ConfigUpstream};
    use bytes::Bytes;
    use http_body_util::Full;
    use hyper::Method;

    fn get_request(path: &str) -> Request<Full<Bytes>> {
        Request::builder()
            .method(Method::GET)
            .uri(path)
            .body(Full::new(Bytes::new()))
            .expect("request")
    }

    #[tokio::test]
    async fn send_without_endpoints_returns_no_endpoints() {
        // Only reachable via unvalidated direct construction; send() is
        // the guard, so the handle dials no fabricated address.
        let up = ConfigUpstream {
            name: "bare".into(),
            load_balancer: LoadBalancer::RoundRobin,
            hash_on: None,
            protocol: UpstreamProtocol::Http1,
            endpoints: vec![],
            connection_cap: None,
            slow_start_ms: None,
            health: None,
            active_health: None,
            retries: None,
            timeouts: None,
            breaker: None,
            max_pending: None,
            trusted_ca_file: None,
            use_system_roots: false,
            oauth2_client_credentials: None,
            dns_discovery: None,
            peak_ewma: None,
            locality: None,
            pq: false,
            cert_pinning: None,
            mtls: None,
            pool: None,
        };
        let handle = build_handle(
            &up,
            crate::security::tls::webpki_root_store(),
            None,
            None,
            Arc::new(DnsCache::default()),
        );
        assert!(matches!(
            handle.send(get_request("/x")).await,
            Err(UpstreamError::NoEndpoints)
        ));
    }
}