legacy Client

主线位置：客户端主线的起点。Client::request 算出连接池的键，拿到一条连接（池里的或新建的），按连接的协议改写请求，再交给那条连接的 SendRequest；请求被「还回来」时换一条连接重发。reqwest 发请求走的就是这里。

flowchart TB
  req["Client::request"] --> key["extract_domain：池的键"]
  key --> loopnode["send_request：重试循环"]
  loopnode --> trysend["try_send_request"]
  trysend --> conn["connection_for：池或新建"]
  conn --> fix["HTTP/1：补 Host，改 URI 形式"]
  fix --> sendreq["SendRequest::try_send_request"]
  sendreq -->|成功| back["连接回池，返回响应"]
  sendreq -->|"请求被还回"| retry{"开了重试且连接是复用的?"}
  retry -->|是| loopnode
  retry -->|否| errnode["返回 Canceled"]
  sendreq -->|"请求已写出"| errnode2["返回 SendRequest 错误"]

文件开头的模块文档说明了它的身份：「The legacy HTTP Client from 0.14.x」，「This Client will eventually be deconstructed into more composable parts. For now, to enable people to use hyper 1.0 quicker, this Client exists in much the same way it did in hyper 0.14.」本页讲 client.rs：请求怎么走、什么时候重发。连接池本身见连接池，DNS 和 TCP 见 HttpConnector，每条连接上的通道和回调见 conn::http1 客户端。

Client 与 Builder

Client：

pub struct Client<C, B> {
    config: Config,
    connector: C,
    exec: Exec,
    #[cfg(feature = "http1")]
    h1_builder: hyper::client::conn::http1::Builder,
    #[cfg(feature = "http2")]
    h2_builder: hyper::client::conn::http2::Builder<Exec>,
    pool: pool::Pool<PoolClient<B>, PoolKey>,
}

C 是连接器，B 是请求 body 的类型，一个 Client 只能发一种 body。两个 hyper 的连接级 Builder 原样存着，每建一条连接就拿它们 handshake 一次。Clone 把每个字段克隆一遍，pool 内部是 Arc，所以克隆出来的 Client 共用一个池；文档建议的共享方式就是克隆。

Builder::new 必须给一个执行器（Client::builder(TokioExecutor::new())），因为每条连接的 Connection 要 spawn 出去。默认值：

选项	默认	说明
`retry_canceled_requests`	`true`	本页后半
`set_host`	`true`	请求没有 `Host` 时按 URI 补上
协议（`Config::ver`）	`Ver::Auto`	HTTP/1，连接器报告 ALPN 协商出 h2 时用 HTTP/2；`http2_only(true)` 改成只用 HTTP/2
`pool_idle_timeout`	90 秒	要配 `pool_timer` 才生效，见连接池
`pool_max_idle_per_host`	`usize::MAX`	同上
`pool_timer`	无	同上

build_http 用默认的 HttpConnector，并把池的空闲超时设成连接器的 TCP keepalive。HTTP/1 的各个 http1_* 选项直接转给 h1_builder，含义见 conn::http1 客户端和 03 部分各章。

request：先查请求，再算池的键

request 同步地做两项检查，然后把剩下的事包进一个装箱的 future：

版本：HTTP/1.1 和 HTTP/2 放行；HTTP/1.0 放行，但 CONNECT 不行（UserUnsupportedRequestMethod）；其他版本（比如 HTTP/0.9）返回 UserUnsupportedVersion。
extract_domain：URI 必须带 scheme 和 authority，否则 UserAbsoluteUriRequired。例外是 CONNECT host:port：按端口补 scheme，443 补 https，其他补 http。

池的键就是 (Scheme, Authority)，源码里的类型别名旁边注释着「We might change this... :shrug:」。http://a.com 和 https://a.com 是两个池，a.com 和 a.com:80 也是两个：Authority 的相等只是忽略大小写的字符串比较（http 1.x），不会把默认端口规范掉。

