Rust in Everything
基于 hyper 1.12.0

connection closed before message completed

主线位置：两条主线的「读消息头」一步。客户端主线上，请求已经写出、响应头还没读全，连接读到了 EOF；服务端主线上，请求头读到一半，客户端走了。两边报的是同一个错误。
flowchart TB
  eof["读到 EOF"] --> where{"这时在等什么"}
  where -->|"空闲，缓冲为空"| clean["正常关闭，不报错"]
  where -->|"消息头没收全"| inc["IncompleteMessage"]
  where -->|"body 没收全"| bodyerr["Body 错误，另一个报错"]
  inc --> c1["复用了刚被关的空闲连接"]
  inc --> c2["对端在响应头中途关闭"]
  inc --> c3["请求进了直接断开的代理"]

这行字是 Kind::IncompleteMessage 的 Display（error.rs 第 532 行），判断它用 is_incomplete_message。经过 hyper-util 的 legacy Client（reqwest 也是）时，外面再包一层 client error (SendRequest)，只打印 {} 看不到它，要沿着 source() 往下找（见 legacy Client 的「错误」一节）。

这行字从哪来

读消息头时，Buffered::parse 每次解析不完整就再从 IO 读一次，读到 0 个字节就返回它：

            if ready!(self.poll_read_from_io(cx)).map_err(crate::Error::new_io)? == 0 {
                trace!("parse eof");
                return Poll::Ready(Err(crate::Error::new_incomplete()));
            }

拿到这个错误的 Conn::on_read_head_error 再决定它算不算错：读缓冲里有半截消息，或者 should_error_on_eof 为真（客户端角色，且连接不在空闲状态，也就是请求已经发出去在等响应），就交给上层；否则是一次正常的关闭，连接安静地结束。所以服务端上一条什么都没收到的连接被客户端关掉，serve_connection 返回 Ok(())；收到半个请求头就被关掉，返回 IncompleteMessage。

另外几处也构造这个错误：客户端连接空闲时读到 EOF 走 require_empty_read，空闲则正常关闭，忙则报它；消息进行到一半、没在读的时候读到 EOF 走 mid_message_detect_eof；连接任务被 drop 时还拿着 body 的发送端，SenderDropGuard 往 body 里发一个它（这一种放在 dispatch task is gone 里讲）。

复现一：服务端先关了空闲连接

线上最常见的来源。服务端（或者中间的负载均衡）给 keep-alive 连接设了空闲上限，客户端的连接池不知道这个上限，按自己的 pool_idle_timeout（默认 90 秒）留着连接。下一个请求的间隔恰好落在服务端的上限附近，就会拿到一条服务端正在关的连接。

下面的服务端用 hyper 自己的 header_read_timeout 做空闲上限：它同时管 keep-alive 连接的空闲时间（Executor 与 Timer 动脑筋里的 E），这里设成 100ms。客户端是默认配置的 legacy Client，每个间隔发 40 个请求，按结果计数；第二轮把客户端的 pool_idle_timeout 改成 80ms，比服务端短。

hyper = { version = "=1.12.0", features = ["client", "server", "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-multi-thread", "net", "time", "io-util"] }

use std::collections::BTreeMap;
use std::convert::Infallible;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty, Full};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::client::legacy::Client;
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use tokio::net::TcpListener;

// Server: closes a keep-alive connection after 100ms without a new request head.
const SERVER_IDLE: Duration = Duration::from_millis(100);

async fn serve(listener: TcpListener) {
  loop {
    let (stream, _) = listener.accept().await.unwrap();
    tokio::spawn(async move {
      let svc = service_fn(|_req: Request<hyper::body::Incoming>| async {
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"ok"))))
      });
      let _ = http1::Builder::new()
        .timer(TokioTimer::new())
        .header_read_timeout(SERVER_IDLE)
        .serve_connection(TokioIo::new(stream), svc)
        .await;
    });
  }
}

fn short(e: &hyper_util::client::legacy::Error) -> String {
  let mut src = std::error::Error::source(e);
  let mut last = e.to_string();
  while let Some(inner) = src {
    last = inner.to_string();
    src = inner.source();
  }
  last
}

