praxis-proxy-core 0.7.5

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 Praxis Contributors

//! Internal connector types and metric constants for sub-request execution.
//!
//! Provides [`SubRequestConnector`], which wraps the build's egress
//! [`Connector`] with circuit breaker integration and bounded concurrency
//! via semaphores. Defines metric name constants shared across buffered
//! and streaming execution paths.
//!
//! [`Connector`]: crate::egress::Connector

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

use http::HeaderMap;
use metrics::{counter, histogram};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tracing::debug;

use super::types::{SubRequestError, SubResponse};
use crate::{
    backend,
    circuit::{CircuitBreakerConfig, CircuitBreakerRegistry, CircuitToken, PeerKey},
    egress::{Connector, EgressError},
};

// -----------------------------------------------------------------------------
// Metric names
// -----------------------------------------------------------------------------

/// Metric name for total streaming sub-request count (with termination label).
pub(super) const SUBREQUEST_STREAMS_TOTAL: &str = "praxis_subrequest_streams_total";
/// Metric name for streaming sub-request duration histogram.
pub(super) const SUBREQUEST_STREAM_DURATION_SECONDS: &str = "praxis_subrequest_stream_duration_seconds";
/// Metric name for streaming sub-request total bytes counter.
pub(super) const SUBREQUEST_STREAM_BYTES_TOTAL: &str = "praxis_subrequest_stream_bytes_total";
/// Metric name for header phase duration (shared buffered/streaming).
pub(super) const SUBREQUEST_HEADER_DURATION_SECONDS: &str = "praxis_subrequest_header_duration_seconds";

// -----------------------------------------------------------------------------
// SubRequestConnectorOptions
// -----------------------------------------------------------------------------

/// Options for constructing a [`SubRequestConnector`].
#[derive(Debug)]
pub struct SubRequestConnectorOptions {
    /// Number of idle connections to keep in the pool.
    pub keepalive_pool_size: usize,

    /// Maximum number of concurrently active exchanges.
    pub max_connections: Option<usize>,

    /// Circuit breaker configuration for peer-level failure tracking.
    pub circuit_breaker: Option<CircuitBreakerConfig>,
}

// -----------------------------------------------------------------------------
// SubRequestConnector
// -----------------------------------------------------------------------------

/// Shared HTTP connector for sub-requests.
///
/// Wraps the backend's egress [`Connector`] so that all filter instances
/// share a single connection pool. Created once at server startup and
/// passed through unchanged on config reload.
///
/// An optional admission semaphore limits the number of concurrently
/// active sub-request exchanges.
///
/// ```
/// use praxis_core::subrequest::SubRequestConnector;
/// praxis_tls::provider::install(); // required before any connector is built
///
/// let connector = SubRequestConnector::new(128, None);
/// let _clone = connector.clone();
/// ```
#[derive(Clone)]
pub struct SubRequestConnector<C: Connector = backend::Connector> {
    /// The backend's connector and pool. Shared, not cloned, so that its
    /// address tells callers whether two holders dial through one pool.
    pub(super) inner: Arc<C>,

    /// Admission semaphore bounding concurrently active exchanges.
    pub(super) admission: Option<Arc<Semaphore>>,

    /// The configured concurrency limit, retained for error reporting.
    pub(super) configured_max_connections: Option<usize>,

    /// Per-peer circuit breaker registry.
    pub(super) circuit_breakers: Option<Arc<CircuitBreakerRegistry>>,
}

impl SubRequestConnector {
    /// Create a connector on the build's backend with the given keepalive
    /// pool size and optional active-connection limit.
    ///
    /// ```
    /// use praxis_core::subrequest::SubRequestConnector;
    /// praxis_tls::provider::install(); // required before any connector is built
    ///
    /// let connector = SubRequestConnector::new(64, None);
    /// let bounded = SubRequestConnector::new(64, Some(256));
    /// ```
    pub fn new(keepalive_pool_size: usize, max_connections: Option<usize>) -> Self {
        Self::with_connector(backend::Connector::new(keepalive_pool_size), max_connections, None)
    }

    /// Create a connector on the build's backend from [`SubRequestConnectorOptions`].
    ///
    /// ```
    /// use praxis_core::subrequest::{SubRequestConnector, SubRequestConnectorOptions};
    /// praxis_tls::provider::install(); // required before any connector is built
    ///
    /// let connector = SubRequestConnector::with_options(SubRequestConnectorOptions {
    ///     keepalive_pool_size: 64,
    ///     max_connections: Some(256),
    ///     circuit_breaker: None,
    /// });
    /// ```
    pub fn with_options(opts: SubRequestConnectorOptions) -> Self {
        Self::with_connector(
            backend::Connector::new(opts.keepalive_pool_size),
            opts.max_connections,
            opts.circuit_breaker,
        )
    }
}

