polyoxide_core/
transport.rs

1//! The seam between the send loop and the network.
2//!
3//! [`HttpTransport`] takes one built [`HttpRequest`] and answers with a
4//! [`Response`], or with a [`TransportError`] when no response head arrived.
5//! The send loop decides everything else: whether a status is a failure,
6//! whether to retry, what to hold. A client sends on the target's default
7//! transport unless one is injected.
8
9use std::fmt;
10use std::time::Duration;
11
12use bytes::Bytes;
13use http::{HeaderMap, Method};
14use polyoxide_rt::BoxFuture;
15use polyoxide_venue::{Class, Classify};
16use thiserror::Error;
17use url::Url;
18
19use crate::response::Response;
20
21/// One attempt's request, as the send loop builds it.
22///
23/// Only core builds one: the struct is `non_exhaustive` and has no public
24/// constructor, so a transport reads the fields it is handed.
25#[derive(Debug, Clone)]
26#[non_exhaustive]
27pub struct HttpRequest {
28    /// The HTTP method.
29    pub method: Method,
30    /// The full URL: the base URL joined with the path, with the query
31    /// form-urlencoded onto it.
32    pub url: Url,
33    /// The headers, signed.
34    pub headers: HeaderMap,
35    /// The body, sent as given.
36    pub body: Option<Bytes>,
37    /// How long the whole exchange may take, the body's read included: the
38    /// request's own timeout, else the client's. Always set.
39    pub timeout: Duration,
40}
41
42/// Sends one request and answers with its response.
43///
44/// The send loop is the only caller. A client holds one transport and shares
45/// it with its `with_base_url` siblings.
46pub trait HttpTransport: Send + Sync + fmt::Debug + 'static {
47    /// Send `request` and answer with its response, whatever its status.
48    ///
49    /// An implementation:
50    ///
51    /// - returns a [`Response`] for every status, an error status included:
52    ///   judging a status is the send loop's job, never the transport's;
53    /// - fails with a [`TransportError`] only when no response head arrived;
54    /// - bounds the whole exchange, the body's read included, by
55    ///   `request.timeout`;
56    /// - leaves the body unread until the [`Response`] is asked for it.
57    fn send(&self, request: HttpRequest) -> BoxFuture<Result<Response, TransportError>>;
58}
59
60/// Where a transport's exchange broke.
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62#[non_exhaustive]
63pub enum TransportKind {
64    /// The connection could not be made.
65    Connect,
66    /// The timeout ran out.
67    Timeout,
68    /// The body could not be read to its end.
69    Body,
70    /// The request could not be built, or sent as built.
71    Request,
72    /// Anything else.
73    Other,
74}
75
76impl fmt::Display for TransportKind {
77    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
78        f.write_str(match self {
79            Self::Connect => "connect",
80            Self::Timeout => "timeout",
81            Self::Body => "body",
82            Self::Request => "request",
83            Self::Other => "other",
84        })
85    }
86}
87
88/// Why no response arrived, or why its body could not be read.
89///
90/// [`kind`](Self::kind) says where the exchange broke, and the boxed source is
91/// the transport's own error, kept for its message and its chain. A custom
92/// [`HttpTransport`] builds one with [`new`](Self::new). Its class is
93/// [`Class::Network`] for every kind but [`TransportKind::Request`], a request
94/// that could not be built, which is [`Class::InvalidRequest`].
95#[derive(Debug, Error)]
96#[non_exhaustive]
97#[error("{kind}: {source}")]
98pub struct TransportError {
99    kind: TransportKind,
100    source: Box<dyn std::error::Error + Send + Sync>,
101}
102
103impl TransportError {
104    /// A transport failure of `kind`, caused by `source`.
105    pub fn new(
106        kind: TransportKind,
107        source: impl Into<Box<dyn std::error::Error + Send + Sync>>,
108    ) -> Self {
109        Self {
110            kind,
111            source: source.into(),
112        }
113    }
114
115    /// Where the exchange broke.
116    pub fn kind(&self) -> TransportKind {
117        self.kind
118    }
119}
120
121impl Classify for TransportError {
122    fn class(&self) -> Class {
123        match self.kind {
124            TransportKind::Request => Class::InvalidRequest,
125            TransportKind::Connect
126            | TransportKind::Timeout
127            | TransportKind::Body
128            | TransportKind::Other => Class::Network,
129        }
130    }
131}
132
133/// The transport the native targets use: reqwest.
134///
135/// The client builder makes one from its timeout, pool size and `gzip`; build
136/// one with [`new`](Self::new) to send on a client of your own.
137#[cfg(not(target_arch = "wasm32"))]
138#[derive(Debug, Clone)]
139pub struct ReqwestTransport {
140    client: reqwest::Client,
141}
142
143#[cfg(not(target_arch = "wasm32"))]
144impl ReqwestTransport {
145    /// A transport that sends on `client`.
146    pub fn new(client: reqwest::Client) -> Self {
147        Self { client }
148    }
149
150    /// The default native transport, built from a client builder's settings:
151    /// its timeout, its idle connections per host and its `gzip` choice
152    /// (unset keeps reqwest's default, so every client asks for gzip), with a
153    /// ten second connect timeout and no redirects.
154    pub(crate) fn from_settings(
155        timeout: Duration,
156        pool_size: usize,
157        gzip: Option<bool>,
158    ) -> Result<Self, TransportError> {
159        let mut builder = reqwest::Client::builder();
160        if let Some(enabled) = gzip {
161            builder = builder.gzip(enabled);
162        }
163        let client = builder
164            .timeout(timeout)
165            .connect_timeout(Duration::from_secs(10))
166            .redirect(reqwest::redirect::Policy::none())
167            .pool_max_idle_per_host(pool_size)
168            .build()
169            .map_err(map_error)?;
170        Ok(Self::new(client))
171    }
172}
173
174/// The default transport, built from a client builder's settings: reqwest's on
175/// a host target.
176#[cfg(not(target_arch = "wasm32"))]
177pub(crate) fn default_transport(
178    timeout: Duration,
179    pool_size: usize,
180    gzip: Option<bool>,
181) -> Result<std::sync::Arc<dyn HttpTransport>, TransportError> {
182    Ok(std::sync::Arc::new(ReqwestTransport::from_settings(
183        timeout, pool_size, gzip,
184    )?))
185}
186
187/// The default transport on `wasm32`: the browser's `fetch`. The timeout is set
188/// per request, and the browser has its own pool and compression, so the
189/// builder's pool size and `gzip` choice are unused.
190#[cfg(target_arch = "wasm32")]
191pub(crate) fn default_transport(
192    _timeout: Duration,
193    _pool_size: usize,
194    _gzip: Option<bool>,
195) -> Result<std::sync::Arc<dyn HttpTransport>, TransportError> {
196    Ok(std::sync::Arc::new(crate::FetchTransport::new()))
197}
198
199#[cfg(not(target_arch = "wasm32"))]
200impl HttpTransport for ReqwestTransport {
201    fn send(&self, request: HttpRequest) -> BoxFuture<Result<Response, TransportError>> {
202        let mut builder = self
203            .client
204            .request(request.method, request.url)
205            .headers(request.headers)
206            .timeout(request.timeout);
207        if let Some(body) = request.body {
208            builder = builder.body(body);
209        }
210        Box::pin(async move {
211            let response = builder.send().await.map_err(map_error)?;
212            let status = response.status();
213            let headers = response.headers().clone();
214            let url = response.url().clone();
215            Ok(Response::new(
216                status,
217                headers,
218                url,
219                Box::pin(async move { response.bytes().await.map_err(map_error) }),
220            ))
221        })
222    }
223}
224
225/// The transport error for a reqwest one, by the rule core has always used:
226/// a failure in transit is `Network` whatever reqwest labels it, so it is
227/// decided before a request that could not be built.
228#[cfg(not(target_arch = "wasm32"))]
229fn map_error(err: reqwest::Error) -> TransportError {
230    let flags = Flags {
231        timeout: err.is_timeout(),
232        connect: err.is_connect(),
233        body: err.is_body() || err.is_decode(),
234        builder: err.is_builder(),
235    };
236    TransportError::new(kind_of(flags, &err), err)
237}
238
239/// What a reqwest error reports about itself.
240#[cfg(not(target_arch = "wasm32"))]
241#[derive(Clone, Copy)]
242struct Flags {
243    timeout: bool,
244    connect: bool,
245    /// A body that broke off; reqwest labels that one a decode error too.
246    body: bool,
247    builder: bool,
248}
249
250/// The kind for a reqwest error's flags and its source chain: a timeout, a
251/// failed connect, a body that broke off, or an I/O error or a body error
252/// anywhere in the chain is a failure in transit, checked before a request
253/// that could not be built; anything else is `Other`.
254#[cfg(not(target_arch = "wasm32"))]
255fn kind_of(flags: Flags, err: &(dyn std::error::Error + 'static)) -> TransportKind {
256    if flags.timeout {
257        TransportKind::Timeout
258    } else if flags.connect {
259        TransportKind::Connect
260    } else if flags.body {
261        TransportKind::Body
262    } else if failed_in_transit(err) {
263        TransportKind::Other
264    } else if flags.builder {
265        TransportKind::Request
266    } else {
267        TransportKind::Other
268    }
269}
270
271/// Whether an I/O error or a body-kind reqwest error is anywhere in the
272/// error's source chain, however reqwest labels the error itself.
273#[cfg(not(target_arch = "wasm32"))]
274fn failed_in_transit(err: &(dyn std::error::Error + 'static)) -> bool {
275    let mut source = err.source();
276    while let Some(inner) = source {
277        if inner.is::<std::io::Error>()
278            || inner
279                .downcast_ref::<reqwest::Error>()
280                .is_some_and(reqwest::Error::is_body)
281        {
282            return true;
283        }
284        source = inner.source();
285    }
286    false
287}
288
289#[cfg(not(target_arch = "wasm32"))]
290#[cfg(test)]
291mod tests {
292    use std::io::{Read, Write};
293    use std::net::TcpListener;
294
295    use http::{HeaderValue, StatusCode};
296
297    use super::*;
298    use crate::error::ApiError;
299
300    #[test]
301    fn a_kind_displays_in_lower_case() {
302        for (kind, text) in [
303            (TransportKind::Connect, "connect"),
304            (TransportKind::Timeout, "timeout"),
305            (TransportKind::Body, "body"),
306            (TransportKind::Request, "request"),
307            (TransportKind::Other, "other"),
308        ] {
309            assert_eq!(kind.to_string(), text);
310        }
311    }
312
313    #[test]
314    fn an_error_displays_its_kind_then_its_source() {
315        let err = TransportError::new(TransportKind::Timeout, "no answer");
316        assert_eq!(err.kind(), TransportKind::Timeout);
317        assert_eq!(err.to_string(), "timeout: no answer");
318        assert_eq!(
319            std::error::Error::source(&err).map(ToString::to_string),
320            Some("no answer".to_owned())
321        );
322        assert_eq!(
323            ApiError::from(err).to_string(),
324            "Transport error: timeout: no answer"
325        );
326    }
327
328    #[test]
329    fn only_a_request_that_could_not_be_built_is_not_a_network_failure() {
330        for kind in [
331            TransportKind::Connect,
332            TransportKind::Timeout,
333            TransportKind::Body,
334            TransportKind::Other,
335        ] {
336            let err = TransportError::new(kind, "x");
337            assert_eq!(err.class(), Class::Network, "{kind}");
338            assert!(err.is_retriable(), "{kind}");
339            assert!(err.is_fault(), "{kind}");
340        }
341        let err = TransportError::new(TransportKind::Request, "x");
342        assert_eq!(err.class(), Class::InvalidRequest);
343        assert!(!err.is_retriable());
344    }
345
346    /// An error with an I/O error behind it, as a reqwest error that failed
347    /// in transit has.
348    #[derive(Debug, Error)]
349    #[error("wrapped")]
350    struct Wrapped(#[source] std::io::Error);
351
352    #[test]
353    fn an_io_error_in_the_source_chain_is_never_a_request_error() {
354        let err = Wrapped(std::io::Error::from(std::io::ErrorKind::ConnectionReset));
355        for (flags, expected) in [
356            (
357                Flags {
358                    timeout: false,
359                    connect: false,
360                    body: false,
361                    builder: true,
362                },
363                TransportKind::Other,
364            ),
365            (
366                Flags {
367                    timeout: false,
368                    connect: false,
369                    body: false,
370                    builder: false,
371                },
372                TransportKind::Other,
373            ),
374            (
375                Flags {
376                    timeout: true,
377                    connect: false,
378                    body: false,
379                    builder: true,
380                },
381                TransportKind::Timeout,
382            ),
383            (
384                Flags {
385                    timeout: false,
386                    connect: true,
387                    body: false,
388                    builder: true,
389                },
390                TransportKind::Connect,
391            ),
392            (
393                Flags {
394                    timeout: false,
395                    connect: false,
396                    body: true,
397                    builder: true,
398                },
399                TransportKind::Body,
400            ),
401        ] {
402            let kind = kind_of(flags, &err);
403            assert_eq!(kind, expected);
404            assert_eq!(TransportError::new(kind, "x").class(), Class::Network);
405        }
406        // With no I/O error behind it, a builder error stays a request error.
407        let plain = TransportError::new(
408            kind_of(
409                Flags {
410                    timeout: false,
411                    connect: false,
412                    body: false,
413                    builder: true,
414                },
415                &fmt::Error,
416            ),
417            "x",
418        );
419        assert_eq!(plain.kind(), TransportKind::Request);
420    }
421
422    fn request(url: &str) -> HttpRequest {
423        HttpRequest {
424            method: Method::GET,
425            url: Url::parse(url).unwrap(),
426            headers: HeaderMap::new(),
427            body: None,
428            timeout: Duration::from_secs(5),
429        }
430    }
431
432    /// A transport on a client with no proxy, so a proxy set in the
433    /// environment cannot answer for the local addresses below.
434    fn transport() -> ReqwestTransport {
435        ReqwestTransport::new(reqwest::Client::builder().no_proxy().build().unwrap())
436    }
437
438    #[tokio::test]
439    async fn the_head_comes_back_whole_and_the_body_is_read_on_demand() {
440        let mut server = mockito::Server::new_async().await;
441        let mock = server
442            .mock("GET", "/v1/rows")
443            .match_query(mockito::Matcher::UrlEncoded("limit".into(), "2".into()))
444            .match_header("x-trace", "t-1")
445            .with_status(201)
446            .with_header("x-served-by", "mock")
447            .with_body("rows")
448            .create_async()
449            .await;
450        let mut sent = request(&format!("{}/v1/rows?limit=2", server.url()));
451        sent.headers
452            .insert("x-trace", HeaderValue::from_static("t-1"));
453
454        let response = transport().send(sent.clone()).await.unwrap();
455
456        assert_eq!(response.status(), StatusCode::CREATED);
457        assert_eq!(
458            response
459                .headers()
460                .get("x-served-by")
461                .and_then(|value| value.to_str().ok()),
462            Some("mock")
463        );
464        assert_eq!(response.url(), &sent.url);
465        assert_eq!(response.text().await.unwrap(), "rows");
466        mock.assert_async().await;
467    }
468
469    #[tokio::test]
470    async fn the_method_and_the_body_are_sent() {
471        let mut server = mockito::Server::new_async().await;
472        let mock = server
473            .mock("POST", "/v1/order")
474            .match_body(r#"{"size":1}"#)
475            .with_status(200)
476            .with_body("{}")
477            .create_async()
478            .await;
479        let mut sent = request(&format!("{}/v1/order", server.url()));
480        sent.method = Method::POST;
481        sent.body = Some(Bytes::from_static(br#"{"size":1}"#));
482
483        let response = transport().send(sent).await.unwrap();
484
485        assert_eq!(response.status(), StatusCode::OK);
486        mock.assert_async().await;
487    }
488
489    #[tokio::test]
490    async fn an_error_status_is_a_response_and_never_a_transport_error() {
491        let mut server = mockito::Server::new_async().await;
492        let _down = server
493            .mock("GET", "/v1/rows")
494            .with_status(503)
495            .with_body("down")
496            .create_async()
497            .await;
498
499        let response = transport()
500            .send(request(&format!("{}/v1/rows", server.url())))
501            .await
502            .unwrap();
503
504        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
505        assert_eq!(response.text().await.unwrap(), "down");
506    }
507
508    #[tokio::test]
509    async fn a_refused_connection_is_a_connect_failure() {
510        // Bind a port, then free it, so nothing is listening there.
511        let port = TcpListener::bind("127.0.0.1:0")
512            .unwrap()
513            .local_addr()
514            .unwrap()
515            .port();
516
517        let err = transport()
518            .send(request(&format!("http://127.0.0.1:{port}/")))
519            .await
520            .unwrap_err();
521
522        assert_eq!(err.kind(), TransportKind::Connect, "{err:?}");
523        assert_eq!(err.class(), Class::Network);
524    }
525
526    #[tokio::test]
527    async fn a_host_that_never_answers_is_a_timeout() {
528        // Accepted by the kernel, never answered.
529        let silent = TcpListener::bind("127.0.0.1:0").unwrap();
530        let mut sent = request(&format!("http://{}/", silent.local_addr().unwrap()));
531        sent.timeout = Duration::from_millis(200);
532
533        let err = transport().send(sent).await.unwrap_err();
534
535        assert_eq!(err.kind(), TransportKind::Timeout, "{err:?}");
536        assert_eq!(err.class(), Class::Network);
537    }
538
539    /// A server that answers with a head promising 100 bytes and six of them,
540    /// then holds the connection for `hold` before closing it.
541    fn short_body_server(hold: Duration) -> (String, std::thread::JoinHandle<()>) {
542        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
543        let url = format!("http://{}/", listener.local_addr().unwrap());
544        let server = std::thread::spawn(move || {
545            let (mut stream, _) = listener.accept().unwrap();
546            let _ = stream.read(&mut [0; 1024]);
547            let _ = stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n{\"a\":");
548            std::thread::sleep(hold);
549        });
550        (url, server)
551    }
552
553    #[tokio::test]
554    async fn a_body_that_stops_short_is_a_body_failure() {
555        let (url, server) = short_body_server(Duration::ZERO);
556
557        let response = transport().send(request(&url)).await.unwrap();
558        let err = response.bytes().await.unwrap_err();
559        server.join().unwrap();
560
561        assert!(
562            matches!(&err, ApiError::Transport(e) if e.kind() == TransportKind::Body),
563            "{err:?}"
564        );
565        assert_eq!(err.class(), Class::Network);
566    }
567
568    #[tokio::test]
569    async fn the_timeout_bounds_the_body_as_well_as_the_head() {
570        let (url, server) = short_body_server(Duration::from_millis(1_500));
571        let mut sent = request(&url);
572        sent.timeout = Duration::from_millis(300);
573
574        // The head arrives at once, so the send succeeds; the body then
575        // stalls past the timeout.
576        let response = transport().send(sent).await.unwrap();
577        let err = response.bytes().await.unwrap_err();
578        server.join().unwrap();
579
580        assert!(
581            matches!(&err, ApiError::Transport(e) if e.kind() == TransportKind::Timeout),
582            "{err:?}"
583        );
584    }
585
586    #[tokio::test]
587    async fn a_scheme_the_client_refuses_is_a_request_failure() {
588        let err = transport()
589            .send(request("ftp://127.0.0.1/file"))
590            .await
591            .unwrap_err();
592
593        assert_eq!(err.kind(), TransportKind::Request, "{err:?}");
594        assert_eq!(err.class(), Class::InvalidRequest);
595    }
596
597    #[tokio::test]
598    async fn the_default_transport_follows_no_redirect() {
599        let mut server = mockito::Server::new_async().await;
600        let _moved = server
601            .mock("GET", "/moved")
602            .with_status(302)
603            .with_header("location", "/elsewhere")
604            .create_async()
605            .await;
606        let elsewhere = server
607            .mock("GET", "/elsewhere")
608            .with_status(200)
609            .expect(0)
610            .create_async()
611            .await;
612        let http = crate::HttpClientBuilder::new(server.url()).build().unwrap();
613
614        let err = http
615            .send(crate::RequestParts::new(Method::GET, "/moved"), &[], None)
616            .await
617            .unwrap_err();
618
619        assert!(
620            matches!(&err, ApiError::Response(r) if r.status == StatusCode::FOUND),
621            "{err:?}"
622        );
623        elsewhere.assert_async().await;
624    }
625}
polyoxide_core/transport.rs

polyoxide_core/
transport.rs
