apollo_http_client/protocol/
mod.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::sync::{Arc, Mutex};
4use std::time::Duration;
5use tokio::time::Instant;
6use tokio_util::sync::CancellationToken;
7
8use bytes::Bytes;
9use http_body_util::combinators::BoxBody;
10use hyper::body::Incoming;
11use tower::BoxError;
12
13pub(crate) use apollo_http_shared::method::normalize_method;
14pub(crate) use apollo_http_shared::protocol::{ProtocolVersionCell, version_str};
15
16use crate::error::HttpClientError;
17
18pub(crate) mod alpn;
19pub(crate) mod h1;
20pub(crate) mod h2;
21pub(crate) mod handshake;
22#[cfg(unix)]
23pub(crate) mod unix;
24
25/// Identifies a connection pool by the destination it serves.
26#[derive(Clone, Debug, Hash, Eq, PartialEq)]
27pub(crate) struct Origin {
28    pub(crate) scheme: http::uri::Scheme,
29    pub(crate) authority: http::uri::Authority,
30}
31
32impl Origin {
33    /// `server.address` for telemetry: the URI host for TCP connections, or the
34    /// decoded socket path for `unix://`. Falls back to [`INVALID_SOCKET_PATH`]
35    /// when the authority cannot be decoded or Unix sockets are not supported.
36    pub(crate) fn address(&self) -> String {
37        if self.scheme.as_str() == "unix" {
38            #[cfg(unix)]
39            return crate::protocol::unix::authority_to_str(&self.authority)
40                .unwrap_or_else(|_| INVALID_SOCKET_PATH.to_owned());
41            #[cfg(not(unix))]
42            return INVALID_SOCKET_PATH.to_owned();
43        }
44        self.authority.host().to_owned()
45    }
46
47    /// `server.port` for telemetry: the URI port, defaulting to 443 for HTTPS
48    /// and 80 for HTTP. Returns `None` for `unix://` connections.
49    pub(crate) fn port(&self) -> Option<i64> {
50        if self.scheme.as_str() == "unix" {
51            return None;
52        }
53        let port = self.authority.port_u16().map(i64::from).unwrap_or_else(|| {
54            if self.scheme == http::uri::Scheme::HTTPS {
55                443
56            } else {
57                80
58            }
59        });
60        Some(port)
61    }
62}
63
64/// Placeholder recorded as `server.address` when a `unix://` socket path cannot
65/// be decoded from the URI authority — either because the hex is malformed or
66/// because the build has no Unix decoder compiled in.
67pub(crate) const INVALID_SOCKET_PATH: &str = "<invalid socket path>";
68
69/// HTTP request body type for [`crate::HttpClient`].
70///
71/// Use `http_body_util` to construct bodies and call `.map_err(…).boxed()` to convert:
72///
73/// ```rust
74/// use apollo_http_client::HttpBody;
75/// use bytes::Bytes;
76/// use http_body_util::{BodyExt, Empty, Full};
77///
78/// // Empty body (e.g. for GET requests)
79/// let _: HttpBody = Empty::<Bytes>::new()
80///     .map_err(|e: std::convert::Infallible| match e {})
81///     .boxed();
82///
83/// // Fixed-size body (e.g. for POST)
84/// let _: HttpBody = Full::new(Bytes::from_static(b"hello"))
85///     .map_err(|e: std::convert::Infallible| match e {})
86///     .boxed();
87/// ```
88pub type HttpBody = BoxBody<Bytes, BoxError>;
89
90// ---- Host sender types ------------------------------------------------------
91
92/// The future returned when a [`HostSender`] dispatches a request.
93pub(crate) type SendFuture = Pin<
94    Box<dyn Future<Output = Result<http::Response<Incoming>, HttpClientError>> + Send + 'static>,
95>;
96
97/// Sends HTTP requests to a specific host via its connection pool.
98///
99/// One `HostSender` exists per `(scheme, authority)` pair, created on first use and cached
100/// for the lifetime of the client.
101pub(crate) type HostSender =
102    Arc<dyn Fn(http::Request<HttpBody>, ProtocolVersionCell) -> SendFuture + Send + Sync>;
103
104// ---- Pool eviction helpers --------------------------------------------------
105
106/// Spawns a background task that calls `evict` every `interval` to prune stale
107/// connections from the pool.
108///
109/// Returns a guard that, when dropped, cancels the task promptly rather than waiting
110/// for the next interval. The task also exits automatically when the pool is dropped.
111pub(super) fn spawn_eviction_task<C>(
112    cache: &Arc<Mutex<C>>,
113    interval: Duration,
114    mut evict: impl FnMut(&mut C, Instant) + Send + 'static,
115) -> tokio_util::sync::DropGuard
116where
117    C: Send + 'static,
118{
119    let cancel = CancellationToken::new();
120    tokio::spawn({
121        let weak = Arc::downgrade(cache);
122        let cancel = cancel.clone();
123        async move {
124            loop {
125                tokio::select! {
126                    _ = cancel.cancelled() => break,
127                    _ = tokio::time::sleep(interval) => {}
128                }
129                let Some(arc) = weak.upgrade() else { break };
130                evict(&mut arc.lock().unwrap(), Instant::now());
131            }
132        }
133    });
134    cancel.drop_guard()
135}
136
137// ---- TCP connect helper -----------------------------------------------------
138
139/// Builds a synthetic URI from `key` and dispatches it through `connector`
140/// under a connect timeout.
141///
142/// Maps timeouts to [`HttpClientError::ConnectionTimeout`] and any underlying
143/// connector error to [`HttpClientError::Transport`]. The "/" path is required
144/// to build a syntactically valid URI; the connector only uses scheme + authority.
145pub(crate) async fn connect_with_timeout<S, IO>(
146    mut connector: S,
147    origin: Origin,
148    timeout: std::time::Duration,
149) -> Result<IO, HttpClientError>
150where
151    S: tower::Service<http::Uri, Response = IO>,
152    S::Error: Into<BoxError>,
153{
154    let uri = http::Uri::builder()
155        .scheme(origin.scheme)
156        .authority(origin.authority)
157        .path_and_query("/")
158        .build()
159        .map_err(|source| HttpClientError::InvalidUri { source })?;
160    tokio::time::timeout(timeout, connector.call(uri))
161        .await
162        .map_err(|_| HttpClientError::ConnectionTimeout)?
163        .map_err(|e| HttpClientError::Transport { source: e.into() })
164}
165
166// ---- Error helper -----------------------------------------------------------
167
168/// Maps a [`BoxError`] emerging from a pool connector into [`HttpClientError`],
169/// preserving any structured variant that was boxed into it (e.g.
170/// [`HttpClientError::ProxyTunnel`], [`HttpClientError::ConnectionTimeout`]).
171/// Errors that did not originate as an `HttpClientError` are wrapped as
172/// [`HttpClientError::Transport`].
173pub(crate) fn into_http_client_error(e: BoxError) -> HttpClientError {
174    match e.downcast::<HttpClientError>() {
175        Ok(http_err) => *http_err,
176        Err(other) => HttpClientError::Transport { source: other },
177    }
178}
179
180// ---- Pool retain helper -----------------------------------------------------
181
182/// Returns whether a connection should be retained in the pool.
183///
184/// An idle connection is kept if its age is within `idle_timeout` and the running
185/// `idle_count` is within `max_idle`. Active connections (no `idle_since`) are always kept.
186pub(crate) fn retain_predicate(
187    idle_since: Option<Instant>,
188    idle_timeout: Duration,
189    now: Instant,
190    idle_count: &mut usize,
191    max_idle: usize,
192) -> bool {
193    match idle_since {
194        Some(t) => {
195            if now.duration_since(t) >= idle_timeout {
196                return false;
197            }
198            *idle_count += 1;
199            *idle_count <= max_idle
200        }
201        None => true,
202    }
203}
204
205#[cfg(test)]
206mod tests {
207    use std::time::Duration;
208
209    use super::{HttpClientError, into_http_client_error, retain_predicate};
210
211    #[test]
212    fn into_http_client_error_unwraps_boxed_proxy_tunnel() {
213        let original = HttpClientError::ProxyTunnel {
214            status: http::StatusCode::PROXY_AUTHENTICATION_REQUIRED,
215        };
216        let boxed: tower::BoxError = Box::new(original);
217
218        let err = into_http_client_error(boxed);
219        assert!(
220            matches!(
221                err,
222                HttpClientError::ProxyTunnel {
223                    status: http::StatusCode::PROXY_AUTHENTICATION_REQUIRED,
224                }
225            ),
226            "expected ProxyTunnel, got {err}"
227        );
228    }
229
230    #[test]
231    fn into_http_client_error_unwraps_boxed_connection_timeout() {
232        let boxed: tower::BoxError = Box::new(HttpClientError::ConnectionTimeout);
233
234        let err = into_http_client_error(boxed);
235        assert!(
236            matches!(err, HttpClientError::ConnectionTimeout),
237            "expected ConnectionTimeout, got {err}"
238        );
239    }
240
241    #[test]
242    fn into_http_client_error_wraps_unstructured_box_error_as_transport() {
243        let boxed: tower::BoxError = "io error".into();
244
245        let err = into_http_client_error(boxed);
246        assert!(
247            matches!(err, HttpClientError::Transport { .. }),
248            "expected Transport, got {err}"
249        );
250    }
251
252    /// An active connection (idle_since = None) is always retained and does not
253    /// count toward the idle limit.
254    #[tokio::test(start_paused = true)]
255    async fn active_connection_always_retained() {
256        let now = tokio::time::Instant::now();
257        let mut count = 0;
258        assert!(retain_predicate(
259            None,
260            Duration::from_secs(1),
261            now,
262            &mut count,
263            1
264        ));
265        assert_eq!(count, 0);
266    }
267
268    /// An idle connection younger than the timeout is retained.
269    #[tokio::test(start_paused = true)]
270    async fn idle_within_timeout_retained() {
271        let idle_since = tokio::time::Instant::now();
272        tokio::time::advance(Duration::from_secs(4)).await;
273        let now = tokio::time::Instant::now();
274        let mut count = 0;
275        assert!(retain_predicate(
276            Some(idle_since),
277            Duration::from_secs(5),
278            now,
279            &mut count,
280            1
281        ));
282        assert_eq!(count, 1);
283    }
284
285    /// An idle connection that has exceeded the timeout is evicted.
286    #[tokio::test(start_paused = true)]
287    async fn idle_past_timeout_evicted() {
288        let idle_since = tokio::time::Instant::now();
289        tokio::time::advance(Duration::from_secs(6)).await;
290        let now = tokio::time::Instant::now();
291        let mut count = 0;
292        assert!(!retain_predicate(
293            Some(idle_since),
294            Duration::from_secs(5),
295            now,
296            &mut count,
297            1
298        ));
299    }
300
301    /// The idle count limit evicts connections beyond max_idle, even if they are
302    /// within the timeout.
303    #[tokio::test(start_paused = true)]
304    async fn idle_count_limit_evicts_excess() {
305        let idle_since = tokio::time::Instant::now();
306        let now = tokio::time::Instant::now();
307        let mut count = 0;
308        assert!(retain_predicate(
309            Some(idle_since),
310            Duration::from_secs(60),
311            now,
312            &mut count,
313            2
314        ));
315        assert!(retain_predicate(
316            Some(idle_since),
317            Duration::from_secs(60),
318            now,
319            &mut count,
320            2
321        ));
322        assert!(!retain_predicate(
323            Some(idle_since),
324            Duration::from_secs(60),
325            now,
326            &mut count,
327            2
328        ));
329    }
330}
apollo_http_client/protocol/mod.rs

apollo_http_client/protocol/
mod.rs
