apollo-http-client 0.3.0

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

use apollo_opentelemetry::metrics::{
    FutureExt, HistogramExt, RecordDurationGuard, TrackGuard, UpDownCounterExt,
};
use hyper::client::conn::http1;
use hyper_rustls::HttpsConnector;
use hyper_util::client::legacy::connect::HttpConnector;
use hyper_util::client::pool;
use hyper_util::rt::TokioExecutor;
use opentelemetry::KeyValue;
use opentelemetry::metrics::UpDownCounter;
use opentelemetry_semantic_conventions::attribute as semconv;
use tower::BoxError;
use tower::Service;
use tower::ServiceExt as _;

use super::handshake::{Http1Handshake, TcpConnect};
use super::{
    HostSender, HttpBody, Origin, ProtocolVersionCell, into_http_client_error, retain_predicate,
    spawn_eviction_task,
};
use crate::error::HttpClientError;
use crate::metrics::{ConnectionMetrics, STATE_ACTIVE, STATE_IDLE};

// ---- Connection state -------------------------------------------------------

/// Tracks the `http.client.open_connections` state for an HTTP/1.1 connection.
///
/// Starts active on construction. Transitions between active and idle as requests
/// start and complete. Decrements the current state counter on drop.
struct ConnectionState {
    guard: TrackGuard<i64>,
    idle_since: Option<Instant>,
}

impl ConnectionState {
    fn new(counter: &UpDownCounter<i64>, mut base_attrs: Vec<KeyValue>) -> Self {
        base_attrs.push(KeyValue::new(semconv::HTTP_CONNECTION_STATE, STATE_ACTIVE));
        Self {
            guard: counter.track(base_attrs),
            idle_since: None,
        }
    }

    fn is_active(&self) -> bool {
        self.idle_since.is_none()
    }

    fn transition_to_idle(&mut self) {
        if self.is_active() {
            self.guard
                .set(KeyValue::new(semconv::HTTP_CONNECTION_STATE, STATE_IDLE));
            self.idle_since = Some(Instant::now());
        }
    }

    fn transition_to_active(&mut self) {
        if !self.is_active() {
            self.guard
                .set(KeyValue::new(semconv::HTTP_CONNECTION_STATE, STATE_ACTIVE));
            self.idle_since = None;
        }
    }

    fn idle_since(&self) -> Option<Instant> {
        self.idle_since
    }
}

/// Transitions the connection to active on construction and back to idle on drop,
/// covering the full lifetime of a single in-flight request.
struct RequestGuard(Arc<Mutex<ConnectionState>>);

impl RequestGuard {
    fn new(state: Arc<Mutex<ConnectionState>>) -> Self {
        state.lock().unwrap().transition_to_active();
        Self(state)
    }
}

impl Drop for RequestGuard {
    fn drop(&mut self) {
        self.0.lock().unwrap().transition_to_idle();
    }
}

// ---- Connection -------------------------------------------------------------

/// An HTTP/1.1 connection with instrumented open-connection and duration metrics.
pub(crate) struct H1Connection {
    inner: http1::SendRequest<HttpBody>,
    state: Arc<Mutex<ConnectionState>>,
    _connection_duration: RecordDurationGuard,
}

impl Unpin for H1Connection {}

impl H1Connection {
    pub(crate) fn new(
        sender: http1::SendRequest<HttpBody>,
        metrics: &ConnectionMetrics,
        base_attrs: Vec<KeyValue>,
    ) -> Self {
        Self {
            inner: sender,
            state: Arc::new(Mutex::new(ConnectionState::new(
                metrics.counter(),
                base_attrs.clone(),
            ))),
            _connection_duration: metrics
                .duration_histogram()
                .record_duration_on_drop(base_attrs),
        }
    }

    /// Returns when the connection became idle, used by the pool to apply idle timeout
    /// and max-idle eviction. `None` means a request is currently in flight.
    pub(crate) fn idle_since(&self) -> Option<Instant> {
        self.state.lock().unwrap().idle_since()
    }
}

impl std::fmt::Debug for H1Connection {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("H1Connection").finish_non_exhaustive()
    }
}

impl Service<http::Request<HttpBody>> for H1Connection {
    type Response = http::Response<hyper::body::Incoming>;
    type Error = hyper::Error;
    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>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: http::Request<HttpBody>) -> Self::Future {
        // It's easier to always compute the host name, even if we aren't going to use it, to avoid
        // conflicting borrows.
        if let Some(host) = get_host_name(&req) {
            // If we already had a Host header, don't override it, the user might want to do
            // something funky
            req.headers_mut().entry(http::header::HOST).or_insert(host);
        }

        Box::pin(
            self.inner
                .send_request(req)
                .with_guard(RequestGuard::new(self.state.clone())),
        )
    }
}

// ---- Connector --------------------------------------------------------------

