there are extra bytes after body has been decompressed

tower-http = { version = "=0.7.1", features = ["decompression-gzip"] }
tower = { version = "=0.5.3", features = ["util"] }
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http = "1"
http-body-util = "0.1"
bytes = "1"
tokio = { version = "=1.54.1", features = ["macros", "rt", "net", "io-util", "time"] }
reqwest = { version = "0.12", default-features = false, features = ["gzip"] }
flate2 = "1"

use bytes::Bytes;
use http_body_util::{BodyExt, Empty};
use hyper_util::{client::legacy::Client, rt::TokioExecutor};
use std::{error::Error, io::Write, time::Duration};
use tokio::{io::{AsyncReadExt, AsyncWriteExt}, net::TcpListener};
use tower::{ServiceBuilder, ServiceExt};
use tower_http::decompression::DecompressionLayer;

fn gzip(data: &[u8]) -> Vec<u8> {
  let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
  e.write_all(data).unwrap();
  e.finish().unwrap()
}

fn chunk(data: &[u8]) -> Vec<u8> {
  let mut v = format!("{:x}\r\n", data.len()).into_bytes();
  v.extend_from_slice(data);
  v.extend_from_slice(b"\r\n");
  v
}

// 每个请求都回两段拼接的 gzip；/split 时两段各占一个 chunk，中间停一下
async fn serve(listener: TcpListener) {
  loop {
    let (mut sock, _) = listener.accept().await.unwrap();
    tokio::spawn(async move {
      let mut buf = [0u8; 4096];
      let n = sock.read(&mut buf).await.unwrap();
      let split = String::from_utf8_lossy(&buf[..n]).contains("GET /split");
      let (a, b) = (gzip(b"hello "), gzip(b"world"));
      sock.write_all(b"HTTP/1.1 200 OK\r\ncontent-encoding: gzip\r\ntransfer-encoding: chunked\r\nconnection: close\r\n\r\n").await.unwrap();
      if split {
        sock.write_all(&chunk(&a)).await.unwrap();
        sock.flush().await.unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;
        sock.write_all(&chunk(&b)).await.unwrap();
      } else {
        sock.write_all(&chunk(&[a, b].concat())).await.unwrap();
      }
      sock.write_all(b"0\r\n\r\n").await.unwrap();
    });
  }
}

// 把错误链拼成一行
fn chain(e: &dyn Error) -> String {
  let mut s = e.to_string();
  let mut cur = e.source();
  while let Some(c) = cur {
    s.push_str(&format!(" <- {c}"));
    cur = c.source();
  }
  s
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(serve(listener));

  let hyper = Client::builder(TokioExecutor::new()).build_http::<Empty<Bytes>>();
  let reqwest = reqwest::Client::builder().no_proxy().build().unwrap();
  for path in ["/one", "/split"] {
    let url = format!("http://{addr}{path}");
    let svc = ServiceBuilder::new().layer(DecompressionLayer::new()).service(hyper.clone());
    let res = svc.oneshot(http::Request::get(&url).body(Empty::new()).unwrap()).await.unwrap();
    match res.into_body().collect().await {
      Ok(b) => println!("tower-http 0.7.1 {path:<6} -> {:?}", b.to_bytes()),
      Err(e) => println!("tower-http 0.7.1 {path:<6} -> error: {}", chain(&*e)),
    }
    let res = reqwest.get(&url).send().await.unwrap();
    match res.text().await {
      Ok(t) => println!("reqwest 0.12    {path:<6} -> {t:?}"),
      Err(e) => println!("reqwest 0.12    {path:<6} -> error: {}", chain(&e)),
    }
  }
}

tower-http 0.7.1 /one   -> b"hello "
reqwest 0.12    /one   -> "hello "
tower-http 0.7.1 /split -> error: there are extra bytes after body has been decompressed
reqwest 0.12    /split -> error: error decoding response body <- there are extra bytes after body has been decompressed

            Some(Ok(frame)) => match frame.into_data() {
                // Payload after the decompressor reported end-of-stream means
                // the body and the compressed stream disagree about where the
                // content ends.
                Ok(data) if data.has_remaining() => Poll::Ready(Some(Err(
                    "there are extra bytes after body has been decompressed".into(),
                ))),