返回的 ResponseFuture 里是 self.clone().send_request(req, pool_key) 这个 async 块：不 poll 它，什么也不会发生，连接也不会建。

拿连接

connection_for 调 one_connection_for：池关了就直接新建；否则让「从池里取」和「新建一条」用 select 赛跑，输掉的新连接 spawn 出去建完放进池里。赛跑的细节、HTTP/2 的「connecting 锁」都在连接池。connection_for 外面还有一层循环：HTTP/2 下取连接被取消（CheckoutIsClosed）时，开着 retry_canceled_requests 就再来一次。

新建连接的是 connect_to。连接器给出 IO 以后，HTTP/1 这一支：

                                    let (mut tx, conn) =
                                        h1_builder.handshake(io).await.map_err(crate::client::legacy::client::Error::tx)?;
                                    // ...
                                    let (err_tx, err_rx) = tokio::sync::oneshot::channel();
                                    // ...
                                    executor.execute(
                                        conn.with_upgrades()
                                            .map_err(|e| {
                                                // ...
                                                let _ =err_tx.send(e);
                                            })
                                            .map(|_| ()),
                                    );
                                    // ...
                                    match tx.ready().await {
                                        // If ready, the connection is usable for sending requests.
                                        Ok(_) => {
                                            // ...
                                            drop(err_rx);
                                            PoolTx::Http1(tx)
                                        }
                                        Err(e) if e.is_closed() => {
                                            // ...
                                            match err_rx.await {
                                                Ok(err) => {
                                                    // ...
                                                    return Err(crate::client::legacy::client::Error::tx(err));
                                                }
                                                // ...

三件事：Connection 带着 with_upgrades() 交给执行器，所以经 Client 发的请求能拿到升级后的 IO（见协议升级）；等 tx.ready()，也就是等连接任务第一次 poll 空队列、说「要」（conn::http1 客户端），连接才算建好；连接任务如果在 ready 之前就出错退出了，ready() 只能给出一个笼统的 ChannelClosed，于是另开一个 oneshot 把连接任务的真实错误捎回来。最后这一段是 #184（v0.1.13）加的，提交说明：「connection errors (e.g., TLS failures, unexpected server responses) were discarded, resulting in vague ChannelClosed errors」。

HTTP/2 一支同样 handshake、spawn、等 ready()，见 HTTP/2 客户端。

连接任务由 Client 的执行器 spawn，跑在建立这条连接时所在的运行时上；TokioExecutor 就是 tokio::spawn。连接任务被 drop 时在途请求报什么，见 conn::http1 客户端的 dispatch task is gone。

发之前改请求：Host 和 URI 形式

try_send_request 拿到连接后，HTTP/1 连接上：

        if pooled.is_http1() {
            if req.version() == Version::HTTP_2 {
                warn!("Connection is HTTP/1, but request requires HTTP/2");
                return Err(TrySendError::Nope(
                    e!(UserUnsupportedVersion).with_connect_info(pooled.conn_info.clone()),
                ));
            }

            if self.config.set_host {
                let uri = req.uri().clone();
                req.headers_mut().entry(HOST).or_insert_with(|| {
                    let hostname = uri.host().expect("authority implies host");
                    if let Some(port) = get_non_default_port(&uri) {
                        let s = format!("{hostname}:{port}");
                        HeaderValue::from_maybe_shared(bytes::Bytes::from(s))
                    } else {
                        HeaderValue::from_str(hostname)
                    }
                    .expect("uri host is valid header value")
                });
            }

            // CONNECT always sends authority-form, so check it first...
            if req.method() == Method::CONNECT {
                authority_form(req.uri_mut());
            } else if pooled.conn_info.is_proxied {
                absolute_form(req.uri_mut());
            } else {
                origin_form(req.uri_mut());
            }
        }

Host：用 entry(..).or_insert_with，请求里已有的 Host 不动。端口是 scheme 的默认端口（http 的 80、https/wss 的 443）时不写端口（get_non_default_port）。
URI：直连时改成 origin-form（/path?query，空路径变 /）；连接器报告这是代理连接（is_proxied）时保留 absolute-form；CONNECT 只留 authority。conn::http1 的 send_request 把 URI「as-is」写出去，这一步就是替它把形式改对。
HTTP/2 连接上这些都不做：请求保持完整的 URI，由 HTTP/2 编码成 :scheme、:authority、:path 伪头。

❗IMPORTANT

动脑筋：服务端看到的请求行和 Host 是什么？

平台 M2 Max、rustc 1.99.0。用默认 Client 依次发下面四个请求（PORT 是服务端的随机端口，服务端打印时把端口号替换成了 PORT）：GET http://127.0.0.1:PORT/a?x=1；同一个 URI 加上 host: example.com；GET http://127.0.0.1:PORT（没有路径）；GET /relative。四行各打印什么？

hyper = { version = "=1.12.0", features = ["client", "http1"] }
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
tokio = { version = "=1.54.1", features = ["macros", "rt", "net", "io-util"] }

use bytes::Bytes;
use http_body_util::{BodyExt, Empty};
use hyper::Request;
use hyper_util::client::legacy::Client;
use hyper_util::rt::TokioExecutor;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let port = listener.local_addr().unwrap().port();
  // 服务端：打印请求行和 host 头（端口号换成 PORT），回 200
  tokio::spawn(async move {
    let (mut stream, _) = listener.accept().await.unwrap();
    let mut buf = Vec::new();
    loop {
      let mut chunk = [0u8; 1024];
      let k = stream.read(&mut chunk).await.unwrap();
      if k == 0 {
        return;
      }
      buf.extend_from_slice(&chunk[..k]);
      while let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
        let head = String::from_utf8_lossy(&buf[..end]).replace(&port.to_string(), "PORT");
        buf.drain(..end + 4);
        let mut lines = head.lines();
        let line = lines.next().unwrap().to_string();
        let host = lines.find(|l| l.to_ascii_lowercase().starts_with("host:")).unwrap_or("(no host)");
        println!("server saw: {line:<20} {host}");
        stream.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n").await.unwrap();
      }
    }
  });

  let client = Client::builder(TokioExecutor::new()).build_http::<Empty<Bytes>>();
  let base = format!("http://127.0.0.1:{port}");
  let reqs = [
    Request::get(format!("{base}/a?x=1")).body(Empty::new()).unwrap(),
    Request::get(format!("{base}/a?x=1")).header("host", "example.com").body(Empty::new()).unwrap(),
    Request::get(base.clone()).body(Empty::new()).unwrap(),
    Request::get("/relative").body(Empty::new()).unwrap(),
  ];
  for req in reqs {
    match client.request(req).await {
      Ok(resp) => {
        resp.into_body().collect().await.unwrap();
      }
      Err(e) => println!("client error: {e} / {e:?}"),
    }
  }
}

发出去之后：连接什么时候回池

请求发出、响应头回来以后，try_send_request 处理手里这条池化连接（Pooled，drop 时尝试放回池里）：

        if pooled.is_http2() || !pooled.is_pool_enabled() || pooled.is_ready() {
            drop(pooled);
        } else {
            let on_idle = poll_fn(move |cx| pooled.poll_ready(cx)).map(|_| ());
            self.exec.execute(on_idle);
        }

HTTP/2 连接本来就是共享的，立即放回。HTTP/1 连接如果此时 is_ready() 为假（比如还在读响应 body），就 spawn 一个小任务等它的 poll_ready，即等连接任务读完 body、再次说「要」，然后 drop，连接回池。body 一直不读完，这个任务就一直等，连接就一直不回池。is_ready() 在请求在途时误报为真的问题，就是 hyper v1.12.0 修的 #4208（见 conn::http1 客户端）：误报会让这里把一条忙着的连接立即放回池里。Pooled 被 drop 时，池还会先用 is_open（没被标记为 poisoned 且 is_ready()）检查一次，不再可用的连接不放回去，见连接池。

retry_canceled_requests：只重发没写出去的

Builder 上的文档：

Set whether to retry requests that get disrupted before ever starting
to write.

This means a request that is queued, and gets given an idle, reused
connection, and then encounters an error immediately as the idle
connection was found to be unusable.

判断「还没开始写」不靠猜，靠 hyper 有没有把请求还回来。try_send_request 用的是 SendRequest::try_send_request，失败时：

        let mut res = match pooled.try_send_request(req).await {
            Ok(res) => res,
            Err(mut err) => {
                return if let Some(req) = err.take_message() {
                    Err(TrySendError::Retryable {
                        connection_reused: pooled.is_reused(),
                        error: e!(Canceled, err.into_error())
                            .with_connect_info(pooled.conn_info.clone()),
                        req,
                    })
                } else {
                    Err(TrySendError::Nope(
                        e!(SendRequest, err.into_error())
                            .with_connect_info(pooled.conn_info.clone()),
                    ))
                };
            }
        };

错误里带着请求，就是 hyper 保证它没被写出去（conn::http1 客户端的失败表里「能拿回请求」的三行），包成 Retryable；拿不回请求的包成 Nope，种类是 SendRequest。外层的 send_request 是个循环：

        loop {
            req = match self.try_send_request(req, pool_key.clone()).await {
                Ok(resp) => return Ok(resp),
                Err(TrySendError::Nope(err)) => return Err(err),
                Err(TrySendError::Retryable {
                    mut req,
                    error,
                    connection_reused,
                }) => {
                    if !self.config.retry_canceled_requests || !connection_reused {
                        // if client disabled, don't retry
                        // a fresh connection means we definitely can't retry
                        return Err(error);
                    }

                    trace!(
                        "unstarted request canceled, trying again (reason={:?})",
                        error
                    );
                    *req.uri_mut() = uri.clone();
                    req
                }
            }
        }

重发要同时满足三条：请求被还回来、开着 retry_canceled_requests、这条连接是从池里复用的（is_reused）。新建的连接上出这种错，说明对端一连上就不行，再建一条多半也一样，所以不重试。重发前把 URI 恢复成原来的绝对形式，因为上一轮已经把它改成了 origin-form。循环本身没有次数上限，靠「新连接不重试」保证会停。

这个窗口有多宽？下面的程序模拟最常见的场景：服务端关掉一条空闲的 keep-alive 连接，客户端紧接着从池里拿到它发请求。服务端是手写的：路径是 /hold 时回完响应把 socket 交给 main，由 main 关掉。每种设置跑 200 轮，用一个 tracing 层数 hyper-util 打出的「unstarted request canceled, trying again」，按结果分类计数。

hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
tokio = { version = "=1.54.1", features = ["macros", "rt-multi-thread", "net", "time", "io-util", "sync"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", default-features = false, features = ["std", "registry"] }

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty};
use hyper_util::client::legacy::Client;
use hyper_util::rt::TokioExecutor;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc;
use tracing_subscriber::layer::{Context, SubscriberExt};
use tracing_subscriber::Layer;

// 数 hyper-util 打出的 "unstarted request canceled, trying again" 这条 trace
static RETRIES: AtomicUsize = AtomicUsize::new(0);
struct CountRetries;
struct Msg(String);
impl tracing::field::Visit for Msg {
  fn record_debug(&mut self, f: &tracing::field::Field, v: &dyn std::fmt::Debug) {
    if f.name() == "message" {
      self.0 = format!("{v:?}");
    }
  }
}
impl<S: tracing::Subscriber> Layer<S> for CountRetries {
  fn on_event(&self, e: &tracing::Event<'_>, _: Context<'_, S>) {
    let mut m = Msg(String::new());
    e.record(&mut m);
    if m.0.starts_with("unstarted request canceled, trying again") {
      RETRIES.fetch_add(1, Ordering::SeqCst);
    }
  }
}

// 服务端：回 200；路径是 /hold 时回完把 socket 交给 main
async fn serve(listener: TcpListener, hand_over: mpsc::UnboundedSender<TcpStream>) {
  loop {
    let (mut stream, _) = listener.accept().await.unwrap();
    let hand_over = hand_over.clone();
    tokio::spawn(async move {
      let mut buf = Vec::new();
      loop {
        let mut chunk = [0u8; 1024];
        let k = stream.read(&mut chunk).await.unwrap_or(0);
        if k == 0 {
          return;
        }
        buf.extend_from_slice(&chunk[..k]);
        while let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
          let head: Vec<u8> = buf.drain(..end + 4).collect();
          stream.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok").await.unwrap();
          if head.windows(5).any(|w| w == b"/hold") {
            hand_over.send(stream).unwrap();
            return;
          }
        }
      }
    });
  }
}

#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
async fn main() {
  tracing::subscriber::set_global_default(tracing_subscriber::registry().with(CountRetries)).unwrap();
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let base = format!("http://{}", listener.local_addr().unwrap());
  let (hand_over, mut held) = mpsc::unbounded_channel();
  tokio::spawn(serve(listener, hand_over));

  for retry in [true, false] {
    let client = Client::builder(TokioExecutor::new())
      .retry_canceled_requests(retry)
      .build_http::<Empty<Bytes>>();
    let mut tally: BTreeMap<String, usize> = BTreeMap::new();
    for _ in 0..200 {
      let resp = client.get(format!("{base}/hold").parse().unwrap()).await.unwrap();
      resp.into_body().collect().await.unwrap();
      tokio::time::sleep(Duration::from_millis(5)).await;
      let server_side = held.recv().await.unwrap();
      let before = RETRIES.load(Ordering::SeqCst);
      drop(server_side); // 服务端关掉空闲连接，客户端紧接着发请求
      let r = client.get(format!("{base}/next").parse().unwrap()).await;
      let retried = RETRIES.load(Ordering::SeqCst) - before;
      let key = match r {
        Ok(resp) => format!("Ok({}), retries={retried}", resp.status().as_u16()),
        Err(e) => format!("Err({e:?}), retries={retried}"),
      };
      *tally.entry(key).or_default() += 1;
    }
    println!("retry_canceled_requests({retry}):");
    for (k, v) in tally {
      println!("  {v:>3} x {k}");
    }
  }
}

平台 M2 Max、rustc 1.99.0（tracing 0.1.44、tracing-subscriber 0.3.23），跑 3 次，计数每次不同：

retry_canceled_requests(true):
  167 x Err(hyper_util::client::legacy::Error(SendRequest, hyper::Error(IncompleteMessage))), retries=0
    2 x Ok(200), retries=0
   31 x Ok(200), retries=1
retry_canceled_requests(false):
    3 x Err(hyper_util::client::legacy::Error(Canceled, hyper::Error(Canceled, "connection closed"))), retries=0
  197 x Err(hyper_util::client::legacy::Error(SendRequest, hyper::Error(IncompleteMessage))), retries=0
---
retry_canceled_requests(true):
  195 x Err(hyper_util::client::legacy::Error(SendRequest, hyper::Error(IncompleteMessage))), retries=0
    2 x Err(hyper_util::client::legacy::Error(SendRequest, hyper::Error(Io, Os { code: 54, kind: ConnectionReset, message: "Connection reset by peer" }))), retries=0
    3 x Ok(200), retries=1
retry_canceled_requests(false):
    8 x Err(hyper_util::client::legacy::Error(Canceled, hyper::Error(Canceled, "connection closed"))), retries=0
  192 x Err(hyper_util::client::legacy::Error(SendRequest, hyper::Error(IncompleteMessage))), retries=0
---
retry_canceled_requests(true):
  194 x Err(hyper_util::client::legacy::Error(SendRequest, hyper::Error(IncompleteMessage))), retries=0
    1 x Err(hyper_util::client::legacy::Error(SendRequest, hyper::Error(Io, Os { code: 54, kind: ConnectionReset, message: "Connection reset by peer" }))), retries=0
    5 x Ok(200), retries=1
retry_canceled_requests(false):
    2 x Err(hyper_util::client::legacy::Error(Canceled, hyper::Error(Canceled, "connection closed"))), retries=0
  198 x Err(hyper_util::client::legacy::Error(SendRequest, hyper::Error(IncompleteMessage))), retries=0

（--- 是三次运行之间的分隔。）四种结果对应请求和「发现连接已关」之间的四种先后：

结果	发生了什么
`Ok(200), retries=0`	连接任务先发现了 EOF，连接退出、`is_ready()` 变假，池不再借出它，请求直接用了一条新连接
`Ok(200), retries=1`	请求已经放进这条连接的通道，连接任务在写它之前发现 EOF（`poll_loop` 每圈先读后写），请求随 `Envelope` 被还回来，换新连接重发
`Canceled, "connection closed"`	同上一行，但关了重试，错误原样返回
`SendRequest, IncompleteMessage` 或 `ConnectionReset`	连接任务在发现 EOF 之前把请求写进了 socket，然后才读到 EOF（或对端回了 RST），请求拿不回来

最后一种占了绝大多数。原因在 tokio：客户端 socket 上的 FIN 要等反应器下一次 turn 才会置上就绪位，在此之前 poll_read 看到的就绪位是旧的，直接返回 Pending，连接任务于是往下去写（见 ScheduledIo 与就绪状态、从 TcpStream::read 到挂起）。main 关掉 socket 后马上发请求，请求进通道、连接任务被唤醒，这一串通常赶在反应器处理 FIN 之前。retry_canceled_requests 救下的只是「请求进了通道，但连接任务被唤醒之前反应器已经处理过 FIN」的那一小段。

在真实网络里，服务端的 FIN 要走一个往返才到，这个窗口的形状会不同，但结论不变：服务端的 keep-alive 超时比客户端池的空闲超时短，就会有请求撞上刚被关掉的连接，其中一部分以 IncompleteMessage（connection closed before message completed）失败而不会自动重试。

📝NOTE

没有蠢问题

问：那些 IncompleteMessage 的请求，服务端其实一个都没处理（socket 已经关了），为什么不也重发？

答：客户端不知道。它只知道请求的字节已经交给了内核，之后读到了 EOF；服务端是在读到请求之前关的，还是读到、处理了、来不及回就崩了，从客户端看是一样的。对 POST 这种非幂等请求重发，可能让服务端执行两次。hyper 的规则只看一件事：请求有没有开始写。开始写了就不重发，交给上层决定。要减少这类失败，一是把客户端池的空闲超时设得比服务端的 keep-alive 超时短（见连接池），二是对幂等请求在外面套一层重试，比如 tower 的 Retry（见 Policy 与重试状态机）。

「请求还没写出去就换连接重发」这条规则，在 hyper-util 里是分三步才走到现在这个样子的。

📝NOTE

冷知识：重试循环在 hyper-util 里被注释掉过

hyper 0.14 的 Client 能重发未开始的请求，因为池和连接在同一个 crate 里，内部有私有的 send_request_retryable。1.0 把 Client 搬到 hyper-util 以后，hyper 没有公开的办法把请求还回来，hyper-util v0.1.5 的 client.rs 里，这段循环整个是注释（/* async fn retryably_send_request(...) */）。

hyper 的 #3691（v1.4.0）加了 SendRequest::try_send_request 和公开的 TrySendError。PR 写的问题是：「If you try to send_request(req) at the same time the server is closing the connection, it's possible hyper will notice the HUP before the request gets serialized」，可「there isn't a mechanism for hyper to give the request back. And since the pool was moved out of hyper, it doesn't know how to just grab another connection.」

hyper-util 的 #133（v0.1.6，CHANGELOG：「Fix legacy client to better use dying pooled connections」）用它恢复了循环，但新的 TrySendError::Retryable 里没有带上 connection_reused，判断只剩 !self.config.retry_canceled_requests。#150（v0.1.9）报告了后果：服务端接受连接后立刻关掉时，「the client opens a new connection and may try opening connections many many times」。修法是把「连接是否复用过」加回 Retryable，也就是上面代码里的 || !connection_reused；CHANGELOG 写作「Fix client::legacy wrongly retrying pooled connections that were created successfully but failed immediately after, resulting in a retry loop.」

capture_connection 与 poison：看见并废掉某一条连接

Client 不对外暴露池里的连接，但留了一个口子：capture_connection 往请求的 extensions 里放一个 watch::Sender，返回接收端 CaptureConnection。try_send_request 拿到连接后，在第 286 行把这条连接的 Connected 写进去：

        if let Some(conn) = req.extensions_mut().get_mut::<CaptureConnectionExtension>() {
            conn.set(&pooled.conn_info);
        }

set 是 send_replace，所以上一节的重试发生时，后一条连接会覆盖前一条，调用方最后看到的是真正发出请求的那条。拿到的 Connected 能做的事：读 is_proxied()、is_negotiated_h2()，用 get_extras 取出连接器放进去的附加信息（HttpConnector 放的是 HttpInfo，含本地和远端地址），以及 poison()。Connected 内部的 poisoned 是一个 Arc<AtomicBool>，克隆出去的和池里那条连接的 conn_info 共享，所以在外面 poison，池里的那条就被标记了；PoolClient::is_open 是 !self.is_poisoned() && self.is_ready()，被标记的连接不再被取出，见连接池。

实测：同一个 Client 连发三个请求，第一个请求读完响应后 poison 它的连接，等 300ms 再发第二个：

  for (i, poison) in [(1, true), (2, false), (3, false)] {
    let mut req = Request::get(url.as_str()).body(Empty::new()).unwrap();
    let cap = capture_connection(&mut req);
    let resp = client.request(req).await.unwrap();
    resp.into_body().collect().await.unwrap();
    let meta = cap.connection_metadata();
    let c = meta.as_ref().unwrap();
    let mut ext = http::Extensions::new();
    c.get_extras(&mut ext);
    let local = ext.get::<HttpInfo>().map(|h| h.local_addr().port());
    println!("request {i}: local port {local:?}, proxied={}, h2={}", c.is_proxied(), c.is_negotiated_h2());
    if poison {
      c.poison();
      println!("request {i}: poisoned, sleeping 300ms");
      tokio::time::sleep(Duration::from_millis(300)).await;
      println!("request {i}: woke up");
    }
    // ...
  }

  server: accept #1
request 1: local port Some(50886), proxied=false, h2=false
request 1: poisoned, sleeping 300ms
request 1: woke up
  server: close #1
  server: accept #2
request 2: local port Some(50897), proxied=false, h2=false
request 3: local port Some(50897), proxied=false, h2=false
accepts: 2

第二个请求换了一条新连接，第三个复用第二条。注意服务端看到第一条连接关闭的时刻：不是 poison() 的时候，而是 300ms 后第二个请求 checkout 时。poison 只是打标记，连接还在池里空闲着，直到下一次 checkout 把它挑出来丢掉（或者清理任务按空闲超时清掉），连接任务才随 SendRequest 被 drop 而结束。

wait_for_connection_metadata 的文档说，请求一直没连上（比如 DNS 失败）时「this method will never return」。实际上，request 失败返回时请求被 drop，extensions 里的 Sender 随之 drop，changed() 返回错误，方法返回 None。用一个连不上的 SOCKS 代理让连接失败，请求返回后等它 2 秒：

D: capture_connection on failed connect
  request is_err=true
  wait returned, is_some=false

立即返回了 None，没有挂住。不过它返回的时机取决于请求什么时候被 drop：如果在 request 还在进行时就去等，它会一直等到请求结束。

错误

Error 在 hyper 的错误外面再包一层，种类是私有的 ErrorKind：

种类	来源
`Connect`	连接器失败，或取连接失败且不重试
`Canceled`	请求被还回来但没有重发（关了重试，或连接是新建的）
`SendRequest`	请求已开始发送后失败；也包括 `handshake`、`ready()` 失败
`ChannelClosed`	`PoolClient::poll_ready` 失败（`Error::closed`）；唯一的调用者是上面的 `on_idle`，结果被丢掉，所以调用方见不到
`UserUnsupportedRequestMethod`、`UserUnsupportedVersion`、`UserAbsoluteUriRequired`	`request` 和 `try_send_request` 里的检查

Display 只打印 client error (种类)，hyper 的错误在 source() 里，所以日志里只打 {} 会看不到 IncompleteMessage。公开的判断只有 is_connect()，另有 connect_info() 取出那条连接的 Connected 信息。hyper 自己的错误种类见 hyper::Error。

Client 也是 tower 的 Service

Client<C, B> 和 &Client<C, B> 都实现了 tower_service::Service<Request<B>>：

    fn poll_ready(&mut self, _: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: Request<B>) -> Self::Future {
        self.request(req)
    }

poll_ready 永远就绪。Client 对同一个 host 的并发连接数没有上限：池里没有空闲连接就新建一条，单条 HTTP/1 连接上的「一次一个」不会传到 Client 这一层。需要背压就在外面套 tower 的中间件，比如 ConcurrencyLimit；poll_ready 的约定见 Service trait。

动脑筋答案

server saw: GET /a?x=1 HTTP/1.1  host: 127.0.0.1:PORT
server saw: GET /a?x=1 HTTP/1.1  host: example.com
server saw: GET / HTTP/1.1       host: 127.0.0.1:PORT
client error: client error (UserAbsoluteUriRequired) / hyper_util::client::legacy::Error(UserAbsoluteUriRequired)

跑了 3 次，输出相同。

第 1 行：绝对 URI 被 origin_form 改成 /a?x=1；Host 是补上的，端口不是 80，所以带端口。
第 2 行：请求自带的 host: example.com 原样保留，or_insert_with 不覆盖。连接仍然连向 127.0.0.1，因为池的键和连接目标都来自 URI，不看 Host 头。
第 3 行：没有路径的 URI 改成 /。
第 4 行：相对 URI 在 request 里就被拒绝，服务端什么也没收到。Display 只有 client error (UserAbsoluteUriRequired)。三个请求共用了一条连接（服务端只 accept 了一次）。

小结

Client 是 hyper 0.14 客户端的移植：池的键是 (Scheme, Authority)，每条连接用 hyper 的 conn::http1 / conn::http2 建立，Connection 带 with_upgrades() 交给执行器，等 ready() 后才算建好。
HTTP/1 连接上发请求前补 Host（已有不覆盖，默认端口不写）并把 URI 改成 origin-form（代理连接保留 absolute-form，CONNECT 用 authority-form）；HTTP/2 不改。
HTTP/2 连接响应回来就放回池；HTTP/1 连接要等连接任务再次说「要」（body 读完）才回池。
retry_canceled_requests 只重发 hyper 还回来的请求，并且只在复用来的连接上；新连接不重试（#150）。请求开始写以后失败一律不重发。
服务端关掉空闲连接时，实测大多数撞上的请求已经写出，以 IncompleteMessage 失败；重试只覆盖很窄的一段时间。
错误的 Display 只有种类，真实原因在 source()；Client 的 poll_ready 永远就绪，没有背压。
capture_connection 拿到实际用的那条连接的 Connected（重试时被覆盖）；poison() 只打标记，连接在下次 checkout 时才被丢掉。请求失败后 wait_for_connection_metadata 返回 None，不像文档说的永远不返回。