impl<C: Connector> SubRequestConnector<C> {
    /// Wrap an already-built egress connector.
    ///
    /// The build's backend is the usual choice ([`new`] and
    /// [`with_options`]); this is for a caller that supplies its own, such as
    /// a test double.
    ///
    /// ```
    /// use praxis_core::{
    ///     egress::{Connector as _, Unavailable},
    ///     subrequest::SubRequestConnector,
    /// };
    ///
    /// let connector = SubRequestConnector::with_connector(Unavailable::new(8), Some(4), None);
    /// assert_eq!(connector.configured_max_connections(), Some(4));
    /// ```
    ///
    /// [`new`]: SubRequestConnector::new
    /// [`with_options`]: SubRequestConnector::with_options
    pub fn with_connector(
        inner: C,
        max_connections: Option<usize>,
        circuit_breaker: Option<CircuitBreakerConfig>,
    ) -> Self {
        Self {
            inner: Arc::new(inner),
            admission: max_connections.map(|limit| Arc::new(Semaphore::new(limit))),
            configured_max_connections: max_connections,
            circuit_breakers: circuit_breaker.map(|cfg| Arc::new(CircuitBreakerRegistry::new(cfg))),
        }
    }

    /// Access the underlying egress [`Connector`].
    pub fn connector(&self) -> &C {
        &self.inner
    }

    /// Whether a sub-request circuit breaker registry is wired.
    ///
    /// `true` only when the connector was built (via [`with_options`]) with a
    /// circuit-breaker config; [`new`] never wires one. Exposed so callers and
    /// tests can assert that a configured `runtime.subrequest_circuit_breaker`
    /// actually reached the connector.
    ///
    /// [`with_options`]: Self::with_options
    /// [`new`]: Self::new
    #[must_use]
    pub fn has_circuit_breaker(&self) -> bool {
        self.circuit_breakers.is_some()
    }

    /// The configured max-connections admission limit, if any.
    #[must_use]
    pub fn configured_max_connections(&self) -> Option<usize> {
        self.configured_max_connections
    }

    /// Acquire an admission permit if a concurrency limit is
    /// configured. Returns `None` when no limit is set.
    ///
    /// The returned permit must be held for the entire sub-request
    /// exchange. Dropping it releases the slot.
    pub async fn acquire_permit(&self) -> Option<OwnedSemaphorePermit> {
        let semaphore = self.admission.as_ref()?;
        Arc::clone(semaphore).acquire_owned().await.ok()
    }

    /// Try to acquire an admission permit within the given deadline.
    ///
    /// Returns `Ok(Some(permit))` if acquired, `Ok(None)` if no
    /// concurrency limit is configured, or `Err` if the deadline
    /// expires before a slot opens.
    ///
    /// # Errors
    ///
    /// Returns [`SubRequestError::AdmissionTimeout`] when the
    /// semaphore cannot be acquired within the timeout.
    pub async fn try_acquire_permit(&self, timeout: Duration) -> Result<Option<OwnedSemaphorePermit>, SubRequestError> {
        let Some(semaphore) = self.admission.as_ref() else {
            return Ok(None);
        };
        let configured = self.configured_max_connections.unwrap_or(0);
        match tokio::time::timeout(timeout, Arc::clone(semaphore).acquire_owned()).await {
            Ok(Ok(permit)) => Ok(Some(permit)),
            Ok(Err(_closed)) => Err(SubRequestError::AdmissionTimeout {
                max_connections: configured,
            }),
            Err(_elapsed) => Err(SubRequestError::AdmissionTimeout {
                max_connections: configured,
            }),
        }
    }
}

impl<C: Connector> std::fmt::Debug for SubRequestConnector<C> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SubRequestConnector")
            .field("pool", &std::any::type_name::<C>())
            .field("max_connections", &self.configured_max_connections)
            .field("circuit_breakers", &self.circuit_breakers.is_some())
            .finish()
    }
}

// -----------------------------------------------------------------------------
// RawExchange
// -----------------------------------------------------------------------------

/// Live HTTP exchange after validated response headers.
///
/// Returned by `open_exchange()` and consumed by either
/// `execute()` (buffered collection) or `send_streaming()`
/// (body ownership handoff).
pub(super) struct RawExchange<'conn, 'reg, C: Connector> {
    /// Live transport session.
    pub(super) session: C::Session,
    /// The peer's operator-configured per-read timeout.
    pub(super) read_timeout: Option<Duration>,
    /// Connector ref for session release.
    pub(super) connector: &'conn SubRequestConnector<C>,
    /// HTTP status code.
    pub(super) status: u16,
    /// Protocol version of the upstream response.
    pub(super) version: http::Version,
    /// Sanitized response headers.
    pub(super) headers: HeaderMap,
    /// Optional circuit breaker guard.
    pub(super) circuit_guard: Option<CircuitGuard<'reg>>,
    /// Admission permit.
    pub(super) permit: Option<OwnedSemaphorePermit>,
    /// Absolute deadline for the entire exchange.
    pub(super) deadline: tokio::time::Instant,
    /// Per-attempt HTTP client span retained through response-body completion.
    pub(super) client_span: tracing::Span,
}