/// Establishes HTTP/1.1 connections to a given [`Origin`].
///
/// Each call performs a TCP/TLS handshake and returns an [`H1Connection`] ready for use.
#[derive(Clone)]
pub(crate) struct H1Connector {
    inner: HttpsConnector<HttpConnector>,
    metrics: ConnectionMetrics,
    connect_timeout: std::time::Duration,
}

impl H1Connector {
    pub(crate) fn new(
        inner: HttpsConnector<HttpConnector>,
        metrics: ConnectionMetrics,
        connect_timeout: std::time::Duration,
    ) -> Self {
        Self {
            inner,
            metrics,
            connect_timeout,
        }
    }

    /// Consumes the connector and returns its TLS, metrics, and timeout components.
    ///
    /// Used by the ALPN connector, which builds its own lower-level connection pipeline
    /// from these parts rather than using `H1Connector` as a service.
    pub(crate) fn into_parts(
        self,
    ) -> (
        HttpsConnector<HttpConnector>,
        ConnectionMetrics,
        std::time::Duration,
    ) {
        (self.inner, self.metrics, self.connect_timeout)
    }
}

impl Service<Origin> for H1Connector {
    type Response = H1Connection;
    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>> {
        self.inner.poll_ready(cx).map_err(BoxError::from)
    }

    fn call(&mut self, origin: Origin) -> Self::Future {
        let attrs = self
            .metrics
            .connection_attrs(Some(http::Version::HTTP_11), &origin);
        let clone = self.inner.clone();
        let inner = std::mem::replace(&mut self.inner, clone);
        let mut handshake = Http1Handshake::new(
            TcpConnect::new(inner, self.connect_timeout),
            self.metrics.clone(),
            attrs,
        );
        handshake.call(origin)
    }
}

fn get_host_name<B>(req: &http::Request<B>) -> Option<http::HeaderValue> {
    req.uri()
        .authority()
        .and_then(|authority| http::HeaderValue::from_str(authority.as_str()).ok())
}

// ---- Pool -------------------------------------------------------------------

