apollo-http-client 0.3.0

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

#[macro_use]
mod common;

use std::net::SocketAddr;
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;
use bytes::Bytes;
use http_body_util::{BodyExt as _, Full};
use hyper_util::rt::{TokioExecutor, TokioIo};
use indoc::indoc;
use opentelemetry_sdk::metrics::{Aggregation, Instrument, StreamBuilder};
use tower::ServiceExt as _;

use common::*;

// --- H2: infrastructure ---------------------------------------------------

/// Spawns a local H2c (HTTP/2 cleartext, no TLS) server.
///
/// Unlike [`spawn_server`], this uses `hyper::server::conn::http2` directly so
/// that the client can connect with H2 prior-knowledge without TLS negotiation.
async fn spawn_h2c_server(router: Router) -> SocketAddr {
    use hyper::server::conn::http2;
    use hyper_util::service::TowerToHyperService;

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    tokio::spawn(async move {
        loop {
            let Ok((stream, _)) = listener.accept().await else {
                break;
            };
            let io = TokioIo::new(stream);
            let svc = TowerToHyperService::new(router.clone());
            tokio::spawn(async move {
                http2::Builder::new(TokioExecutor::new())
                    .serve_connection(io, svc)
                    .await
                    .ok();
            });
        }
    });

    addr
}

fn h2_config() -> HttpClientConfig {
    parse_config("protocol: http2")
}

/// Returns a [`TelemetryContext`] that records only `http.client.open_connections`.
///
/// Used in H2 connection-state tests to keep snapshots focused without the noise
/// of request-level metrics. `http.client.connection.duration` is tested
/// separately in [`h2_connection_duration_is_recorded`].
fn h2_connections_context() -> TelemetryContext {
    TelemetryContext::builder()
        .with_view(|instrument: &Instrument| {
            if instrument.name() != "http.client.open_connections" {
                Some(
                    StreamBuilder::default()
                        .with_aggregation(Aggregation::Drop)
                        .build()
                        .unwrap(),
                )
            } else {
                None
            }
        })
        .build()
}

// --- H2: error path -------------------------------------------------------

