header_read_timeout 不生效，半个前言挂住连接

axum = "=0.8.9"
hyper = { version = "=1.12.0", features = ["server", "http1", "http2"] }
hyper-util = { version = "=0.1.21", features = ["server-auto", "http1", "http2", "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::convert::Infallible;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

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

async fn hello(_: Request<hyper::body::Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  Ok(Response::new(Full::new(Bytes::from_static(b"ok"))))
}

// 1. axum::serve with its defaults.
async fn axum_server() -> SocketAddr {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let app = axum::Router::new().route("/", axum::routing::get(|| async { "ok" }));
  tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
  addr
}

// 2. Own accept loop: auto::Builder with a timer on both halves.
// 3. The same, plus a deadline for the first request to reach the service.
async fn own_server(first_request_deadline: Option<Duration>) -> SocketAddr {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    loop {
      let (stream, _) = listener.accept().await.unwrap();
      tokio::spawn(async move {
        let mut builder = auto::Builder::new(TokioExecutor::new());
        builder.http1().timer(TokioTimer::new()); // header_read_timeout: default 30s
        builder.http2().timer(TokioTimer::new());
        let seen = Arc::new(AtomicBool::new(false));
        let seen2 = seen.clone();
        let svc = service_fn(move |req| {
          seen2.store(true, Ordering::Relaxed);
          hello(req)
        });
        let conn = builder.serve_connection_with_upgrades(TokioIo::new(stream), svc);
        tokio::pin!(conn);
        let Some(deadline) = first_request_deadline else {
          let _ = conn.await;
          return;
        };
        tokio::select! {
          _ = conn.as_mut() => {}
          _ = tokio::time::sleep(deadline) => {
            if !seen.load(Ordering::Relaxed) {
              return; // drop the connection: no request in time
            }
            let _ = conn.await;
          }
        }
      });
    }
  });
  addr
}

// Send `bytes`, then wait up to 35s for the server to close the connection.
async fn probe(addr: SocketAddr, bytes: &'static [u8]) -> String {
  let start = Instant::now();
  let mut tcp = TcpStream::connect(addr).await.unwrap();
  tcp.write_all(bytes).await.unwrap();
  let mut got = 0;
  let mut buf = [0u8; 1024];
  loop {
    let left = Duration::from_secs(35).saturating_sub(start.elapsed());
    match tokio::time::timeout(left, tcp.read(&mut buf)).await {
      Ok(Ok(0)) => return format!("closed at {:.1}s ({got} bytes)", start.elapsed().as_secs_f64()),
      Ok(Ok(n)) => got += n,
      Ok(Err(e)) => return format!("{} at {:.1}s", e.kind(), start.elapsed().as_secs_f64()),
      Err(_) => return format!("open at 35s ({got} bytes)"),
    }
  }
}

const CASES: [(&str, &[u8]); 4] = [
  ("nothing", b""),
  ("half head", b"GET / HTTP/1.1\r\nHost: x\r\n"),
  ("half preface", b"PRI * HTTP/2.0\r\n"),
  ("1 request, idle", b"GET / HTTP/1.1\r\nHost: x\r\n\r\n"),
];

#[tokio::main]
async fn main() {
  let servers = [
    ("axum::serve", axum_server().await),
    ("auto + timer", own_server(None).await),
    ("auto + timer + 10s first-request deadline", own_server(Some(Duration::from_secs(10))).await),
  ];
  let mut jobs = Vec::new();
  for (name, addr) in servers {
    for (case, bytes) in CASES {
      jobs.push((name, case, tokio::spawn(probe(addr, bytes))));
    }
  }
  let mut last = "";
  for (name, case, job) in jobs {
    if name != last {
      println!("{name}:");
      last = name;
    }
    println!("  {case:<16} {}", job.await.unwrap());
  }
}

axum::serve:
  nothing          open at 35s (0 bytes)
  half head        open at 35s (0 bytes)
  half preface     open at 35s (0 bytes)
  1 request, idle  open at 35s (118 bytes)
auto + timer:
  nothing          open at 35s (0 bytes)
  half head        closed at 30.0s (0 bytes)
  half preface     open at 35s (0 bytes)
  1 request, idle  closed at 30.0s (77 bytes)
auto + timer + 10s first-request deadline:
  nothing          closed at 10.0s (0 bytes)
  half head        closed at 10.0s (0 bytes)
  half preface     closed at 10.0s (0 bytes)
  1 request, idle  closed at 30.0s (77 bytes)

    tokio::spawn(async move {
        #[allow(unused_mut)]
        let mut builder = Builder::new(TokioExecutor::new());
        // CONNECT protocol needed for HTTP/2 websockets
        #[cfg(feature = "http2")]
        builder.http2().enable_connect_protocol();

        let mut conn = pin!(builder.serve_connection_with_upgrades(io, hyper_service));