执行与 Pending

    pub(super) fn execute_request(&self, req: Request) -> Pending {
        let (method, url, mut headers, body, version, extensions) = req.pieces();
        if url.scheme() != "http" && url.scheme() != "https" {
            return Pending::new_err(error::url_bad_scheme(url));
        }

        // check if we're in https_only mode and check the scheme of the current URL
        if self.inner.https_only && url.scheme() != "https" {
            return Pending::new_err(error::url_bad_scheme(url));
        }

        // insert default headers in the request headers
        // without overwriting already appended headers.
        for (key, value) in &self.inner.headers {
            if let Entry::Vacant(entry) = headers.entry(key) {
                entry.insert(value.clone());
            }
        }

        let uri = match try_uri(&url) {
            Ok(uri) => uri,
            _ => return Pending::new_err(error::url_invalid_uri(url)),
        };

        let body = body.unwrap_or_else(Body::empty);

        self.proxy_auth(&uri, &mut headers);
        self.proxy_custom_headers(&uri, &mut headers);

        let in_flight = match version {
            #[cfg(feature = "http3")]
            http::Version::HTTP_3 if self.inner.h3_client.is_some() => {
                // ...
            }
            _ => {
                let mut req = builder.body(body).expect("valid request parts");
                *req.headers_mut() = headers.clone();
                let mut hyper = self.inner.hyper.clone();
                ResponseFuture::Default(hyper.call(req))
            }
        };

        let hyper_client = builder.build(connector_builder.build(config.connector_layers));
        let hyper_service = HyperService {
            hyper: hyper_client,
        };
        // ...
        let svc = tower::retry::Retry::new(retry_policy.clone(), hyper_service);

        #[cfg(feature = "cookies")]
        let svc = CookieService::new(svc, config.cookie_store.clone());
        let hyper = FollowRedirect::with_policy(svc, redirect_policy.clone());
        #[cfg(any(
            feature = "gzip",
            feature = "brotli",
            feature = "zstd",
            feature = "deflate"
        ))]
        let hyper = Decompression::new(hyper)
            // set everything to NO, in case tower-http has it enabled but
            // reqwest does not. then set to config value if cfg allows.
            .no_gzip()
            .no_deflate()
            .no_br()
            .no_zstd();
        #[cfg(feature = "gzip")]
        let hyper = hyper.gzip(config.accepts.gzip);
        // ...

type LayeredService<T> = MaybeDecompression<
    FollowRedirect<
        MaybeCookieService<tower::retry::Retry<crate::retry::Policy, T>>,
        TowerRedirectPolicy,
    >,
>;

== cookie set on a 302 vs on a retried 503
[  16ms server] /set      cookie=[] accept-encoding=["gzip"]
[  18ms server] /land     cookie=["hop=1"] accept-encoding=["gzip"]
client: /set -> 200 OK http://127.0.0.1:62036/land
[  19ms server] /flaky    cookie=["hop=1"] accept-encoding=["gzip"]
[  19ms server] /flaky    cookie=["hop=1"] accept-encoding=["gzip"]
client: /flaky -> 200 OK
[  19ms server] /after    cookie=["hop=1"] accept-encoding=["gzip"]

    let c = reqwest::Client::builder()
      .no_proxy()
      .connector_layer(LogLayer("inner"))
      .connector_layer(LogLayer("outer"))
      .build()
      .unwrap();
    let base = ms();
    let fut = c.get(srv.url("/lazy")).send();
    println!("[{:>4}ms] send() returned (+{}ms)", ms(), ms() - base);
    tokio::time::sleep(Duration::from_millis(200)).await;
    println!("[{:>4}ms] first poll", ms());
    let r = fut.await.unwrap();

== when does the connector run?
[  19ms] send() returned (+0ms)
[ 222ms] first poll
[ 222ms] connector layer "outer" called
[ 222ms] connector layer "inner" called
[ 224ms server] /lazy     cookie=[] accept-encoding=["gzip"]
[ 224ms] response 200 OK

    fn call(&mut self, req: hyper::Request<crate::async_impl::body::Body>) -> Self::Future {
        let clone = self.hyper.clone();
        let mut inner = std::mem::replace(&mut self.hyper, clone);
        Box::pin(async move { inner.call(req).await.map_err(crate::error::request) })
    }

enum PendingInner {
    Request(Pin<Box<PendingRequest>>),
    Error(Option<crate::Error>),
}

        #[pin]
        in_flight: ResponseFuture,
        #[pin]
        total_timeout: Option<Pin<Box<Sleep>>>,
        #[pin]
        read_timeout_fut: Option<Pin<Box<Sleep>>>,
        read_timeout: Option<Duration>,

        let total_timeout = self
            .inner
            .total_timeout
            .fetch(&extensions)
            .copied()
            .map(tokio::time::sleep)
            .map(Box::pin);

        if let Some(delay) = self.as_mut().total_timeout().as_mut().as_pin_mut() {
            if let Poll::Ready(()) = delay.poll(cx) {
                return Poll::Ready(Err(
                    crate::error::request(crate::error::TimedOut).with_url(self.url.clone())
                ));
            }
        }
        // read_timeout_fut 同样处理
        let res = match self.as_mut().in_flight().get_mut() {
            ResponseFuture::Default(r) => match ready!(Pin::new(r).poll(cx)) {
                Err(e) => {
                    return Poll::Ready(Err(e.if_no_url(|| self.url.clone())));
                }
                Ok(res) => res.map(super::body::boxed),
            },
            // ...
        };

size_of_val(send() future) = 16 bytes (request) / 16 bytes (builder error)

== timeout(300ms), await 400ms after send()
err after 0ms: error sending request for url (http://127.0.0.1:62036/late) / is_timeout=true / reqwest::Error { kind: Request, url: "http://127.0.0.1:62036/late", source: TimedOut }
new connections: 0
[ 882ms server] /late2    cookie=[] accept-encoding=["gzip"]
awaited 250ms after send(): Ok(200)