apollo-http-client 0.3.0

// Items in this module are used by `#[cfg(test)]` tests but are not yet wired into
// `HttpClient`. This `allow(dead_code)` will be removed when scheme-aware dispatch
// lands and any remaining dead items will surface naturally.
#![allow(dead_code)]

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;

use http::uri::Authority;
use hyper_util::rt::TokioIo;
use tokio::net::UnixStream;
use tower::BoxError;
use tower::Service;

use super::handshake::{FixedOrigin, Http1Handshake, Http2Handshake};
use super::{HostSender, HttpBody, Origin};
use crate::config::Protocol;
use crate::error::HttpClientError;
use crate::metrics::ConnectionMetrics;

// ---- Path / authority conversion --------------------------------------------

// `unix://` URIs hex-encode the socket path bytes into the URI authority — e.g.
// `/var/run/app.sock` becomes `unix://2f7661722f72756e2f6170702e736f636b/`.
// `http::Uri`'s authority rejects `/` and percent-encoded `%2F`, so the raw path
// can't appear there directly. Hex digits are a strict subset of valid host
// characters and round-trip unambiguously. The encoding is an implementation
// detail; callers should use [`unix_uri`] rather than constructing URIs by hand.

/// Converts a Unix socket path into a URI authority by hex-encoding its bytes.
///
/// Returns an error for an empty path, the only input that produces an invalid authority —
/// hex digits are otherwise always valid host characters.
pub(crate) fn path_to_authority(socket_path: &str) -> Result<Authority, http::uri::InvalidUri> {
    let encoded = hex::encode(socket_path.as_bytes());
    Authority::try_from(encoded.as_bytes())
}

/// Returns a [`http::uri::Builder`] pre-configured with the `unix` scheme and the
/// hex-encoded socket path as authority.
///
/// Use [`http::uri::Builder::path_and_query`] to set the request path before calling
/// [`http::uri::Builder::build`].
///
/// Taking `&str` makes non-UTF-8 socket paths unrepresentable through this API.
///
/// # Errors
/// Returns [`HttpClientError::InvalidUri`] for an empty `socket_path`.
///
/// # Example
/// ```
/// use apollo_http_client::unix_uri;
///
/// let uri = unix_uri("/var/run/app.sock")?
///     .path_and_query("/api/v1/users")
///     .build()?;
/// assert_eq!(uri.scheme_str(), Some("unix"));
/// assert_eq!(uri.path(), "/api/v1/users");
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
pub fn unix_uri(socket_path: &str) -> Result<http::uri::Builder, HttpClientError> {
    let authority =
        path_to_authority(socket_path).map_err(|source| HttpClientError::InvalidUri {
            source: source.into(),
        })?;
    Ok(http::Uri::builder().scheme("unix").authority(authority))
}

/// Converts a `unix://` URI authority back into the Unix socket path.
///
/// Only the host portion of the authority is decoded — any port suffix (e.g. the `:0`
/// used by [hyperlocal]) is ignored. Non-UTF-8 paths are rejected: there is no way to
/// construct a `unix://` URI for one through this crate's helpers, and a hand-crafted
/// URI that decodes to non-UTF-8 bytes returns [`HttpClientError::NonUtf8UnixSocketPath`].
///
/// [hyperlocal]: https://github.com/softprops/hyperlocal
pub(crate) fn authority_to_str(authority: &Authority) -> Result<String, HttpClientError> {
    let bytes = hex::decode(authority.host())
        .map_err(|source| HttpClientError::InvalidUnixSocketPath { source })?;
    String::from_utf8(bytes).map_err(|source| HttpClientError::NonUtf8UnixSocketPath { source })
}

// ---- IO connector -----------------------------------------------------------

/// Establishes a Unix-domain-socket IO stream for a given [`Origin`].
///
/// Decodes the socket path from the URI authority, applies the connect timeout,
/// and returns a `TokioIo<UnixStream>` ready to be fed into [`Http1Handshake`] or
/// [`Http2Handshake`].
#[derive(Clone)]
pub(crate) struct UnixIoConnector {
    connect_timeout: Duration,
}

impl UnixIoConnector {
    pub(crate) fn new(connect_timeout: Duration) -> Self {
        Self { connect_timeout }
    }
}