#[tokio::main]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let url: hyper::Uri = format!("http://{}/", listener.local_addr().unwrap()).parse().unwrap();
  tokio::spawn(serve(listener));

  let default_pool = Client::builder(TokioExecutor::new()).build_http::<Empty<Bytes>>();
  let mut short_pool = Client::builder(TokioExecutor::new());
  short_pool.pool_idle_timeout(Duration::from_millis(80));
  let short_pool = short_pool.build_http::<Empty<Bytes>>();

  for (name, client) in [("pool_idle_timeout 90s (default)", &default_pool), ("pool_idle_timeout 80ms", &short_pool)] {
    println!("{name}:");
    for gap_ms in [50u64, 95, 98, 99, 100, 101, 102, 105, 150] {
      let mut tally: BTreeMap<String, usize> = BTreeMap::new();
      for _ in 0..40 {
        let r = client.get(url.clone()).await;
        let key = match r {
          Ok(resp) => {
            resp.into_body().collect().await.unwrap();
            "ok".to_string()
          }
          Err(e) => short(&e),
        };
        *tally.entry(key).or_default() += 1;
        tokio::time::sleep(Duration::from_millis(gap_ms)).await;
      }
      println!("  gap {gap_ms:>3}ms: {tally:?}");
    }
  }
}

M2 Max、rustc 1.99.0、--release，第一次运行：

pool_idle_timeout 90s (default):
  gap  50ms: {"ok": 40}
  gap  95ms: {"ok": 40}
  gap  98ms: {"connection closed before message completed": 16, "ok": 24}
  gap  99ms: {"connection closed before message completed": 20, "ok": 20}
  gap 100ms: {"Connection reset by peer (os error 54)": 1, "connection closed before message completed": 18, "ok": 21}
  gap 101ms: {"connection closed before message completed": 19, "ok": 21}
  gap 102ms: {"connection closed before message completed": 16, "ok": 24}
  gap 105ms: {"ok": 40}
  gap 150ms: {"ok": 40}
pool_idle_timeout 80ms:
  gap  50ms: {"ok": 40}
  gap  95ms: {"ok": 40}
  gap  98ms: {"ok": 40}
  gap  99ms: {"ok": 40}
  gap 100ms: {"ok": 40}
  gap 101ms: {"ok": 40}
  gap 102ms: {"ok": 40}
  gap 105ms: {"ok": 40}
  gap 150ms: {"ok": 40}

又跑了两次，形状不变：默认池在 98 到 102ms 这五档里每档失败 13 到 20 次，偶尔夹一次 Connection reset by peer，其他间隔一次不错；80ms 的池三次都是全部成功。

  • 只在一个几毫秒宽的窗口里出错。 间隔 95ms，服务端还没关，复用成功；间隔 105ms 或 150ms，服务端早就关了，客户端的连接任务已经读到 EOF 并退出，池子不再借出它（连接池里 checkout 检查 is_open），请求用了一条新连接。只有请求恰好在「服务端关」和「客户端发现关了」之间发出，才会写进一条已经关掉的连接。线上这个窗口的位置就是服务端的 keep-alive 上限，所以它表现为「偶发」，和负载、和代码改动都对不上。
  • 失败和成功大致交替。 一次失败之后，那条连接没了，下一个请求新建连接，必然成功；新连接回到池里，再下一个请求又撞进窗口。
  • 重试没有救下它们。 legacy Client 默认开着 retry_canceled_requests，但它只重发「还没写出去」的请求。这里的请求已经写进了 socket，hyper 拿不回来（legacy Client 里的 race 实验专门量过这个窗口里各种结果的比例）。
  • ConnectionReset 是同一个窗口里的另一种结果。 服务端关 socket 时，接收缓冲里如果已经有客户端刚写进去、还没被读走的请求，内核回的是 RST 而不是 FIN，客户端读到的就是 Connection reset by peer（IO 错误），而不是 EOF。