/// H2 configuration does not panic on a connection-refused error; `error.type: "_OTHER"`
/// is recorded on `http.client.request.duration`.
#[tokio::test]
async fn h2_connection_refused_records_error() {
    let ctx = TelemetryContext::new();

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

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

    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: "2"
              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 an H2c server accepts the TCP connection but immediately closes it,
/// the HTTP/2 handshake succeeds (the client sends its preface before the FIN
/// arrives) but the subsequent request fails. `error.type: "_OTHER"` is
/// recorded. `http.client.open_connections` is incremented then returned to
/// zero on connection close, and `connection.duration` is emitted.
#[tokio::test]
async fn h2_server_closes_before_handshake() {
    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(&h2_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: "2"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "2"
              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: "2"
              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 H2 connection while a request is in-flight, the
/// request fails with `error.type: "_OTHER"`. Open connection counters return
/// to zero and all RAII metric guards fire correctly.
#[tokio::test]
async fn h2_server_closes_connection_during_request() {
    use hyper::server::conn::http2;
    use hyper_util::service::TowerToHyperService;

    let ctx = integration_context();

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    tokio::spawn(async move {
        let Ok((stream, _)) = listener.accept().await else {
            return;
        };
        let io = TokioIo::new(stream);
        // Handler sleeps forever; connection is dropped from the outside after
        // the request arrives.
        let svc = TowerToHyperService::new(Router::new().route(
            "/",
            get(|| async {
                tokio::time::sleep(Duration::MAX).await;
                ""
            }),
        ));
        let conn = http2::Builder::new(TokioExecutor::new()).serve_connection(io, svc);
        tokio::pin!(conn);
        // Drive the connection long enough for the request to arrive, then
        // close it. The client's in-flight request will get a stream or
        // connection error.
        let _ = tokio::time::timeout(Duration::from_millis(50), &mut conn).await;
        // conn is dropped here, closing the TCP connection.
    });

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

    let err = new_client(&h2_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: "2"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "2"
              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: "2"
              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
    "#);
}

// --- H2: success path -----------------------------------------------------

/// A successful H2 GET records all metrics with `network.protocol.version: "2"`,
/// including body size metrics (enabled explicitly).
#[tokio::test]
async fn h2_successful_get_records_all_metrics() {
    let ctx = integration_context();

    let addr = spawn_h2c_server(Router::new().route("/", get(|| async { "hello" }))).await;
    let config = parse_config(indoc! {"
        protocol: http2
        telemetry:
          metrics:
            request_body_size: true
            response_body_size: true
            url_scheme: true
    "});
    let client = new_client(&config);
    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");
    resp.into_body().collect().await.unwrap();
    drop(client);

    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: "2"
              server.address: 127.0.0.1
              server.port: "<port>"
              url.scheme: http
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "2"
              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: "2"
              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: "2"
              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: "2"
              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 404 response over H2 records `http.response.status_code: 404` and
/// `error.type: "404"` on the request duration histogram, with
/// `network.protocol.version: "2"`.
#[tokio::test]
async fn h2_4xx_response_records_status_and_error_type() {
    let ctx = integration_context();

    let addr =
        spawn_h2c_server(Router::new().route("/", get(|| async { http::StatusCode::NOT_FOUND })))
            .await;
    let client = new_client(&h2_config());
    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");
    assert_eq!(resp.status(), http::StatusCode::NOT_FOUND);
    drop(resp);
    drop(client);

    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: "2"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "2"
              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: "2"
              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
    "#);
}

// --- H2: connection state -------------------------------------------------

/// After completed requests the H2 connection stays `idle: 1` as long as the
/// client is alive (the singleton holds it open for reuse), dropping to
/// `idle: 0` only when the client is dropped.
///
/// This contrasts with H1, where pooled connections sit in `active` state
/// between requests (idle is only set during `poll_ready`).
#[tokio::test]
async fn h2_connection_stays_idle_between_requests() {
    let ctx = h2_connections_context();

    let addr = spawn_h2c_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let client = new_client(&h2_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");
        resp.into_body().collect().await.unwrap();
    }

    // Connection is held open by the singleton — idle while no requests are in flight.
    snapshot!(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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "2"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 1
        is_monotonic: false
        temporality: Cumulative
    "#);

    drop(client);

    // Connection released when the client (and its singleton) is dropped.
    snapshot!(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: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.protocol.version: "2"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    "#);
}

/// `http.client.connection.duration` is recorded for H2 connections with
/// `network.protocol.version: "2"`. Verified with `assert_metric!` because the
/// wall-clock sum is non-deterministic.
#[tokio::test]
async fn h2_connection_duration_is_recorded() {
    let ctx = TelemetryContext::new();

    let addr = spawn_h2c_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(&h2_config())
        .oneshot(req)
        .await
        .expect("request ok");

    assert_metric!(
        ctx,
        "http.client.connection.duration",
        "network.protocol.version" = "2"
    );
}

/// Idle H2 connections are freed immediately when the client is dropped.
/// See [`h1_connections_freed_immediately_on_client_drop`] for the rationale.
#[tokio::test]
async fn h2_connections_freed_immediately_on_client_drop() {
    let ctx = TelemetryContext::new();
    let addr = spawn_h2c_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let client = new_client(&h2_config());
    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");
    resp.into_body().collect().await.unwrap();

    drop(client);
    assert_metric!(
        ctx,
        "http.client.connection.duration",
        "network.protocol.version" = "2"
    );
}

/// HTTP/2 doesn't use the Host header, but an :authority pseudo-header.
/// It's tempting to put the Host derivation at the front of the HttpClient, but that's not where it
/// belongs, and this test makes sure that we apply it to HTTP/1.1 only.
#[tokio::test]
async fn h2_does_not_derive_host_header() {
    let _ctx = TelemetryContext::new();

    let router = Router::new().route(
        "/",
        get(async move |req: axum::http::Request<axum::body::Body>| {
            assert_eq!(
                req.headers().get("host"),
                None,
                "should not have host header in http/2"
            );
            "hello world"
        }),
    );
    let addr = spawn_h2c_server(router).await;

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

    let resp = client.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");
}