impl Service<Origin> for UnixIoConnector {
    type Response = TokioIo<UnixStream>;
    type Error = BoxError;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, origin: Origin) -> Self::Future {
        let connect_timeout = self.connect_timeout;
        Box::pin(async move {
            let socket_path = authority_to_str(&origin.authority)?;
            let stream = tokio::time::timeout(connect_timeout, UnixStream::connect(&socket_path))
                .await
                .map_err(|_| HttpClientError::ConnectionTimeout)?
                .map_err(|e| HttpClientError::Transport {
                    source: Box::new(e),
                })?;
            Ok(TokioIo::new(stream))
        })
    }
}

// ---- Senders ----------------------------------------------------------------

/// All Unix-socket-related configuration extracted from
/// [`crate::config::HttpClientConfig`].
pub(crate) struct UnixSenderConfig {
    pub(crate) protocol: Protocol,
    pub(crate) connect_timeout: Duration,
    pub(crate) h1_max_idle: usize,
    pub(crate) h1_idle_timeout: Duration,
    pub(crate) h2_idle_timeout: Duration,
    pub(crate) keep_alive_interval: Duration,
    pub(crate) keep_alive_timeout: Duration,
    pub(crate) keep_alive_while_idle: bool,
    pub(crate) conn_metrics: ConnectionMetrics,
}

/// Returns a [`HostSender`] that routes requests for `key` through a Unix
/// domain socket using the configured HTTP protocol.
///
/// - `Protocol::Http1` and `Protocol::Alpn` use a pooled HTTP/1.1 sender.
///   UDS has no TLS to negotiate, so ALPN falls back to HTTP/1.1.
/// - `Protocol::Http2` uses a single multiplexed HTTP/2 connection.
pub(crate) fn new_sender(config: &Arc<UnixSenderConfig>, origin: Origin) -> HostSender {
    match config.protocol {
        Protocol::Http2 => new_h2_sender(config, origin),
        Protocol::Http1 | Protocol::Alpn => new_h1_sender(config, origin),
    }
}

fn new_h1_sender(config: &Arc<UnixSenderConfig>, origin: Origin) -> HostSender {
    let attrs = config
        .conn_metrics
        .connection_attrs(Some(http::Version::HTTP_11), &origin);
    let handshake = Http1Handshake::new(
        UnixIoConnector::new(config.connect_timeout),
        config.conn_metrics.clone(),
        attrs,
    );
    crate::protocol::h1::new_sender(
        handshake,
        config.h1_max_idle,
        config.h1_idle_timeout,
        origin,
        rewrite_authority_to_localhost,
    )
}

fn new_h2_sender(config: &Arc<UnixSenderConfig>, origin: Origin) -> HostSender {
    let attrs = config
        .conn_metrics
        .connection_attrs(Some(http::Version::HTTP_2), &origin);
    let handshake = Http2Handshake {
        inner: FixedOrigin {
            inner: UnixIoConnector::new(config.connect_timeout),
            origin,
        },
        metrics: config.conn_metrics.clone(),
        attrs,
        keep_alive_interval: config.keep_alive_interval,
        keep_alive_timeout: config.keep_alive_timeout,
        keep_alive_while_idle: config.keep_alive_while_idle,
    };
    crate::protocol::h2::new_sender(
        handshake,
        config.h2_idle_timeout,
        (),
        rewrite_authority_to_localhost,
    )
}