修法是让客户端先放手：客户端池的空闲上限比服务端的 keep-alive 上限短，连接在服务端关之前就被客户端丢掉。80ms 那一轮就是这样。上限要比对端短多少，取决于两边的计时精度和网络延迟；对端的上限不知道时，只能对幂等请求在外面加一层重试（比如 tower 的 Retry，见 Policy 与重试状态机）。

复现二：对端在消息中途关闭

不经过连接池也能撞上：对端读完请求以后关连接，不管是崩了、被 OOM 杀了，还是主动关。下面用一个裸 TCP 服务端，读完请求头写一段事先准备好的字节就关 socket；最后一项反过来，客户端发了半个请求头就走，看 hyper 服务端报什么。依赖同上。

use std::convert::Infallible;
use std::error::Error as _;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty, Full};
use hyper::client::conn::http1 as client_h1;
use hyper::server::conn::http1 as server_h1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

fn chain(e: &hyper::Error) -> String {
  let mut s = e.to_string();
  let mut src = e.source();
  while let Some(inner) = src {
    s.push_str(&format!(": {inner}"));
    src = inner.source();
  }
  s
}

// A raw server that reads one request head, writes `reply`, then closes.
async fn raw_server(reply: &'static [u8]) -> std::net::SocketAddr {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    let (mut stream, _) = listener.accept().await.unwrap();
    let mut buf = vec![0u8; 4096];
    let mut got = Vec::new();
    while !got.windows(4).any(|w| w == b"\r\n\r\n") {
      let n = stream.read(&mut buf).await.unwrap();
      got.extend_from_slice(&buf[..n]);
    }
    stream.write_all(reply).await.unwrap();
    // dropping `stream` closes the connection
  });
  addr
}

async fn client_case(name: &str, reply: &'static [u8]) {
  let addr = raw_server(reply).await;
  let io = TokioIo::new(TcpStream::connect(addr).await.unwrap());
  let (mut tx, conn) = client_h1::handshake::<_, Empty<Bytes>>(io).await.unwrap();
  tokio::spawn(conn);
  let req = Request::get("/").header("host", "x").body(Empty::new()).unwrap();
  match tx.send_request(req).await {
    Err(e) => println!("{name:<24} send_request: Err({}) incomplete={}", chain(&e), e.is_incomplete_message()),
    Ok(resp) => {
      let status = resp.status();
      match resp.into_body().collect().await {
        Ok(b) => println!("{name:<24} {status}, body {:?}", b.to_bytes()),
        Err(e) => println!("{name:<24} {status}, body: Err({}) incomplete={}", chain(&e), e.is_incomplete_message()),
      }
    }
  }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  println!("client side:");
  client_case("close, no response", b"").await;
  client_case("close inside head", b"HTTP/1.1 200 OK\r\ncontent-le").await;
  client_case("close inside body", b"HTTP/1.1 200 OK\r\ncontent-length: 10\r\n\r\nabc").await;
  client_case("close inside chunked", b"HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n3\r\nabc\r\n").await;

  println!("server side:");
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let svc = service_fn(|_req: Request<hyper::body::Incoming>| async {
      Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"ok"))))
    });
    server_h1::Builder::new().serve_connection(TokioIo::new(stream), svc).await
  });
  let mut tcp = TcpStream::connect(addr).await.unwrap();
  tcp.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n").await.unwrap(); // head never finished
  tokio::time::sleep(Duration::from_millis(50)).await;
  drop(tcp);
  match server.await.unwrap() {
    Ok(()) => println!("{:<24} serve_connection: Ok(())", "client left inside head"),
    Err(e) => println!("{:<24} serve_connection: Err({}) incomplete={}", "client left inside head", chain(&e), e.is_incomplete_message()),
  }
}

连跑 3 次输出相同：

client side:
close, no response       send_request: Err(connection closed before message completed) incomplete=true
close inside head        send_request: Err(connection closed before message completed) incomplete=true
close inside body        200 OK, body: Err(error reading a body from connection: end of file before message length reached) incomplete=false
close inside chunked     200 OK, body: Err(error reading a body from connection: unexpected EOF during chunk size line) incomplete=false
server side:
client left inside head  serve_connection: Err(connection closed before message completed) incomplete=true

