apollo-http-client 0.3.0

//! Integration tests for HTTP/1.1 paths.
#![allow(unused_imports)]

#[macro_use]
mod common;

use std::net::SocketAddr;
use std::sync::{Arc, OnceLock};
use std::time::Duration;

use apollo_http_client::{HttpBody, HttpClient, HttpClientConfig};
use apollo_opentelemetry::metrics::Clock;
use apollo_opentelemetry_test::{TelemetryContext, assert_metric, assert_metrics_snapshot};
use axum::Router;
use axum::routing::{get, post};
use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper_util::rt::{TokioExecutor, TokioIo};
use indoc::indoc;
use opentelemetry_sdk::metrics::{Aggregation, Instrument, StreamBuilder};
use tower::ServiceExt as _;

use common::*;

// --- H1: success path -----------------------------------------------------

/// A successful GET records all six metrics with the expected values, including
/// body size metrics (enabled via `config_with_body_size()`).
/// `http.client.connection.duration` is verified separately in
/// [`h1_connection_duration_is_recorded`].
#[tokio::test]
async fn h1_successful_get_records_all_metrics() {
    let ctx = integration_context();

    let addr = spawn_server(Router::new().route("/", get(|| async { "hello" }))).await;
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&config_with_body_size())
        .oneshot(req)
        .await
        .expect("request ok");
    // Consume the body so response.body.size records actual bytes ("hello" = 5 B).
    resp.into_body().collect().await.unwrap();

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.body.size
      description: Size of HTTP request bodies sent by the client
      unit: By
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0
              - 1024
              - 4096
              - 16384
              - 65536
              - 262144
              - 1048576
              - 4194304
              - 16777216
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    - name: http.client.response.body.size
      description: Size of HTTP response bodies received by the client
      unit: By
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 5
            min: 5
            max: 5
            bounds:
              - 0
              - 1024
              - 4096
              - 16384
              - 65536
              - 262144
              - 1048576
              - 4194304
              - 16777216
            bucket_counts:
              - 0
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// A POST with a 1 KiB body records `http.client.request.body.size` sum = 1024.
#[tokio::test]
async fn h1_post_request_body_size_recorded() {
    let ctx = integration_context();

    let addr =
        spawn_server(Router::new().route("/", post(|_body: axum::body::Bytes| async { "ok" })))
            .await;

    let req = http::Request::builder()
        .method(http::Method::POST)
        .uri(format!("http://{addr}/"))
        .body(bytes_body(Bytes::from(vec![0u8; 1024])))
        .unwrap();

    let resp = new_client(&config_with_body_size())
        .oneshot(req)
        .await
        .expect("request ok");
    drop(resp);

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: POST
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.body.size
      description: Size of HTTP request bodies sent by the client
      unit: By
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: POST
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 1024
            min: 1024
            max: 1024
            bounds:
              - 0
              - 1024
              - 4096
              - 16384
              - 65536
              - 262144
              - 1048576
              - 4194304
              - 16777216
            bucket_counts:
              - 0
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: POST
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    - name: http.client.response.body.size
      description: Size of HTTP response bodies received by the client
      unit: By
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: POST
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0
              - 1024
              - 4096
              - 16384
              - 65536
              - 262144
              - 1048576
              - 4194304
              - 16777216
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// Dropping the response without reading records 0 bytes for `response.body.size`.
/// The guard must fire even if the body is never polled.
#[tokio::test]
async fn h1_response_body_dropped_unread_records_zero_bytes() {
    let ctx = integration_context();

    let addr = spawn_server(Router::new().route("/", get(|| async { "hello world" }))).await;
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&config_with_body_size())
        .oneshot(req)
        .await
        .expect("request ok");
    drop(resp); // body never read

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.body.size
      description: Size of HTTP request bodies sent by the client
      unit: By
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0
              - 1024
              - 4096
              - 16384
              - 65536
              - 262144
              - 1048576
              - 4194304
              - 16777216
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    - name: http.client.response.body.size
      description: Size of HTTP response bodies received by the client
      unit: By
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0
              - 1024
              - 4096
              - 16384
              - 65536
              - 262144
              - 1048576
              - 4194304
              - 16777216
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

// --- H1: error cases ------------------------------------------------------

/// A 404 response records `http.response.status_code: 404` and `error.type: "404"`
/// on the request duration histogram.
#[tokio::test]
async fn h1_4xx_response_records_status_and_error_type() {
    let ctx = integration_context();

    let addr =
        spawn_server(Router::new().route("/", get(|| async { http::StatusCode::NOT_FOUND }))).await;

    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&default_config())
        .oneshot(req)
        .await
        .expect("request ok");
    assert_eq!(resp.status(), http::StatusCode::NOT_FOUND);
    drop(resp);

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              error.type: "404"
              http.request.method: GET
              http.response.status_code: "404"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// A 500 response records `error.type: "500"` and `http.response.status_code: 500`.
#[tokio::test]
async fn h1_5xx_response_records_error_type() {
    let ctx = integration_context();

    let addr = spawn_server(Router::new().route(
        "/",
        get(|| async { http::StatusCode::INTERNAL_SERVER_ERROR }),
    ))
    .await;

    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&default_config())
        .oneshot(req)
        .await
        .expect("request ok");
    assert_eq!(resp.status(), http::StatusCode::INTERNAL_SERVER_ERROR);
    drop(resp);

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              error.type: "500"
              http.request.method: GET
              http.response.status_code: "500"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

// --- H1: connection state -------------------------------------------------

/// `http.client.open_connections` is tracked: the connection transitions
/// between active (during request) and idle (when returned to pool) and returns
/// to zero once the connection is closed.
#[tokio::test]
async fn h1_open_connections_tracked() {
    let ctx = integration_context();

    let addr = spawn_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&default_config())
        .oneshot(req)
        .await
        .expect("request ok");
    // Consume body so the connection can be returned to the pool.
    resp.into_body().collect().await.unwrap();

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// `http.client.connection.duration` is recorded for every established connection.
/// Verified separately because the wall-clock sum value is non-deterministic.
#[tokio::test]
async fn h1_connection_duration_is_recorded() {
    let ctx = TelemetryContext::new();

    let addr = spawn_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    new_client(&default_config())
        .oneshot(req)
        .await
        .expect("request ok");

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

/// Idle connections are freed immediately when the client is dropped — not deferred
/// until the next eviction task wakeup. `http.client.connection.duration` is recorded
/// synchronously on drop, proving the connection was freed right away.
///
/// Uses `client.clone()` for the request so the original `client` remains alive until
/// the explicit `drop(client)`, making the timing unambiguous.
#[tokio::test]
async fn h1_connections_freed_immediately_on_client_drop() {
    let ctx = TelemetryContext::new();
    let addr = spawn_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let client = new_client(&default_config());
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    // Use a clone so `client` retains the HostSender — connection is idle in pool.
    let resp = client.clone().oneshot(req).await.expect("request ok");
    resp.into_body().collect().await.unwrap();

    // Drop should free the idle connection immediately.
    drop(client);
    assert_metric!(ctx, "http.client.connection.duration");
}

// --- H1: pool behaviour ---------------------------------------------------

/// Two sequential requests to the same host both succeed; cumulative
/// `request.duration` count = 2 is recorded correctly.
#[tokio::test]
async fn h1_pool_reuses_connection_for_same_host() {
    let ctx = integration_context();

    let addr = spawn_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let client = new_client(&default_config());

    for _ in 0..2 {
        let req = http::Request::builder()
            .method(http::Method::GET)
            .uri(format!("http://{addr}/"))
            .body(empty_body())
            .unwrap();
        let resp = client.clone().oneshot(req).await.expect("request ok");
        // Consume body to release the connection back to the pool.
        resp.into_body().collect().await.unwrap();
    }

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 1
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 2
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 2
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

// --- Chaos / fault-injection ----------------------------------------------

/// Cancelling a request mid-flight must not leak metric guards.
///
/// All RAII guards (`ActiveGuard`, `RecordDurationGuard`, `RecordValueGuard`)
/// must fire on `Drop` even when the request future is cancelled via
/// `tokio::time::timeout`. In particular:
///
/// - `active_requests` must return to 0.
/// - `request.duration` must record one observation (without `error.type`,
///   because the `match` arm that sets it never ran before cancellation).
///
/// `request.body.size` is NOT asserted here: when the future is cancelled,
/// the `CountingBody` may still be held by hyper's connection task and its
/// guard fires asynchronously, making the timing non-deterministic.
#[tokio::test]
async fn h1_request_cancelled_guards_fire() {
    // Drop request.body.size in addition to connection.duration: the CountingBody
    // may be dropped asynchronously from hyper's connection task after cancellation,
    // making its recording timing non-deterministic.
    let ctx = TelemetryContext::builder()
        .with_view(|instrument: &Instrument| match instrument.name() {
            "http.client.connection.duration" | "http.client.request.body.size" => Some(
                StreamBuilder::default()
                    .with_aggregation(Aggregation::Drop)
                    .build()
                    .unwrap(),
            ),
            _ => None,
        })
        .build();

    let addr = spawn_server(Router::new().route(
        "/",
        get(|| async {
            tokio::time::sleep(Duration::MAX).await;
            ""
        }),
    ))
    .await;

    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    // Timeout fires before the server responds — future is dropped.
    let _ = tokio::time::timeout(
        Duration::from_millis(100),
        new_client(&default_config()).oneshot(req),
    )
    .await;

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// When the server closes the connection while the client is sending the
/// request body, all RAII guards must still fire and `error.type: "_OTHER"`
/// must be recorded on the duration histogram.
///
/// `request.body.size` is excluded from the snapshot: the number of body bytes
/// polled before hyper detects the TCP close is OS-scheduling-dependent.
#[tokio::test]
async fn h1_server_closes_during_request_body() {
    let ctx = TelemetryContext::builder()
        .with_view(|instrument: &Instrument| match instrument.name() {
            "http.client.connection.duration" | "http.client.request.body.size" => Some(
                StreamBuilder::default()
                    .with_aggregation(Aggregation::Drop)
                    .build()
                    .unwrap(),
            ),
            _ => None,
        })
        .build();

    let addr = spawn_drop_server().await;

    // 64 KiB body — large enough that the client cannot send it before the
    // server closes the TCP connection.
    let body = bytes_body(Bytes::from(vec![0u8; 65536]));
    let req = http::Request::builder()
        .method(http::Method::POST)
        .uri(format!("http://{addr}/"))
        .body(body)
        .unwrap();

    let _ = new_client(&default_config()).oneshot(req).await;

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: POST
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              error.type: _OTHER
              http.request.method: POST
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// H1 configuration records `error.type: "_OTHER"` when the TCP connection is
/// refused. All RAII guards fire; `http.client.open_connections` is never
/// incremented because the `ConnectionStateGuard` is created only after a
/// successful TCP connection and HTTP handshake.
#[tokio::test]
async fn h1_connection_refused_records_error() {
    let ctx = TelemetryContext::new();

    let config = parse_config("connect_timeout: 1s");
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri("http://127.0.0.1:1/")
        .body(empty_body())
        .unwrap();

    let err = new_client(&config)
        .oneshot(req)
        .await
        .expect_err("should fail");
    assert!(matches!(
        err,
        apollo_http_client::HttpClientError::Transport { .. }
    ));

    assert_metrics_snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "1"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              error.type: _OTHER
              http.request.method: GET
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "1"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// When the server accepts the TCP connection but stalls during the TLS
/// handshake, `connect_timeout` fires and the request fails with
/// `error.type: "_OTHER"`. `http.client.open_connections` is never
/// incremented because the connection was never established.
#[tokio::test]
async fn connect_timeout_fires_during_tls_handshake() {
    let ctx = TelemetryContext::new();

    let addr = spawn_stall_server().await;
    let config = parse_config("connect_timeout: 100ms\ntls:\n  danger_accept_invalid_certs: true");
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("https://{addr}/"))
        .body(empty_body())
        .unwrap();

    let err = new_client(&config)
        .oneshot(req)
        .await
        .expect_err("should fail");
    assert!(matches!(
        err,
        apollo_http_client::HttpClientError::ConnectionTimeout
    ));

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              error.type: _OTHER
              http.request.method: GET
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// When the server accepts the TCP connection but immediately closes it before
/// sending any HTTP bytes, `error.type: "_OTHER"` is recorded. The connection
/// state guard does fire (the TCP handshake succeeded and a `ConnectionStateGuard`
/// was created), but both counters return to zero when the client is dropped.
#[tokio::test]
async fn h1_server_closes_before_response() {
    let ctx = integration_context();

    let addr = spawn_drop_server().await;
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    let err = new_client(&default_config())
        .oneshot(req)
        .await
        .expect_err("should fail");
    assert!(matches!(
        err,
        apollo_http_client::HttpClientError::Request { .. }
    ));

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              error.type: _OTHER
              http.request.method: GET
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// Spawns a raw TCP server that sends valid HTTP/1.1 response headers and a
/// partial chunked body, then closes the TCP connection without the terminal
/// `0\r\n\r\n` chunk.
///
/// The client receives a `200 OK` status from the headers but gets a
/// connection-close error when it tries to read the body to completion.
async fn spawn_headers_then_close_server() -> SocketAddr {
    use tokio::io::AsyncWriteExt as _;
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        while let Ok((mut socket, _)) = listener.accept().await {
            let _ = socket
                .write_all(
                    b"HTTP/1.1 200 OK\r\n\
                      content-type: text/plain\r\n\
                      transfer-encoding: chunked\r\n\
                      \r\n\
                      7\r\npartial\r\n",
                )
                .await;
            // Dropping `socket` closes the TCP connection without the terminal chunk.
        }
    });
    addr
}

/// When the server sends response headers and then closes the connection before
/// the body is complete, `request.duration` records the status code from the
/// headers and `response.body.size` records however many bytes arrived.
#[tokio::test]
async fn h1_server_drops_connection_after_headers() {
    let ctx = integration_context();

    let addr = spawn_headers_then_close_server().await;

    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{addr}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&default_config())
        .oneshot(req)
        .await
        .expect("headers received");
    // Consume the body to completion — fails because the terminal chunk never arrives.
    let _ = resp.into_body().collect().await;

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - 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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

#[tokio::test]
async fn h1_derives_host_header() {
    let _ctx = TelemetryContext::new();

    let addr = Arc::new(OnceLock::new());

    let router = Router::new().route(
        "/",
        get({
            let addr = addr.clone();
            async move |req: axum::http::Request<axum::body::Body>| {
                let addr = addr.get().expect("address should be populated");
                let host = req.headers().get("host").expect("should have host header");
                let host = host.to_str().expect("host header should be utf8");
                assert_eq!(
                    host,
                    format!("{addr}"),
                    "host header must be set to IP:PORT"
                );
                "hello world"
            }
        }),
    );
    addr.set(spawn_server(router).await).expect("only set once");

    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!(
            "http://{}/",
            addr.get().expect("just populated it")
        ))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&default_config())
        .oneshot(req)
        .await
        .expect("request ok");

    let body = resp.into_body().collect().await.expect("should read body");
    let body = String::from_utf8(body.to_bytes().to_vec()).expect("body should be utf8");
    assert_eq!(body, "hello world", "received the expected response");
}