// -----------------------------------------------------------------------------
// Circuit Breaker Guard
// -----------------------------------------------------------------------------

/// RAII guard ensuring every acquired circuit token is finalized.
///
/// On drop without explicit [`finalize`](Self::finalize) or [`fail`](Self::fail),
/// records a failure. That covers deadline exits, panics, and any early-return
/// path after token acquisition. [`abandon_as_neutral`](Self::abandon_as_neutral)
/// changes that drop into a release: the in-flight slot is freed and neither
/// success nor failure is recorded. Callers that set the flag still call
/// [`fail`](Self::fail) on a real error so only an aborted attempt is neutral.
pub(super) struct CircuitGuard<'reg> {
    /// The registry that issued the token.
    registry: &'reg CircuitBreakerRegistry,
    /// Logical peer identity the token was acquired for.
    peer: PeerKey,
    /// The generation token; `None` after finalization.
    token: Option<CircuitToken>,
    /// Drop releases the slot instead of recording a failure.
    neutral_on_drop: bool,
}

impl<'reg> CircuitGuard<'reg> {
    /// Create a guard from an acquired token.
    pub(super) fn new(registry: &'reg CircuitBreakerRegistry, peer: PeerKey, token: CircuitToken) -> Self {
        Self {
            registry,
            peer,
            token: Some(token),
            neutral_on_drop: false,
        }
    }

    /// A later drop releases the slot without a success or a failure.
    pub(super) fn abandon_as_neutral(&mut self) {
        self.neutral_on_drop = true;
    }

    /// Record this error and consume the token.
    ///
    /// Connect, I/O, and deadline errors are failures. Every other error
    /// is a success, matching [`finalize`](Self::finalize).
    pub(super) fn fail(mut self, err: &SubRequestError) {
        let Some(token) = self.token.take() else {
            return;
        };
        if matches!(
            err,
            SubRequestError::Connect(_) | SubRequestError::Io(_) | SubRequestError::DeadlineExceeded
        ) {
            self.registry.record_failure(&self.peer, token);
        } else {
            self.registry.record_success(&self.peer, token);
        }
    }

    /// Finalize the guard as a success regardless of later body outcome.
    pub(super) fn finalize_success(mut self) {
        if let Some(token) = self.token.take() {
            self.registry.record_success(&self.peer, token);
        }
    }

    /// Finalize the guard with the actual exchange outcome.
    pub(super) fn finalize(mut self, result: &Result<SubResponse, SubRequestError>) {
        let Some(token) = self.token.take() else {
            return;
        };
        match result {
            Err(SubRequestError::Connect(_) | SubRequestError::Io(_) | SubRequestError::DeadlineExceeded) => {
                self.registry.record_failure(&self.peer, token);
            },
            Ok(_) | Err(_) => {
                self.registry.record_success(&self.peer, token);
            },
        }
    }
}

impl Drop for CircuitGuard<'_> {
    fn drop(&mut self) {
        if let Some(token) = self.token.take() {
            if self.neutral_on_drop {
                self.registry.release(&self.peer, token);
            } else {
                self.registry.record_failure(&self.peer, token);
            }
        }
    }
}

// -----------------------------------------------------------------------------
// Header-phase termination
// -----------------------------------------------------------------------------

/// Record metrics and trace for header-phase stream termination.
///
/// Used by `send_streaming()` when the response completes at header
/// time (HEAD, 204, 304, zero-length, or H2 error).
pub(super) fn record_header_termination(termination: &'static str) {
    counter!(SUBREQUEST_STREAMS_TOTAL, "termination" => termination).increment(1);
    histogram!(SUBREQUEST_STREAM_DURATION_SECONDS).record(0.0);
    debug!(termination, "sub-request: stream terminated at header phase");
}

// -----------------------------------------------------------------------------
// Header sanitization
// -----------------------------------------------------------------------------

/// Headers that apply only to one HTTP connection and must not be forwarded
/// across a sub-request boundary. Re-exported from the canonical
/// [`crate::reserved_headers::HOP_BY_HOP_HEADERS`] so the sub-request and
/// protocol paths share one source of truth.
pub(super) use crate::reserved_headers::HOP_BY_HOP_HEADERS;