分界线在响应头上：头没收全就断，是 IncompleteMessage，错误出现在 send_request 上；头收全了，send_request 已经返回了 200 OK，body 读到一半断开，错误出现在读 body 时，是 Kind::Body（error reading a body from connection），原因在 source() 里，Length 解码器是 IncompleteBody 的「end of file before message length reached」，chunked 解码器是 chunk 大小行读到 EOF。按 is_incomplete_message() 判断「连接被对端断了」的代码会漏掉后两种。

服务端那一行对应线上日志里的这条错误：客户端发了半个请求头就断开（超时放弃、进程被杀、扫描器），serve_connection 返回 IncompleteMessage。这不是服务端的问题，记 debug 级别就够了。

📝NOTE
没有蠢问题

问： is_incomplete_message 的文档举的第二个例子是「A message body is only partially receive before the connection reports EOF」，为什么上面 body 读一半断开，它返回 false？

答： 文档开头写着「Some cases this could happen (not exhaustive)」，这一条和 1.12.0 的 HTTP/1 解码器对不上：Length 和 Chunked 两种解码器读到 EOF 返回的是 io::Error（UnexpectedEof），Dispatcher 把它包成 Kind::Body 送进 body。body 上真正会出现 IncompleteMessage 的，是连接任务被 drop 时 SenderDropGuard 补发的那一个（dispatch task is gone）。要判断「body 没收全」，看 body 返回了错误本身就够了，不要靠种类。

复现三：请求进了代理

第三种来源和 hyper 无关，但报的是同一句话：请求根本没到你的服务端，而是被送进了一个代理，代理读完请求就把连接关了。reqwest 默认读代理配置，本地测试打 127.0.0.1 时也会走代理。

下面的程序在进程里把 HTTP_PROXY 指向一个自己起的「代理」：它读一个请求头、打印请求行、关连接。NO_PROXY 清空，效果和 shell 里配了代理却没把 127.0.0.1 写进 NO_PROXY 一样。被测的服务端是一个正常的 hyper 服务。

reqwest = { version = "=0.12.28", default-features = false, features = ["system-proxy"] }
hyper = { version = "=1.12.0", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
tokio = { version = "=1.54.1", features = ["rt-multi-thread", "net", "io-util"] }

use std::convert::Infallible;
use std::error::Error as _;

use bytes::Bytes;
use http_body_util::Full;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::io::AsyncReadExt;
use tokio::net::TcpListener;

fn chain(e: &reqwest::Error) -> String {
  let mut s = e.to_string();
  let mut src = e.source();
  while let Some(inner) = src {
    s.push_str(&format!("\n    caused by: {inner}"));
    src = inner.source();
  }
  s
}

fn main() {
  // The "proxy": accepts, reads one request head, prints its first line, closes.
  let proxy = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
  let proxy_addr = proxy.local_addr().unwrap();
  // Same effect as a proxy configured in the shell, with no NO_PROXY.
  unsafe {
    std::env::set_var("HTTP_PROXY", format!("http://{proxy_addr}"));
    for k in ["http_proxy", "NO_PROXY", "no_proxy", "ALL_PROXY", "all_proxy"] {
      std::env::remove_var(k);
    }
  }

  let rt = tokio::runtime::Runtime::new().unwrap();
  rt.block_on(async move {
    let proxy = TcpListener::from_std({
      proxy.set_nonblocking(true).unwrap();
      proxy
    })
    .unwrap();
    tokio::spawn(async move {
      loop {
        let (mut s, _) = proxy.accept().await.unwrap();
        let mut buf = vec![0u8; 4096];
        let n = s.read(&mut buf).await.unwrap();
        let text = String::from_utf8_lossy(&buf[..n]);
        println!("  proxy got: {:?}", text.lines().next().unwrap_or(""));
      }
    });

    // The real server under test.
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}/", listener.local_addr().unwrap());
    tokio::spawn(async move {
      loop {
        let (stream, _) = listener.accept().await.unwrap();
        tokio::spawn(async move {
          let svc = service_fn(|_req: Request<hyper::body::Incoming>| async {
            Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"ok"))))
          });
          let _ = http1::Builder::new().serve_connection(TokioIo::new(stream), svc).await;
        });
      }
    });

    for (name, client) in [
      ("Client::new()", reqwest::Client::new()),
      ("Client::builder().no_proxy()", reqwest::Client::builder().no_proxy().build().unwrap()),
    ] {
      println!("{name} GET {url}");
      match client.get(&url).send().await {
        Ok(resp) => println!("  -> {} {:?}", resp.status(), resp.text().await.unwrap()),
        Err(e) => println!("  -> Err: {}", chain(&e)),
      }
    }
  });
}