/// Rewrites the request URI authority to `localhost` so HTTP/1.1 `Host` headers
/// and HTTP/2 `:authority` pseudo-headers carry a human-readable value rather
/// than the hex-encoded socket path. The path, query, and scheme are preserved.
fn rewrite_authority_to_localhost(req: &mut http::Request<HttpBody>) {
    let mut parts = req.uri().clone().into_parts();
    parts.authority = Some(Authority::from_static("localhost"));
    if let Ok(new_uri) = http::Uri::from_parts(parts) {
        *req.uri_mut() = new_uri;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use std::path::Path;

    use apollo_opentelemetry::metrics::Clock;
    use apollo_opentelemetry_test::{TelemetryContext, assert_metric, assert_metrics_snapshot};
    use axum::Router;
    use axum::routing::get;
    use bytes::Bytes;
    use http::uri::Scheme;
    use http_body_util::{BodyExt as _, Full};
    use opentelemetry_sdk::metrics::{Aggregation, Instrument, StreamBuilder};

    use crate::metrics::HttpMetrics;
    use crate::protocol::ProtocolVersionCell;

    // ---- Hex helper tests ---------------------------------------------------

    /// A path round-trips to and from a hex-encoded authority unchanged.
    #[test]
    fn path_authority_round_trip() {
        let path = "/var/run/app.sock";
        let authority = path_to_authority(path).expect("encode");
        let decoded = authority_to_str(&authority).expect("decode");
        assert_eq!(decoded, path);
    }

    /// A non-hex authority decodes to a typed `InvalidUnixSocketPath` error.
    #[test]
    fn non_hex_authority_returns_invalid_unix_path() {
        let authority = Authority::from_static("not-hex-zz");
        let err = authority_to_str(&authority).expect_err("should fail");
        assert!(matches!(err, HttpClientError::InvalidUnixSocketPath { .. }));
    }

    /// A hex authority that decodes to non-UTF-8 bytes is rejected with a typed
    /// `NonUtf8UnixSocketPath` error. Non-UTF-8 socket paths are not supported.
    #[test]
    fn non_utf8_hex_authority_returns_non_utf8_unix_path() {
        // Valid hex, but the bytes (0xff 0xfe) are not valid UTF-8.
        let authority = Authority::from_static("fffe");
        let err = authority_to_str(&authority).expect_err("should fail");
        assert!(matches!(err, HttpClientError::NonUtf8UnixSocketPath { .. }));
    }

    /// An empty path cannot be encoded as a valid authority.
    #[test]
    fn empty_path_cannot_be_encoded() {
        assert!(path_to_authority("").is_err());
    }

    /// A `host:port` style authority decodes to the path encoded in the host portion;
    /// the port is ignored. Matches how hyperlocal-style URIs (`unix://<hex>:0/`) are
    /// produced and accepted.
    #[test]
    fn authority_with_port_uses_host_only() {
        let with_port = Authority::from_static("2f736f636b:0");
        let decoded = authority_to_str(&with_port).expect("decode");
        assert_eq!(decoded, "/sock");
    }

    /// `unix_uri` builds a URI whose authority decodes back to the original socket path.
    #[test]
    fn unix_uri_builds_uri_with_encoded_authority() {
        let uri = unix_uri("/var/run/app.sock")
            .expect("encode")
            .path_and_query("/api")
            .build()
            .expect("build");

        assert_eq!(uri.scheme_str(), Some("unix"));
        assert_eq!(uri.path(), "/api");
        let decoded = authority_to_str(uri.authority().expect("authority")).expect("decode");
        assert_eq!(decoded, "/var/run/app.sock");
    }

    /// `unix_uri` rejects an empty socket path.
    #[test]
    fn unix_uri_rejects_empty_path() {
        assert!(unix_uri("").is_err());
    }

    // ---- Decorate tests -----------------------------------------------------

    /// `rewrite_authority_to_localhost` replaces a hex authority with `localhost`
    /// while preserving the scheme, path, and query.
    #[test]
    fn rewrite_authority_replaces_hex_with_localhost() {
        let mut req = http::Request::builder()
            .uri(
                unix_uri("/var/run/app.sock")
                    .unwrap()
                    .path_and_query("/api?x=1")
                    .build()
                    .unwrap(),
            )
            .body(empty_body())
            .unwrap();

        rewrite_authority_to_localhost(&mut req);

        assert_eq!(req.uri().scheme_str(), Some("unix"));
        assert_eq!(req.uri().authority().unwrap().as_str(), "localhost");
        assert_eq!(req.uri().path(), "/api");
        assert_eq!(req.uri().query(), Some("x=1"));
    }

    // ---- Real UnixListener integration tests --------------------------------

    /// A `TelemetryContext` that drops `http.client.connection.duration` so its
    /// non-deterministic wall-clock values don't disturb metric snapshots. The metric
    /// is verified separately via [`assert_metric!`].
    fn integration_context() -> TelemetryContext {
        TelemetryContext::builder()
            .with_view(|instrument: &Instrument| {
                if instrument.name() == "http.client.connection.duration" {
                    Some(
                        StreamBuilder::default()
                            .with_aggregation(Aggregation::Drop)
                            .build()
                            .unwrap(),
                    )
                } else {
                    None
                }
            })
            .build()
    }

    fn empty_body() -> HttpBody {
        Full::new(Bytes::new())
            .map_err(|never: std::convert::Infallible| match never {})
            .boxed()
    }

    /// Spawns an axum server bound to a Unix socket inside `tmp_dir`, returning the
    /// socket path. Each test owns its own `tempfile::tempdir`, so the filename is fixed.
    async fn spawn_unix_server(tmp_dir: &Path, router: Router) -> String {
        let socket_path = tmp_dir
            .join("test.sock")
            .into_os_string()
            .into_string()
            .expect("tempdir path is UTF-8");
        let listener = tokio::net::UnixListener::bind(&socket_path).unwrap();
        tokio::spawn(async move {
            axum::serve(listener, router).await.unwrap();
        });
        socket_path
    }

    /// Spawns a raw Unix listener that immediately closes each accepted connection,
    /// modelling a server that drops the request before responding.
    async fn spawn_unix_drop_server(tmp_dir: &Path) -> String {
        let socket_path = tmp_dir
            .join("test.sock")
            .into_os_string()
            .into_string()
            .expect("tempdir path is UTF-8");
        let listener = tokio::net::UnixListener::bind(&socket_path).unwrap();
        tokio::spawn(async move {
            while let Ok((socket, _)) = listener.accept().await {
                drop(socket);
            }
        });
        socket_path
    }

    fn make_metrics(record_url_scheme: bool) -> ConnectionMetrics {
        let (clock, _mock) = Clock::mock();
        HttpMetrics::new(clock, false, false, record_url_scheme).connection_metrics()
    }

    fn make_request(authority: &Authority) -> http::Request<HttpBody> {
        http::Request::builder()
            .method(http::Method::GET)
            .uri(
                http::Uri::builder()
                    .scheme("unix")
                    .authority(authority.clone())
                    .path_and_query("/")
                    .build()
                    .unwrap(),
            )
            .body(empty_body())
            .unwrap()
    }

    fn make_config(protocol: Protocol, record_url_scheme: bool) -> Arc<UnixSenderConfig> {
        Arc::new(UnixSenderConfig {
            protocol,
            connect_timeout: Duration::from_secs(5),
            h1_max_idle: 10,
            h1_idle_timeout: Duration::from_secs(60),
            h2_idle_timeout: Duration::from_secs(60),
            keep_alive_interval: Duration::from_secs(30),
            keep_alive_timeout: Duration::from_secs(10),
            keep_alive_while_idle: false,
            conn_metrics: make_metrics(record_url_scheme),
        })
    }

    fn make_sender_for(
        authority: &Authority,
        protocol: Protocol,
        record_url_scheme: bool,
    ) -> HostSender {
        let config = make_config(protocol, record_url_scheme);
        new_sender(
            &config,
            Origin {
                scheme: Scheme::try_from("unix").unwrap(),
                authority: authority.clone(),
            },
        )
    }

    /// Snapshots `http.client.open_connections` after a successful request, masking the
    /// dynamic socket path. The snapshot captures the full attribute set — verifying that
    /// `server.address` carries the decoded path and that `server.port` is **absent**
    /// (it would otherwise appear in the snapshot).
    macro_rules! snapshot_open_connections {
        ($ctx:expr, @$snapshot:literal) => {{
            insta::with_settings!({
                filters => [(r"server\.address: .*", "server.address: <path>")]
            }, {
                assert_metrics_snapshot!($ctx, @$snapshot);
            });
        }};
    }

    // ---- HTTP/1.1 over UDS --------------------------------------------------

    /// A successful HTTP/1.1 GET over a Unix socket dispatches via `new_sender`,
    /// returns the expected body, and emits `open_connections` with the right
    /// attributes (decoded path as `server.address`, HTTP/1.1, no `server.port`).
    #[tokio::test]
    async fn h1_successful_request_returns_response_and_emits_metrics() {
        let ctx = integration_context();
        let tmp = tempfile::tempdir().unwrap();
        let path = spawn_unix_server(
            tmp.path(),
            Router::new().route("/", get(|| async { "hello" })),
        )
        .await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http1, false);
        let resp = sender(
            make_request(&authority),
            ProtocolVersionCell::new(http::Version::HTTP_11),
        )
        .await
        .expect("request ok");
        let body = resp.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(&body[..], b"hello");

        snapshot_open_connections!(ctx, @r#"
        - name: http.client.open_connections
          description: Number of open connections in the HTTP client pool
          unit: "{connection}"
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  network.protocol.version: "1.1"
                  server.address: <path>
                value: 0
              - attributes:
                  http.connection.state: idle
                  network.protocol.version: "1.1"
                  server.address: <path>
                value: 1
            is_monotonic: false
            temporality: Cumulative
        "#);
    }

    /// `url.scheme = "unix"` is added as a metric attribute when the opt-in flag is on.
    #[tokio::test]
    async fn h1_url_scheme_attr_present_when_enabled() {
        let ctx = TelemetryContext::new();
        let tmp = tempfile::tempdir().unwrap();
        let path =
            spawn_unix_server(tmp.path(), Router::new().route("/", get(|| async { "ok" }))).await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http1, true);
        sender(
            make_request(&authority),
            ProtocolVersionCell::new(http::Version::HTTP_11),
        )
        .await
        .expect("request ok")
        .into_body()
        .collect()
        .await
        .unwrap();

        assert_metric!(ctx, "http.client.open_connections", "url.scheme" = "unix");
    }

    /// Two sequential HTTP/1.1 requests to the same socket reuse a single pooled
    /// connection — only one connection ends up in the idle state.
    #[tokio::test]
    async fn h1_pool_reuses_connection_for_same_path() {
        let ctx = integration_context();
        let tmp = tempfile::tempdir().unwrap();
        let path =
            spawn_unix_server(tmp.path(), Router::new().route("/", get(|| async { "ok" }))).await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http1, false);
        for _ in 0..2 {
            let resp = sender(
                make_request(&authority),
                ProtocolVersionCell::new(http::Version::HTTP_11),
            )
            .await
            .expect("request ok");
            resp.into_body().collect().await.unwrap();
        }

        assert_metric!(
            ctx,
            "http.client.open_connections",
            value: 1,
            "http.connection.state" = "idle"
        );
        assert_metric!(
            ctx,
            "http.client.open_connections",
            value: 0,
            "http.connection.state" = "active"
        );
    }

    /// Connecting to a non-existent socket path returns a typed transport error and
    /// never increments the connection counter.
    #[tokio::test]
    async fn h1_connection_refused_returns_transport_error() {
        let ctx = TelemetryContext::new();
        let tmp = tempfile::tempdir().unwrap();
        let missing_path = tmp
            .path()
            .join("does-not-exist.sock")
            .into_os_string()
            .into_string()
            .expect("tempdir path is UTF-8");
        let authority = path_to_authority(&missing_path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http1, false);
        let err = sender(
            make_request(&authority),
            ProtocolVersionCell::new(http::Version::HTTP_11),
        )
        .await
        .expect_err("should fail");
        assert!(matches!(err, HttpClientError::Transport { .. }));

        // No connection was established, so open_connections must not be recorded.
        let metrics = ctx.metrics();
        let names: Vec<&str> = metrics
            .iter()
            .flat_map(|rm| rm.scope_metrics())
            .flat_map(|sm| sm.metrics())
            .map(|m| m.name())
            .collect();
        assert!(
            !names.contains(&"http.client.open_connections"),
            "open_connections should not be recorded when connect fails"
        );
    }

    /// `http.client.connection.duration` is recorded on every successfully established
    /// Unix HTTP/1.1 connection.
    #[tokio::test]
    async fn h1_connection_duration_is_recorded() {
        let ctx = TelemetryContext::new();
        let tmp = tempfile::tempdir().unwrap();
        let path =
            spawn_unix_server(tmp.path(), Router::new().route("/", get(|| async { "ok" }))).await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http1, false);
        let resp = sender(
            make_request(&authority),
            ProtocolVersionCell::new(http::Version::HTTP_11),
        )
        .await
        .expect("request ok");
        resp.into_body().collect().await.unwrap();
        // Drop the sender so idle pooled connections are freed and the duration histogram fires.
        drop(sender);

        assert_metric!(ctx, "http.client.connection.duration");
    }

    /// When the server accepts the connection then closes it before responding, the
    /// dispatcher returns a `Request` error rather than `Transport`.
    #[tokio::test]
    async fn h1_server_closes_before_response_returns_request_error() {
        let _ctx = TelemetryContext::new();
        let tmp = tempfile::tempdir().unwrap();
        let path = spawn_unix_drop_server(tmp.path()).await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http1, false);
        let err = sender(
            make_request(&authority),
            ProtocolVersionCell::new(http::Version::HTTP_11),
        )
        .await
        .expect_err("should fail");
        assert!(matches!(err, HttpClientError::Request { .. }));
    }

    /// The hex-encoded socket path doesn't leak into the upstream `Host` header —
    /// it's rewritten to `localhost` before dispatch.
    #[tokio::test]
    async fn h1_host_header_is_localhost_not_hex_authority() {
        use std::sync::Arc as StdArc;
        let captured: StdArc<std::sync::Mutex<Option<String>>> =
            StdArc::new(std::sync::Mutex::new(None));
        let captured_clone = captured.clone();

        let router = Router::new().route(
            "/",
            get(move |headers: axum::http::HeaderMap| {
                let captured = captured_clone.clone();
                async move {
                    *captured.lock().unwrap() =
                        headers.get("host").map(|v| v.to_str().unwrap().to_string());
                    "ok"
                }
            }),
        );

        let tmp = tempfile::tempdir().unwrap();
        let path = spawn_unix_server(tmp.path(), router).await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http1, false);
        sender(
            make_request(&authority),
            ProtocolVersionCell::new(http::Version::HTTP_11),
        )
        .await
        .expect("request ok")
        .into_body()
        .collect()
        .await
        .unwrap();

        let host = captured.lock().unwrap().clone();
        assert_eq!(host.as_deref(), Some("localhost"));
    }

    // ---- HTTP/2 over UDS ----------------------------------------------------

    /// A successful HTTP/2 GET over a Unix socket. Verifies that the connector
    /// negotiates h2 prior-knowledge over the socket and that the response
    /// completes correctly. `network.protocol.version = "2"` is recorded.
    #[tokio::test]
    async fn h2_successful_request_returns_response_and_emits_metrics() {
        let ctx = integration_context();
        let tmp = tempfile::tempdir().unwrap();
        let path = spawn_unix_server(
            tmp.path(),
            Router::new().route("/", get(|| async { "hello" })),
        )
        .await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http2, false);
        let resp = sender(
            make_request(&authority),
            ProtocolVersionCell::new(http::Version::HTTP_2),
        )
        .await
        .expect("request ok");
        assert_eq!(resp.version(), http::Version::HTTP_2);
        let body = resp.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(&body[..], b"hello");

        snapshot_open_connections!(ctx, @r#"
        - name: http.client.open_connections
          description: Number of open connections in the HTTP client pool
          unit: "{connection}"
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  network.protocol.version: "2"
                  server.address: <path>
                value: 0
              - attributes:
                  http.connection.state: idle
                  network.protocol.version: "2"
                  server.address: <path>
                value: 1
            is_monotonic: false
            temporality: Cumulative
        "#);
    }

    /// HTTP/2 over UDS multiplexes — repeated requests reuse the single h2
    /// connection rather than opening new ones.
    #[tokio::test]
    async fn h2_pool_reuses_connection_for_same_path() {
        let ctx = integration_context();
        let tmp = tempfile::tempdir().unwrap();
        let path =
            spawn_unix_server(tmp.path(), Router::new().route("/", get(|| async { "ok" }))).await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Http2, false);
        for _ in 0..3 {
            let resp = sender(
                make_request(&authority),
                ProtocolVersionCell::new(http::Version::HTTP_2),
            )
            .await
            .expect("request ok");
            resp.into_body().collect().await.unwrap();
        }

        // Exactly one idle h2 connection — singleton pool reused for every request.
        assert_metric!(
            ctx,
            "http.client.open_connections",
            value: 1,
            "http.connection.state" = "idle"
        );
    }

    /// `Protocol::Alpn` over UDS falls back to HTTP/1.1 (no TLS to negotiate).
    #[tokio::test]
    async fn alpn_falls_back_to_h1_over_uds() {
        let ctx = TelemetryContext::new();
        let tmp = tempfile::tempdir().unwrap();
        let path =
            spawn_unix_server(tmp.path(), Router::new().route("/", get(|| async { "ok" }))).await;
        let authority = path_to_authority(&path).expect("encode");

        let sender = make_sender_for(&authority, Protocol::Alpn, false);
        let resp = sender(
            make_request(&authority),
            ProtocolVersionCell::new(http::Version::HTTP_11),
        )
        .await
        .expect("request ok");
        assert_eq!(resp.version(), http::Version::HTTP_11);
        resp.into_body().collect().await.unwrap();

        assert_metric!(
            ctx,
            "http.client.open_connections",
            "network.protocol.version" = "1.1"
        );
    }
}