saluki_io/net/client/http/
conn.rs

1use std::{
2    future::Future,
3    io,
4    pin::Pin,
5    task::{Context, Poll},
6    time::{Duration, Instant},
7};
8#[cfg(unix)]
9use std::{path::PathBuf, sync::Arc};
10
11use http::{Extensions, Uri};
12use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder, MaybeHttpsStream};
13use hyper_util::{
14    client::legacy::connect::{CaptureConnection, Connected, Connection, HttpConnector},
15    rt::TokioIo,
16};
17use metrics::Counter;
18use pin_project_lite::pin_project;
19use rustls::ClientConfig;
20use saluki_error::GenericError;
21use tokio::net::TcpStream;
22#[cfg(target_os = "linux")]
23use tokio_vsock::{VsockAddr, VsockStream};
24use tower::{BoxError, Service};
25use tracing::debug;
26
27use super::telemetry::HttpTransactionErrorTelemetry;
28use crate::net::dns::{DnsError, SystemHttpConnector, SystemResolver};
29
30/// Imposes a limit on the age of a connection.
31///
32/// In many cases, it's undesirable to hold onto a connection indefinitely, even if it can be theoretically reused.
33/// Doing so can make it more difficult to perform maintenance on infrastructure, as the expectation of old connections
34/// being eventually closed and replaced isn't upheld.
35///
36/// This extension allows tracking the age of a connection (based on when the connector creates the connection) and
37/// checking if it's expired, or past the configured limit. Callers can then decide how to handle the expiration, such
38/// as by closing the connection.
39#[derive(Clone)]
40struct ConnectionAgeLimit {
41    limit: Duration,
42    created: Instant,
43}
44
45impl ConnectionAgeLimit {
46    fn new(limit: Duration) -> Self {
47        ConnectionAgeLimit {
48            limit,
49            created: Instant::now(),
50        }
51    }
52
53    fn is_expired(&self) -> bool {
54        self.created.elapsed() >= self.limit
55    }
56}
57
58/// An inner transport that abstracts over TCP, Unix domain socket, and vsock connections.
59///
60/// This allows using a single monomorphization of the HTTP/2 and TLS stacks regardless of the
61/// underlying transport, avoiding duplicate code generation for each transport type.
62enum Transport {
63    Tcp(TokioIo<TcpStream>),
64    #[cfg(unix)]
65    Unix(TokioIo<tokio::net::UnixStream>),
66    #[cfg(target_os = "linux")]
67    Vsock(TokioIo<VsockStream>),
68}
69
70impl Connection for Transport {
71    fn connected(&self) -> Connected {
72        match self {
73            Self::Tcp(s) => s.connected(),
74            #[cfg(unix)]
75            Self::Unix(_) => Connected::new(),
76            #[cfg(target_os = "linux")]
77            Self::Vsock(_) => Connected::new(),
78        }
79    }
80}
81
82impl hyper::rt::Read for Transport {
83    fn poll_read(
84        self: Pin<&mut Self>, cx: &mut Context<'_>, buf: hyper::rt::ReadBufCursor<'_>,
85    ) -> Poll<io::Result<()>> {
86        match Pin::get_mut(self) {
87            Self::Tcp(s) => Pin::new(s).poll_read(cx, buf),
88            #[cfg(unix)]
89            Self::Unix(s) => Pin::new(s).poll_read(cx, buf),
90            #[cfg(target_os = "linux")]
91            Self::Vsock(s) => Pin::new(s).poll_read(cx, buf),
92        }
93    }
94}
95
96impl hyper::rt::Write for Transport {
97    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
98        match Pin::get_mut(self) {
99            Self::Tcp(s) => Pin::new(s).poll_write(cx, buf),
100            #[cfg(unix)]
101            Self::Unix(s) => Pin::new(s).poll_write(cx, buf),
102            #[cfg(target_os = "linux")]
103            Self::Vsock(s) => Pin::new(s).poll_write(cx, buf),
104        }
105    }
106
107    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
108        match Pin::get_mut(self) {
109            Self::Tcp(s) => Pin::new(s).poll_flush(cx),
110            #[cfg(unix)]
111            Self::Unix(s) => Pin::new(s).poll_flush(cx),
112            #[cfg(target_os = "linux")]
113            Self::Vsock(s) => Pin::new(s).poll_flush(cx),
114        }
115    }
116
117    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
118        match Pin::get_mut(self) {
119            Self::Tcp(s) => Pin::new(s).poll_shutdown(cx),
120            #[cfg(unix)]
121            Self::Unix(s) => Pin::new(s).poll_shutdown(cx),
122            #[cfg(target_os = "linux")]
123            Self::Vsock(s) => Pin::new(s).poll_shutdown(cx),
124        }
125    }
126
127    fn is_write_vectored(&self) -> bool {
128        match self {
129            Self::Tcp(s) => s.is_write_vectored(),
130            #[cfg(unix)]
131            Self::Unix(s) => s.is_write_vectored(),
132            #[cfg(target_os = "linux")]
133            Self::Vsock(s) => s.is_write_vectored(),
134        }
135    }
136
137    fn poll_write_vectored(
138        self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[io::IoSlice<'_>],
139    ) -> Poll<io::Result<usize>> {
140        match Pin::get_mut(self) {
141            Self::Tcp(s) => Pin::new(s).poll_write_vectored(cx, bufs),
142            #[cfg(unix)]
143            Self::Unix(s) => Pin::new(s).poll_write_vectored(cx, bufs),
144            #[cfg(target_os = "linux")]
145            Self::Vsock(s) => Pin::new(s).poll_write_vectored(cx, bufs),
146        }
147    }
148}
149
150pin_project! {
151    /// A connection that supports both HTTP and HTTPS.
152    pub struct HttpsCapableConnection {
153        #[pin]
154        inner: MaybeHttpsStream<Transport>,
155        bytes_sent: Option<Counter>,
156        error_telemetry: Option<HttpTransactionErrorTelemetry>,
157        conn_age_limit: Option<Duration>,
158    }
159}
160
161impl Connection for HttpsCapableConnection {
162    fn connected(&self) -> Connected {
163        let connected = self.inner.connected();
164
165        if let Some(conn_age_limit) = self.conn_age_limit {
166            debug!("setting connection age limit to {:?}", conn_age_limit);
167            connected.extra(ConnectionAgeLimit::new(conn_age_limit))
168        } else {
169            connected
170        }
171    }
172}
173
174impl hyper::rt::Read for HttpsCapableConnection {
175    fn poll_read(
176        self: Pin<&mut Self>, cx: &mut Context<'_>, buf: hyper::rt::ReadBufCursor<'_>,
177    ) -> Poll<io::Result<()>> {
178        let this = self.project();
179        this.inner.poll_read(cx, buf)
180    }
181}
182
183impl hyper::rt::Write for HttpsCapableConnection {
184    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
185        let this = self.project();
186        match this.inner.poll_write(cx, buf) {
187            Poll::Ready(Ok(n)) => {
188                if let Some(bytes_sent) = this.bytes_sent {
189                    bytes_sent.increment(n as u64);
190                }
191                Poll::Ready(Ok(n))
192            }
193            Poll::Ready(Err(error)) => {
194                if let Some(error_telemetry) = this.error_telemetry.as_ref() {
195                    error_telemetry.increment_wrote_request_error();
196                }
197                Poll::Ready(Err(error))
198            }
199            other => other,
200        }
201    }
202
203    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
204        let this = self.project();
205        match this.inner.poll_flush(cx) {
206            Poll::Ready(Err(error)) => {
207                if let Some(error_telemetry) = this.error_telemetry.as_ref() {
208                    error_telemetry.increment_wrote_request_error();
209                }
210                Poll::Ready(Err(error))
211            }
212            other => other,
213        }
214    }
215
216    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
217        let this = self.project();
218        this.inner.poll_shutdown(cx)
219    }
220
221    fn is_write_vectored(&self) -> bool {
222        self.inner.is_write_vectored()
223    }
224
225    fn poll_write_vectored(
226        self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[io::IoSlice<'_>],
227    ) -> Poll<io::Result<usize>> {
228        let this = self.project();
229        match this.inner.poll_write_vectored(cx, bufs) {
230            Poll::Ready(Ok(n)) => {
231                if let Some(bytes_sent) = this.bytes_sent {
232                    bytes_sent.increment(n as u64);
233                }
234                Poll::Ready(Ok(n))
235            }
236            Poll::Ready(Err(error)) => {
237                if let Some(error_telemetry) = this.error_telemetry.as_ref() {
238                    error_telemetry.increment_wrote_request_error();
239                }
240                Poll::Ready(Err(error))
241            }
242            other => other,
243        }
244    }
245}
246
247/// An inner connector that routes to TCP (via DNS), a Unix domain socket, or a vsock socket.
248///
249/// When a Unix socket path is configured, all connections are routed through that socket regardless
250/// of the URI host. When a vsock CID is configured, all connections are routed through that vsock
251/// socket using the port from the destination URI. Otherwise, connections use the standard DNS +
252/// TCP path.
253#[derive(Clone)]
254struct InnerConnector {
255    http: SystemHttpConnector,
256    #[cfg(unix)]
257    connect_timeout: Duration,
258    error_telemetry: Option<HttpTransactionErrorTelemetry>,
259    #[cfg(unix)]
260    unix_socket_path: Option<Arc<std::path::Path>>,
261    #[cfg(target_os = "linux")]
262    vsock_addr: Option<VsockAddr>,
263}
264
265impl Service<Uri> for InnerConnector {
266    type Response = Transport;
267    type Error = BoxError;
268    type Future = Pin<Box<dyn Future<Output = Result<Transport, BoxError>> + Send>>;
269
270    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
271        // When routing via vsock or a Unix domain socket, the TCP/DNS connector is not used, so we
272        // consider the service immediately ready. vsock takes priority over Unix (matching Agent
273        // behavior) when both are configured.
274        #[cfg(target_os = "linux")]
275        if self.vsock_addr.is_some() {
276            return Poll::Ready(Ok(()));
277        }
278
279        #[cfg(unix)]
280        if self.unix_socket_path.is_some() {
281            return Poll::Ready(Ok(()));
282        }
283
284        self.http.poll_ready(cx).map_err(Into::into)
285    }
286
287    fn call(&mut self, dst: Uri) -> Self::Future {
288        #[cfg(target_os = "linux")]
289        if let Some(addr) = self.vsock_addr {
290            let connect_timeout = self.connect_timeout;
291            let error_telemetry = self.error_telemetry.clone();
292            return Box::pin(async move {
293                let stream = tokio::time::timeout(connect_timeout, VsockStream::connect(addr))
294                    .await
295                    .map_err(|_| -> BoxError {
296                        if let Some(error_telemetry) = &error_telemetry {
297                            error_telemetry.increment_connection_error();
298                        }
299                        Box::new(io::Error::new(io::ErrorKind::TimedOut, "vsock connect timed out"))
300                    })?
301                    .map_err(|e| -> BoxError {
302                        if let Some(error_telemetry) = &error_telemetry {
303                            error_telemetry.increment_connection_error();
304                        }
305                        Box::new(e)
306                    })?;
307                Ok(Transport::Vsock(TokioIo::new(stream)))
308            });
309        }
310
311        #[cfg(unix)]
312        if let Some(path) = self.unix_socket_path.clone() {
313            let connect_timeout = self.connect_timeout;
314            let error_telemetry = self.error_telemetry.clone();
315            return Box::pin(async move {
316                let stream = tokio::time::timeout(connect_timeout, tokio::net::UnixStream::connect(&*path))
317                    .await
318                    .map_err(|_| -> BoxError {
319                        if let Some(error_telemetry) = &error_telemetry {
320                            error_telemetry.increment_connection_error();
321                        }
322                        Box::new(io::Error::new(io::ErrorKind::TimedOut, "unix socket connect timed out"))
323                    })?
324                    .map_err(|e| -> BoxError {
325                        if let Some(error_telemetry) = &error_telemetry {
326                            error_telemetry.increment_connection_error();
327                        }
328                        Box::new(e)
329                    })?;
330                Ok(Transport::Unix(TokioIo::new(stream)))
331            });
332        }
333
334        let fut = self.http.call(dst);
335        let error_telemetry = self.error_telemetry.clone();
336        Box::pin(async move {
337            let tcp = fut.await.map_err(|error| {
338                if !is_dns_error(&error) {
339                    if let Some(error_telemetry) = &error_telemetry {
340                        error_telemetry.increment_connection_error();
341                    }
342                }
343                BoxError::from(error)
344            })?;
345            Ok(Transport::Tcp(tcp))
346        })
347    }
348}
349
350/// HTTP protocol selection for client connections.
351#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
352pub enum HttpProtocol {
353    /// Automatically negotiate HTTP/2 with HTTP/1.1 fallback.
354    #[default]
355    Auto,
356
357    /// Use HTTP/1.1 only.
358    Http1,
359}
360
361/// A connector that supports HTTP or HTTPS.
362#[derive(Clone)]
363pub struct HttpsCapableConnector {
364    inner: HttpsConnector<InnerConnector>,
365    bytes_sent: Option<Counter>,
366    error_telemetry: Option<HttpTransactionErrorTelemetry>,
367    conn_age_limit: Option<Duration>,
368}
369
370impl Service<Uri> for HttpsCapableConnector {
371    type Response = HttpsCapableConnection;
372    type Error = BoxError;
373    type Future = Pin<Box<dyn Future<Output = Result<HttpsCapableConnection, BoxError>> + Send>>;
374
375    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
376        self.inner.poll_ready(cx)
377    }
378
379    fn call(&mut self, dst: Uri) -> Self::Future {
380        let inner = self.inner.call(dst);
381        let bytes_sent = self.bytes_sent.clone();
382        let error_telemetry = self.error_telemetry.clone();
383        let conn_age_limit = self.conn_age_limit;
384        Box::pin(async move {
385            match inner.await {
386                Ok(inner) => Ok(HttpsCapableConnection {
387                    inner,
388                    bytes_sent,
389                    error_telemetry,
390                    conn_age_limit,
391                }),
392                Err(error) => {
393                    if is_tls_error(error.as_ref()) {
394                        if let Some(error_telemetry) = &error_telemetry {
395                            error_telemetry.increment_tls_error();
396                        }
397                    }
398                    Err(error)
399                }
400            }
401        })
402    }
403}
404
405fn build_dns_resolver(error_telemetry: &Option<HttpTransactionErrorTelemetry>) -> SystemResolver {
406    let mut r = SystemResolver::new();
407    if let Some(et) = error_telemetry {
408        r = r.with_lookup_errors_counter(et.dns_errors());
409    }
410    r
411}
412
413/// A builder for `HttpsCapableConnector`.
414#[derive(Default)]
415pub struct HttpsCapableConnectorBuilder {
416    connect_timeout: Option<Duration>,
417    bytes_sent: Option<Counter>,
418    error_telemetry: Option<HttpTransactionErrorTelemetry>,
419    conn_age_limit: Option<Duration>,
420    http_protocol: HttpProtocol,
421    #[cfg(unix)]
422    unix_socket_path: Option<PathBuf>,
423    #[cfg(target_os = "linux")]
424    vsock_addr: Option<VsockAddr>,
425}
426
427impl HttpsCapableConnectorBuilder {
428    /// Sets the timeout when connecting to the remote host.
429    ///
430    /// Defaults to 30 seconds.
431    pub fn with_connect_timeout(mut self, timeout: Duration) -> Self {
432        self.connect_timeout = Some(timeout);
433        self
434    }
435
436    /// Sets the HTTP protocol selection for client connections.
437    ///
438    /// Defaults to [`HttpProtocol::Auto`].
439    pub fn with_http_protocol(mut self, protocol: HttpProtocol) -> Self {
440        self.http_protocol = protocol;
441        self
442    }
443
444    /// Sets the maximum age of a connection before it's closed.
445    ///
446    /// This is distinct from the maximum idle time: if any connection's age exceeds `limit`, it will be closed rather
447    /// than being reused and added to the idle connection pool.
448    ///
449    /// Defaults to no limit.
450    pub fn with_connection_age_limit<L>(mut self, limit: L) -> Self
451    where
452        L: Into<Option<Duration>>,
453    {
454        self.conn_age_limit = limit.into();
455        self
456    }
457
458    /// Sets a counter that gets incremented with the number of bytes sent over the connection.
459    ///
460    /// This tracks bytes sent at the HTTP client level, which includes headers and body but doesn't include underlying
461    /// transport overhead, such as TLS handshaking, and so on.
462    ///
463    /// Defaults to unset.
464    pub fn with_bytes_sent_counter(mut self, counter: Counter) -> Self {
465        self.bytes_sent = Some(counter);
466        self
467    }
468
469    /// Sets the telemetry counters used to track HTTP request lifecycle failures.
470    pub(super) fn with_error_telemetry(mut self, error_telemetry: HttpTransactionErrorTelemetry) -> Self {
471        self.error_telemetry = Some(error_telemetry);
472        self
473    }
474
475    /// Sets a Unix domain socket path to route all connections through.
476    ///
477    /// When set, the connector will connect to this Unix socket instead of performing DNS resolution
478    /// and TCP connection. The URI host is ignored in this case—all requests are sent through the
479    /// configured socket.
480    ///
481    /// Defaults to unset (TCP connections via DNS).
482    #[cfg(unix)]
483    pub fn with_unix_socket_path<P: Into<PathBuf>>(mut self, path: P) -> Self {
484        self.unix_socket_path = Some(path.into());
485        self
486    }
487
488    /// Sets a vsock address to route all connections through.
489    ///
490    /// When set, the connector will connect via AF_VSOCK using the given address, bypassing
491    /// DNS and TCP. This allows connecting to a server process running in a host or hypervisor
492    /// context from within a guest VM (for example, Nitro Enclaves).
493    ///
494    /// Defaults to unset (TCP connections via DNS).
495    #[cfg(target_os = "linux")]
496    pub fn with_vsock_addr(mut self, addr: VsockAddr) -> Self {
497        self.vsock_addr = Some(addr);
498        self
499    }
500
501    /// Builds the `HttpsCapableConnector` from the given TLS configuration.
502    pub fn build(self, tls_config: ClientConfig) -> Result<HttpsCapableConnector, GenericError> {
503        let connect_timeout = self.connect_timeout.unwrap_or(Duration::from_secs(30));
504
505        // Create the HTTP connector, and ensure that we don't enforce _only_ HTTP, since that will break being able to
506        // wrap this in an HTTPS connector.
507        let mut http_connector = HttpConnector::new_with_resolver(build_dns_resolver(&self.error_telemetry));
508        http_connector.set_connect_timeout(Some(connect_timeout));
509        http_connector.enforce_http(false);
510
511        let inner_connector = InnerConnector {
512            http: http_connector,
513            #[cfg(unix)]
514            connect_timeout,
515            error_telemetry: self.error_telemetry.clone(),
516            #[cfg(unix)]
517            unix_socket_path: self.unix_socket_path.map(PathBuf::into_boxed_path).map(Arc::from),
518            #[cfg(target_os = "linux")]
519            vsock_addr: self.vsock_addr,
520        };
521
522        // Create the HTTPS connector.
523        let https_connector_builder = HttpsConnectorBuilder::new().with_tls_config(tls_config).https_or_http();
524        let https_connector = match self.http_protocol {
525            HttpProtocol::Auto => https_connector_builder
526                .enable_all_versions()
527                .wrap_connector(inner_connector),
528            HttpProtocol::Http1 => https_connector_builder.enable_http1().wrap_connector(inner_connector),
529        };
530
531        Ok(HttpsCapableConnector {
532            inner: https_connector,
533            bytes_sent: self.bytes_sent,
534            error_telemetry: self.error_telemetry,
535            conn_age_limit: self.conn_age_limit,
536        })
537    }
538}
539
540#[cfg(test)]
541fn configure_tls_alpn_for_http_protocol(mut tls_config: ClientConfig, protocol: HttpProtocol) -> ClientConfig {
542    match protocol {
543        HttpProtocol::Auto => {
544            tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
545        }
546        HttpProtocol::Http1 => {
547            tls_config.alpn_protocols.clear();
548        }
549    }
550
551    tls_config
552}
553
554fn is_tls_error(error: &(dyn std::error::Error + 'static)) -> bool {
555    let mut current = Some(error);
556    while let Some(error) = current {
557        if error.downcast_ref::<rustls::Error>().is_some() {
558            return true;
559        }
560        current = error.source();
561    }
562    false
563}
564
565fn is_dns_error(error: &(dyn std::error::Error + 'static)) -> bool {
566    let mut current = Some(error);
567    while let Some(error) = current {
568        if error.downcast_ref::<DnsError>().is_some() {
569            return true;
570        }
571        current = error.source();
572    }
573    false
574}
575
576pub(super) fn check_connection_state(captured_conn: CaptureConnection) {
577    let maybe_conn_metadata = captured_conn.connection_metadata();
578    if let Some(conn_metadata) = maybe_conn_metadata.as_ref() {
579        let mut extensions = Extensions::new();
580        conn_metadata.get_extras(&mut extensions);
581
582        // If the connection has an age limit, check to see if the connection is expired (i.e. too old) and "poison"
583        // it if so. Poisoning indicates to `hyper` that the connection should be closed/dropped instead of
584        // returning it back to the idle connection pool.
585        if let Some(conn_age_limit) = extensions.get::<ConnectionAgeLimit>() {
586            if conn_age_limit.is_expired() {
587                debug!("connection is expired; poisoning it");
588                conn_metadata.poison();
589            }
590        }
591    }
592}
593
594#[cfg(test)]
595mod tests {
596    use super::{configure_tls_alpn_for_http_protocol, HttpProtocol};
597
598    fn empty_tls_config() -> rustls::ClientConfig {
599        rustls::ClientConfig::builder_with_provider(default_crypto_provider().into())
600            .with_safe_default_protocol_versions()
601            .expect("default protocol versions should be valid")
602            .with_root_certificates(rustls::RootCertStore::empty())
603            .with_no_client_auth()
604    }
605
606    #[cfg(not(windows))]
607    fn default_crypto_provider() -> rustls::crypto::CryptoProvider {
608        rustls::crypto::aws_lc_rs::default_provider()
609    }
610
611    #[cfg(windows)]
612    fn default_crypto_provider() -> rustls::crypto::CryptoProvider {
613        rustls_cng_crypto::default_provider()
614    }
615
616    #[test]
617    fn auto_protocol_advertises_h2_and_http1_alpn() {
618        let tls_config = configure_tls_alpn_for_http_protocol(empty_tls_config(), HttpProtocol::Auto);
619
620        assert_eq!(tls_config.alpn_protocols, vec![b"h2".to_vec(), b"http/1.1".to_vec()]);
621    }
622
623    #[test]
624    fn http1_protocol_leaves_alpn_empty() {
625        let tls_config = configure_tls_alpn_for_http_protocol(empty_tls_config(), HttpProtocol::Http1);
626
627        assert!(tls_config.alpn_protocols.is_empty());
628    }
629
630    // vsock takes priority over unix when both are configured, matching Agent behavior.
631    // We verify by checking the error does not mention "unix" — if unix had priority it would
632    // fail with a socket-path error; vsock produces a connection or device error instead.
633    #[cfg(target_os = "linux")]
634    #[tokio::test]
635    async fn vsock_takes_priority_over_unix_when_both_set() {
636        use std::sync::Arc;
637
638        use tower::Service as _;
639
640        use super::{InnerConnector, VsockAddr};
641        use crate::net::dns::SystemResolver;
642
643        let mut connector = InnerConnector {
644            http: SystemResolver::new().into_http_connector(),
645            connect_timeout: std::time::Duration::from_secs(1),
646            error_telemetry: None,
647            unix_socket_path: Some(Arc::from(std::path::Path::new("/tmp/test.sock"))),
648            vsock_addr: Some(VsockAddr::new(2, 5001)),
649        };
650
651        // Verify vsock path was taken: if unix had priority the error would mention the socket
652        // path or "unix"; a vsock attempt produces a connection or device error instead.
653        let uri: http::Uri = "https://127.0.0.1:5001/".parse().unwrap();
654        let err = connector.call(uri).await.err().expect("expected a connection error");
655        assert!(
656            !err.to_string().contains("unix"),
657            "expected vsock error (not unix socket error), got: {err}"
658        );
659    }
660}
saluki_io/net/client/http/conn.rs

saluki_io/net/client/http/
conn.rs