连跑 3 次，除了端口号输出相同：

Client::new() GET http://127.0.0.1:56152/
  proxy got: "GET http://127.0.0.1:56152/ HTTP/1.1"
  -> Err: error sending request for url (http://127.0.0.1:56152/)
    caused by: client error (SendRequest)
    caused by: connection closed before message completed
Client::builder().no_proxy() GET http://127.0.0.1:56152/
  -> 200 OK "ok"

「代理」收到的请求行是绝对形式的 GET http://127.0.0.1:56152/，这是发给 HTTP 代理的写法，说明请求确实进了代理。代理不回响应就关，从 hyper 看就是「请求写出去了，响应头一个字节没到就 EOF」，和复现二的第一行一模一样。.no_proxy() 关掉 reqwest 的自动代理（auto_sys_proxy），同一个 URL 直连成功。

在 macOS 上，不设环境变量也可能撞上。reqwest 0.12.28 的自动代理用的是 hyper-util 的 Matcher::from_system，它先读环境变量，HTTP_PROXY / HTTPS_PROXY 为空时再由 mac::with_system 从系统设置里取代理地址。这个函数只读 HTTPEnable、HTTPProxy、HTTPPort 和对应的 HTTPS 三项，系统设置里的例外列表（ExceptionsList）它不读；NO_PROXY 只来自环境变量。所以一个从 IDE 或桌面启动、没有继承 shell 环境变量的测试进程，在开着系统代理的 Mac 上请求 127.0.0.1，会被送到系统代理那里。代理接下来怎么处理一个指向它自己机器回环地址的请求，取决于代理软件；回不了响应就关连接，就是这里的错误。上面的程序只用环境变量复现了「请求进了代理、代理关连接」这一段，系统设置那条路径是读源码得出的，没有跑。

修法

来源怎么认出来修法
复用了服务端刚关的空闲连接偶发，和请求间隔有关；同一个服务的大部分请求正常客户端池的空闲上限比服务端 keep-alive 上限短；幂等请求在外面加重试
对端在响应头中途关闭每次都失败，或者和对端崩溃、重启同时出现查对端；这时只能重试幂等请求
请求进了代理只打本地地址的测试失败；代理日志里能看到请求测试里用 .no_proxy() 构造 client，或者把回环地址写进 NO_PROXY
服务端日志里出现客户端在请求头中途断开不用修，降低日志级别

小结

  • IncompleteMessage 只说明一件事：消息头没收全就读到了 EOF。在客户端，是请求发出后响应头没到；在服务端，是请求头收到一半。空闲连接被关不算错。
  • 最常见的来源是连接池复用了服务端刚关掉的空闲连接，只在请求间隔接近服务端 keep-alive 上限的几毫秒内出现，自动重试救不了已经写出去的请求。让客户端的 pool_idle_timeout 比服务端短。
  • 响应头收全以后断开，报的是 Kind::Body，is_incomplete_message() 为假。
  • 请求被送进一个直接关连接的代理也是这句话。reqwest 默认读代理配置，macOS 上还会读系统代理但不读它的例外列表；测试里用 .no_proxy()。

💬 关联讨论

发起讨论
加载中...