/// Collect the `Connection`-nominated header names, borrowed from the
/// map's own `Connection` values.
pub(super) fn connection_nominated_tokens(headers: &HeaderMap) -> Vec<&str> {
    headers
        .get_all(http::header::CONNECTION)
        .iter()
        .flat_map(crate::reserved_headers::connection_tokens)
        .collect()
}

/// Whether a header must not cross the sub-request boundary in either
/// direction: hop-by-hop (the fixed list or `Connection`-nominated) or
/// a reserved internal prefix (`x-praxis-*`, `x-ext-protocol-*`,
/// `x-ext-agent-*`).
pub(super) fn is_boundary_stripped(name: &http::header::HeaderName, nominated: &[&str]) -> bool {
    // `HeaderName::as_str` is always lowercase, so the fixed list needs
    // no case folding; nominated tokens arrive raw from the wire.
    let name = name.as_str();
    // A client-supplied `Connection` token must not delete a proxy-owned
    // forwarding header (x-forwarded-*, Forwarded, Host, Content-Length): the
    // main upstream path and filtered sub-requests already refuse this, so the
    // core sub-request path honors the same rule. Reserved and fixed hop-by-hop
    // headers are still stripped via their own branches.
    HOP_BY_HOP_HEADERS.contains(&name)
        || crate::reserved_headers::is_reserved(name)
        || (nominated.iter().any(|token| token.eq_ignore_ascii_case(name))
            && !crate::reserved_headers::is_connection_token_protected(name))
}

/// Request-direction predicate: boundary-stripped plus the framing
/// headers the executor re-computes.
pub(super) fn is_request_stripped(name: &http::header::HeaderName, nominated: &[&str]) -> bool {
    let lower = name.as_str();
    lower == "content-length" || lower == "transfer-encoding" || is_boundary_stripped(name, nominated)
}

// -----------------------------------------------------------------------------
// Utilities
// -----------------------------------------------------------------------------

/// Whether a header is a transport-level header that must not be
/// injected via framework metadata.
pub(super) fn is_transport_header(name: &http::header::HeaderName) -> bool {
    HOP_BY_HOP_HEADERS.iter().any(|header| *header == name.as_str()) || name == http::header::CONTENT_LENGTH
}

/// Methods whose empty payload is commonly rejected without explicit framing.
pub(super) fn empty_body_needs_framing(method: &http::Method) -> bool {
    matches!(*method, http::Method::POST | http::Method::PUT | http::Method::PATCH)
}

/// Ensure HTTP/1.1 virtual hosting and HTTP/2 `:authority` are valid by
/// defaulting a missing `Host` to the peer's socket address.
pub(super) fn ensure_host_header(headers: &mut HeaderMap, address: SocketAddr) -> Result<(), SubRequestError> {
    if !headers.contains_key(http::header::HOST) {
        let host = http::HeaderValue::from_str(&address.to_string())
            .map_err(|error| SubRequestError::InvalidRequest(error.to_string()))?;
        headers.insert(http::header::HOST, host);
    }
    Ok(())
}

/// Map a transport failure onto the sub-request error it means.
///
/// Message text is carried through unchanged so callers keep seeing the
/// transport's own diagnostics.
pub(super) fn transport_error(err: EgressError) -> SubRequestError {
    match err {
        EgressError::InvalidRequest(message) => SubRequestError::InvalidRequest(message),
        EgressError::Connect(message) => SubRequestError::Connect(message),
        EgressError::Io(message) => SubRequestError::Io(message),
        unavailable @ EgressError::Unavailable => SubRequestError::Connect(unavailable.to_string()),
    }
}

/// Keep an operator-configured timeout when it is stricter than the deadline.
pub(super) fn min_timeout(configured: Option<Duration>, deadline: Duration) -> Duration {
    configured.map_or(deadline, |configured| configured.min(deadline))
}

/// Classify a timeout expiry as either `DeadlineExceeded` (the overall
/// request deadline fired) or `Io` (a shorter operator-configured
/// read/write timeout fired).  Call this inside the `Err(_elapsed)`
/// arm of `tokio::time::timeout` to preserve the 502-vs-504 distinction.
///
/// Uses the pre-computed budget and configured timeout rather than a
/// post-hoc `Instant::now() >= deadline` check, which races with
/// scheduler jitter when the configured timeout equals the remaining
/// budget.
pub(super) fn classify_timeout(
    remaining_budget: Duration,
    configured_timeout: Option<Duration>,
    phase: &str,
) -> SubRequestError {
    if configured_timeout.is_none_or(|timeout| timeout >= remaining_budget) {
        SubRequestError::DeadlineExceeded
    } else {
        SubRequestError::Io(format!("upstream {phase} timeout"))
    }
}