/// Creates a [`HostSender`] backed by an HTTP/1.1 connection pool for the given host.
///
/// Accepts any [`Service`] producing an [`H1Connection`] keyed by [`Origin`], so both
/// direct connectors (e.g. [`H1Connector`]) and proxy-aware adapters can be wired up the
/// same way. `decorate_request` is called on every outgoing request before dispatch,
/// allowing callers to inject per-host headers (e.g. `Proxy-Authorization`).
///
/// Idle connections are evicted periodically based on `idle_timeout` and `max_idle`.
/// The pool and its eviction task are cleaned up when the sender is dropped.
pub(crate) fn new_sender<C, D>(
    connector: C,
    max_idle: usize,
    idle_timeout: Duration,
    origin: Origin,
    decorate_request: D,
) -> HostSender
where
    C: Service<Origin, Response = H1Connection> + Clone + Send + 'static,
    C::Error: Into<BoxError> + Send + 'static,
    C::Future: Send + Unpin + 'static,
    D: Fn(&mut http::Request<HttpBody>) + Send + Sync + 'static,
{
    let cache = Arc::new(Mutex::new(
        pool::cache::builder()
            .executor(TokioExecutor::new())
            .build(connector),
    ));
    let drop_guard = spawn_eviction_task(&cache, idle_timeout, move |c, now| {
        let mut idle_count = 0;
        c.retain(|conn| {
            retain_predicate(
                conn.idle_since(),
                idle_timeout,
                now,
                &mut idle_count,
                max_idle,
            )
        });
    });
    Arc::new(move |mut req, version: ProtocolVersionCell| {
        // Reference drop_guard so it is captured by this move closure and lives
        // as long as the Arc. Without this, it would not be captured and would
        // fire cancel() immediately after new_sender returns.
        let _guard = &drop_guard;
        let mut c = cache.lock().unwrap().clone();
        let k = origin.clone();
        decorate_request(&mut req);
        Box::pin(async move {
            version.set(http::Version::HTTP_11);
            // Backpressure is not propagated here: the pool selects a connection only at call time,
            // so there is no single connection whose readiness we could poll in advance. A new
            // connection also starts unready until established, which would cause spurious
            // not-ready signals. Meaningful backpressure would require pinning each caller to
            // exactly one target host.
            let mut conn = c
                .ready()
                .await
                .map_err(|e| into_http_client_error(e.into()))?
                .call(k)
                .await
                .map_err(|e| into_http_client_error(e.into()))?;
            let resp = conn
                .ready()
                .await
                .map_err(|source| HttpClientError::Request { source })?
                .call(req)
                .await
                .map_err(|source| HttpClientError::Request { source })?;
            Ok(resp)
        })
    })
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use apollo_opentelemetry_test::{TelemetryContext, assert_metrics_snapshot};
    use opentelemetry::KeyValue;
    use opentelemetry::global;

    use super::{ConnectionState, RequestGuard};

    fn make_counter() -> opentelemetry::metrics::UpDownCounter<i64> {
        global::meter_provider()
            .meter("test")
            .i64_up_down_counter("test.connections")
            .build()
    }

    fn base_attrs() -> Vec<KeyValue> {
        vec![KeyValue::new("server.address", "test-host")]
    }

    /// A new H1 connection immediately increments the active counter (connections
    /// start active, unlike H2 which starts idle).
    #[test]
    fn new_connection_is_active() {
        let ctx = TelemetryContext::new();
        let _state = ConnectionState::new(&make_counter(), base_attrs());
        assert_metrics_snapshot!(ctx, @r"
        - name: test.connections
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  server.address: test-host
                value: 1
            is_monotonic: false
            temporality: Cumulative
        ");
    }

    /// `transition_to_idle` moves the connection from active to idle.
    #[test]
    fn transition_to_idle_marks_idle() {
        let ctx = TelemetryContext::new();
        let mut state = ConnectionState::new(&make_counter(), base_attrs());
        state.transition_to_idle();
        assert_metrics_snapshot!(ctx, @r"
        - name: test.connections
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  server.address: test-host
                value: 0
              - attributes:
                  http.connection.state: idle
                  server.address: test-host
                value: 1
            is_monotonic: false
            temporality: Cumulative
        ");
    }

    /// `transition_to_active` after idle returns the connection to active.
    #[test]
    fn transition_to_active_after_idle() {
        let ctx = TelemetryContext::new();
        let mut state = ConnectionState::new(&make_counter(), base_attrs());
        state.transition_to_idle();
        state.transition_to_active();
        assert_metrics_snapshot!(ctx, @r"
        - name: test.connections
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  server.address: test-host
                value: 1
              - attributes:
                  http.connection.state: idle
                  server.address: test-host
                value: 0
            is_monotonic: false
            temporality: Cumulative
        ");
    }

    /// Calling `transition_to_idle` twice does not double-decrement the active counter.
    #[test]
    fn transition_to_idle_is_idempotent() {
        let ctx = TelemetryContext::new();
        let mut state = ConnectionState::new(&make_counter(), base_attrs());
        state.transition_to_idle();
        state.transition_to_idle();
        assert_metrics_snapshot!(ctx, @r"
        - name: test.connections
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  server.address: test-host
                value: 0
              - attributes:
                  http.connection.state: idle
                  server.address: test-host
                value: 1
            is_monotonic: false
            temporality: Cumulative
        ");
    }

    /// Calling `transition_to_active` twice does not double-decrement the idle counter.
    #[test]
    fn transition_to_active_is_idempotent() {
        let ctx = TelemetryContext::new();
        let mut state = ConnectionState::new(&make_counter(), base_attrs());
        state.transition_to_idle();
        state.transition_to_active();
        state.transition_to_active();
        assert_metrics_snapshot!(ctx, @r"
        - name: test.connections
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  server.address: test-host
                value: 1
              - attributes:
                  http.connection.state: idle
                  server.address: test-host
                value: 0
            is_monotonic: false
            temporality: Cumulative
        ");
    }

    /// The request guard transitions the connection to idle when dropped,
    /// mirroring what happens when the request future completes or is cancelled.
    #[test]
    fn request_guard_transitions_to_idle_on_drop() {
        let ctx = TelemetryContext::new();
        let state = Arc::new(Mutex::new(ConnectionState::new(
            &make_counter(),
            base_attrs(),
        )));
        state.lock().unwrap().transition_to_idle(); // connection idle after previous request
        let guard = RequestGuard::new(state.clone()); // new request starts: transitions to active
        drop(guard); // request completes or is cancelled: transitions to idle
        assert_metrics_snapshot!(ctx, @r"
        - name: test.connections
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  server.address: test-host
                value: 0
              - attributes:
                  http.connection.state: idle
                  server.address: test-host
                value: 1
            is_monotonic: false
            temporality: Cumulative
        ");
    }

    /// Dropping an active connection decrements the active counter.
    #[test]
    fn drop_active_decrements_active() {
        let ctx = TelemetryContext::new();
        let state = ConnectionState::new(&make_counter(), base_attrs());
        drop(state);
        assert_metrics_snapshot!(ctx, @r"
        - name: test.connections
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  server.address: test-host
                value: 0
            is_monotonic: false
            temporality: Cumulative
        ");
    }

    /// Dropping an idle connection decrements the idle counter, not the active counter.
    #[test]
    fn drop_idle_decrements_idle() {
        let ctx = TelemetryContext::new();
        let mut state = ConnectionState::new(&make_counter(), base_attrs());
        state.transition_to_idle();
        drop(state);
        assert_metrics_snapshot!(ctx, @r"
        - name: test.connections
          data:
            type: Sum
            data_points:
              - attributes:
                  http.connection.state: active
                  server.address: test-host
                value: 0
              - attributes:
                  http.connection.state: idle
                  server.address: test-host
                value: 0
            is_monotonic: false
            temporality: Cumulative
        ");
    }
}