rust-web-server 18.8.2

//! HTTP/1.1 over tokio — the one connection handler for plain TCP and TLS in
//! the `http2` / `http3` builds (`Server::process` remains the synchronous
//! handler of the `http1`-only build).
//!
//! Each connection is a cheap task rather than a pool thread, so idle or slow
//! clients can't exhaust the server: 200 idle sockets used to occupy all 200
//! worker threads. Every read is bounded by [`READ_TIMEOUT`], TLS handshakes
//! by [`TLS_HANDSHAKE_TIMEOUT`], and the (synchronous) `Application` runs on
//! tokio's blocking pool so a slow handler or upstream never stalls the async
//! worker threads.

use std::io::Write;
use std::net::SocketAddr;
use std::time::Duration;

use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

use crate::application::Application;
use crate::entry_point::{get_ip_port_thread_count, get_request_allocation_size};
use crate::header::Header;

use crate::log::Log;
use crate::request::Request;
use crate::response::Response;
use crate::core::New;
use crate::server::sendfile::{self, SendfileSocket};
use crate::server::{framing, Address, ConnectionInfo, Server, REQUEST_HEAD_TIMEOUT};

/// Longest wait for a client to send a request head (or the next request on a
/// keep-alive connection) and for each read of a request body.
pub(crate) const READ_TIMEOUT: Duration = Duration::from_secs(30);

/// Bodies up to this size are copied behind the head and written in one call.
const COALESCE_LIMIT: usize = 64 * 1024;
/// Bodies at least this big go out with `writev` beside the head instead of
/// being copied behind it (where the stream supports vectored writes): a
/// 60 KiB static file took ~7% less CPU; at 1–5 KiB `writev` was no better
/// than the copy.
const VECTORED_MIN: usize = 16 * 1024;

/// Longest wait for a TLS handshake to complete after the TCP accept.
pub(crate) const TLS_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);

/// Serve HTTP/1.1 requests on `stream` until the client closes, asks to close,
/// idles past [`READ_TIMEOUT`], or sends something unparseable.
/// `is_tls` adds the `Strict-Transport-Security` and `Alt-Svc` headers.
pub(crate) async fn serve<S, A>(
    mut stream: S,
    peer_addr: SocketAddr,
    sni_hostname: Option<String>,
    is_tls: bool,
    app: A,
) -> Result<(), String>
where
    S: AsyncRead + AsyncWrite + Unpin + Send + SendfileSocket + 'static,
    A: Application + Send + Clone + 'static,
{
    let (server_ip, server_port, _) = get_ip_port_thread_count();
    let request_allocation_size = get_request_allocation_size();
    let max_body_size = crate::entry_point::get_max_body_size();

    // Bytes read but not consumed yet: the rest of a request head, a body, or
    // the next pipelined request. See `super::framing`.
    let mut pending: Vec<u8> = Vec::new();
    let mut read_buffer = vec![0u8; request_allocation_size as usize];
    // One idle timer for the whole connection, pushed back before each read.
    // Moving a deadline later is a lock-free compare-and-swap in tokio;
    // creating a timer per read (`tokio::time::timeout`) took the timer
    // driver's lock, shared by all worker threads, twice per read.
    let idle = tokio::time::sleep(READ_TIMEOUT);
    tokio::pin!(idle);

    // The same for every request on this connection: built once.
    let connection = ConnectionInfo {
        client: Address { ip: peer_addr.ip().to_string(), port: peer_addr.port() as i32 },
        server: Address { ip: server_ip, port: server_port },
        request_size: request_allocation_size,
        sni_hostname,
    };

    'session: loop {
        // Read until the request head is complete — it may span several reads,
        // but must finish within REQUEST_HEAD_TIMEOUT of its first byte.
        let mut head_deadline: Option<tokio::time::Instant> = None;
        let head_len = loop {
            framing::skip_leading_empty_lines(&mut pending);
            match framing::scan_head(&pending) {
                framing::HeadScan::Complete(n) => break Some(n),
                framing::HeadScan::TooLarge => {
                    write_and_flush(&mut stream, &Server::header_fields_too_large_response()).await;
                    return Err("request head too large".to_string());
                }
                framing::HeadScan::Incomplete => {}
            }
            if !pending.is_empty() && head_deadline.is_none() {
                head_deadline = Some(tokio::time::Instant::now() + REQUEST_HEAD_TIMEOUT);
            }
            if !read_more(&mut stream, &mut read_buffer, &mut pending, idle.as_mut(), head_deadline).await {
                // Closed, read error, or a timeout: end of the session. A head
                // that ran out its deadline gets a 408 first.
                if head_deadline.is_some_and(|d| tokio::time::Instant::now() >= d) {
                    write_and_flush(&mut stream, &Server::request_timeout_response()).await;
                }
                break None;
            }
            if head_deadline.is_none() {
                head_deadline = Some(tokio::time::Instant::now() + REQUEST_HEAD_TIMEOUT);
            }
        };
        let Some(head_len) = head_len else { break };

        let parsed = Request::parse(&pending[..head_len])
            .and_then(|request| framing::validate_head(&pending[..head_len], &request).map(|_| request))
            .and_then(|request| framing::body_framing(&request).map(|f| (request, f)));
        pending.drain(..head_len);
        let (mut request, body_framing) = match parsed {
            Ok(parsed) => parsed,
            Err(message) => {
                write_and_flush(&mut stream, &Server::bad_request_response(message.clone())).await;
                return Err(message);
            }
        };

        // RFC 9110 §10.1.1: only `100-continue` is understood.
        let expects_continue = match request.header(Header::_EXPECT) {
            Some(h) if h.value.trim().eq_ignore_ascii_case("100-continue") => true,
            Some(h) => {
                let raw = Server::expectation_failed_response(format!(
                    "417 Expectation Failed: unsupported Expect value '{}'", h.value.trim()
                ));
                write_and_flush(&mut stream, &raw).await;
                break;
            }
            None => false,
        };

        // A body the application wants as a stream (#31): run the handler on
        // the blocking pool now and feed it the body as it arrives.
        if !matches!(body_framing, framing::BodyFraming::None) && app.streams_body(&request, &connection) {
            if let framing::BodyFraming::Length(declared_len) = body_framing {
                if max_body_size > 0 && declared_len > max_body_size {
                    let raw = Server::payload_too_large_response(format!(
                        "413 Payload Too Large: declared Content-Length {} exceeds the {} byte limit",
                        declared_len, max_body_size
                    ));
                    write_and_flush(&mut stream, &raw).await;
                    break;
                }
            }
            if expects_continue && stream.write_all(&Server::continue_response()).await.is_err() {
                break;
            }
            let (tx, body) = crate::server::body_pump::body_channel();
            request.body = body;
            let keep_alive = framing::keep_alive(&request);
            let handler_app = app.clone();
            let handler_connection = connection.clone();
            let started = std::time::Instant::now();
            let mut handler = tokio::task::spawn_blocking(move || {
                let result = handler_app.execute(&request, &handler_connection);
                (request, result)
            });
            let (joined, outcome) = {
                let pump = pump_body(&mut stream, &mut read_buffer, &mut pending, idle.as_mut(), body_framing, max_body_size, tx);
                tokio::pin!(pump);
                let mut outcome = None;
                let joined = loop {
                    tokio::select! {
                        done = &mut pump, if outcome.is_none() => outcome = Some(done),
                        joined = &mut handler => break joined,
                    }
                };
                (joined, outcome)
            };
            let (request, result) = joined.map_err(|e| format!("request handler panicked: {}", e))?;
            // Reuse the connection only when the whole body was read off it.
            let reusable = keep_alive && outcome == Some(crate::server::body_pump::PumpOutcome::Complete);
            let mut response = match result {
                Ok(r) => r,
                Err(message) => {
                    crate::metrics::http::observe(&request.method, 400, is_tls, &request.http_version, started);
                    write_and_flush(&mut stream, &Server::bad_request_response(message.clone())).await;
                    return Err(message);
                }
            };
            crate::metrics::record_request();
            crate::metrics::http::observe(&request.method, response.status_code, is_tls, &request.http_version, started);
            crate::compression::apply_compression(&request, &mut response);
            response.drop_content_for_status();
            if is_tls {
                response.headers.push(Header::get_hsts_header());
                #[cfg(feature = "http3")]
                let alt_svc = format!("h3=\":{}\"", server_port);
                #[cfg(not(feature = "http3"))]
                let alt_svc = format!("h2=\":{}\"", server_port);
                response.headers.push(Header { name: Header::_ALT_SVC.into(), value: alt_svc.into() });
            }
            response.headers.push(Header {
                name: Header::_CONNECTION.into(),
                value: if reusable { "keep-alive" } else { "close" }.into(),
            });
            Log::log_access(&request, &response, &peer_addr);
            if response.stream_pipe.is_some() || response.stream_file.is_some() {
                write_streaming(&mut stream, response, request).await?;
            } else {
                let (head, body) = Response::generate_response_parts(response, request);
                stream.write_all(&head).await.map_err(|e| e.to_string())?;
                stream.write_all(&body).await.map_err(|e| e.to_string())?;
                stream.flush().await.map_err(|e| e.to_string())?;
            }
            if !reusable {
                break;
            }
            continue;
        }

        match body_framing {
            framing::BodyFraming::None => {}
            framing::BodyFraming::Length(declared_len) => {
                if max_body_size > 0 && declared_len > max_body_size {
                    let raw = Server::payload_too_large_response(format!(
                        "413 Payload Too Large: declared Content-Length {} exceeds the {} byte limit",
                        declared_len, max_body_size
                    ));
                    write_and_flush(&mut stream, &raw).await;
                    break;
                }
                if expects_continue
                    && (pending.len() as u64) < declared_len
                    && stream.write_all(&Server::continue_response()).await.is_err()
                {
                    break;
                }
                while (pending.len() as u64) < declared_len {
                    if !read_more(&mut stream, &mut read_buffer, &mut pending, idle.as_mut(), None).await {
                        break 'session; // closed or stalled mid-body
                    }
                }
                request.body = pending.drain(..declared_len as usize).collect::<Vec<u8>>().into();
            }
            framing::BodyFraming::Chunked => {
                if expects_continue
                    && pending.is_empty()
                    && stream.write_all(&Server::continue_response()).await.is_err()
                {
                    break;
                }
                let mut decoder = framing::ChunkedBody::new(max_body_size);
                loop {
                    match decoder.feed(&pending) {
                        Ok((consumed, done)) => {
                            pending.drain(..consumed);
                            if done {
                                break;
                            }
                        }
                        Err(framing::ChunkError::TooLarge) => {
                            let raw = Server::payload_too_large_response(format!(
                                "413 Payload Too Large: chunked body exceeds the {} byte limit", max_body_size
                            ));
                            write_and_flush(&mut stream, &raw).await;
                            break 'session;
                        }
                        Err(framing::ChunkError::Malformed(message)) => {
                            write_and_flush(&mut stream, &Server::bad_request_response(message.clone())).await;
                            return Err(message);
                        }
                    }
                    if !read_more(&mut stream, &mut read_buffer, &mut pending, idle.as_mut(), None).await {
                        break 'session;
                    }
                }
                request.body = decoder.body.into();
            }
        }

        let keep_alive = framing::keep_alive(&request);

        let started = std::time::Instant::now();
        let (request, result) = execute_request(app.clone(), request, &connection)
            .await
            .map_err(|e| format!("request handler panicked: {}", e))?;

        let mut response = match result {
            Ok(r) => r,
            Err(message) => {
                crate::metrics::http::observe(&request.method, 400, is_tls, &request.http_version, started);
                write_and_flush(&mut stream, &Server::bad_request_response(message.clone())).await;
                return Err(message);
            }
        };

        crate::metrics::record_request();
        crate::metrics::http::observe(&request.method, response.status_code, is_tls, &request.http_version, started);
        // A WebSocket upgrade: send the 101, then the connection is the
        // handler's (`WebSocket::upgrade`).
        if let Some(upgrade) = crate::websocket::take_upgrade(&mut response) {
            Log::log_access(&request, &response, &peer_addr);
            stream.write_all(&crate::websocket::upgrade_head(&response)).await.map_err(|e| e.to_string())?;
            stream.flush().await.map_err(|e| e.to_string())?;
            crate::websocket::hand_over_async(stream, std::mem::take(&mut pending), upgrade);
            return Ok(());
        }
        crate::compression::apply_compression(&request, &mut response);
        response.drop_content_for_status();
        if is_tls {
            response.headers.push(Header::get_hsts_header());
            #[cfg(feature = "http3")]
            let alt_svc = format!("h3=\":{}\"", server_port);
            #[cfg(not(feature = "http3"))]
            let alt_svc = format!("h2=\":{}\"", server_port);
            response.headers.push(Header { name: Header::_ALT_SVC.into(), value: alt_svc.into() });
        }
        response.headers.push(Header {
            name: Header::_CONNECTION.into(),
            value: if keep_alive { "keep-alive" } else { "close" }.into(),
        });

        Log::log_access(&request, &response, &peer_addr);

        if response.stream_pipe.is_some() || response.stream_file.is_some() {
            if !write_file_zero_copy(&mut stream, &mut response, &request).await?
                && !write_file_inline(&mut stream, &mut response, &request).await?
            {
                write_streaming(&mut stream, response, request).await?;
            }
        } else {
            // One write — one syscall, one TCP segment — for a small response:
            // with TCP_NODELAY on, writing head and body separately sent two
            // segments and doubled the client's receive work. A stream with
            // vectored writes (plain TCP) gets head and a body of at least
            // VECTORED_MIN in one writev, so the body isn't copied behind the
            // head; a smaller body (where writev costs more than the copy),
            // or any body on a stream without vectored writes (TLS, which
            // copies into records anyway), is appended to a head buffer sized
            // for it.
            let vectored = stream.is_write_vectored();
            let limit = if vectored { VECTORED_MIN - 1 } else { COALESCE_LIMIT };
            let (head, body) = Response::generate_response_coalesced(response, request, limit);
            if vectored && !body.is_empty() {
                write_all_vectored(&mut stream, &head, &body).await.map_err(|e| e.to_string())?;
            } else {
                stream.write_all(&head).await.map_err(|e| e.to_string())?;
                if !body.is_empty() {
                    stream.write_all(&body).await.map_err(|e| e.to_string())?;
                }
            }
            stream.flush().await.map_err(|e| e.to_string())?;
        }

        if !keep_alive {
            break;
        }
    }
    let _ = stream.shutdown().await;
    Ok(())
}

/// Run the `Application` for one request: directly on this task when the app
/// says the request is quick and non-blocking (`Application::runs_inline` —
/// health checks, small static files), otherwise on the blocking pool via
/// [`execute_blocking`]. The blocking-pool handoff (a shared lock and two
/// cross-thread wake-ups) cost more than serving a small file. A panic is
/// caught either way and costs only this request.
///
/// `connection` is borrowed: it's built once per connection, and only the
/// blocking-pool path needs an owned copy.
pub(crate) async fn execute_request<A: Application + Send + 'static>(
    app: A,
    request: Request,
    connection: &ConnectionInfo,
) -> Result<(Request, Result<Response, String>), String> {
    let async_result = match app.execute_async(&request, connection) {
        Some(future) => Some(CatchUnwind(future).await?),
        None => None,
    };
    if let Some(result) = async_result {
        return Ok((request, result));
    }
    if app.runs_inline(&request, connection) {
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| app.execute(&request, connection)))
            .map_err(panic_message)?;
        return Ok((request, result));
    }
    execute_blocking(app, request, connection.clone()).await
}

/// Polls a boxed future, turning a panic into an error (one request lost,
/// not the connection task).
struct CatchUnwind<F>(F);

impl<F: std::future::Future + Unpin> std::future::Future for CatchUnwind<F> {
    type Output = Result<F::Output, String>;

    fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Output> {
        let inner = &mut self.0;
        match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| std::pin::Pin::new(inner).poll(cx))) {
            Ok(std::task::Poll::Ready(output)) => std::task::Poll::Ready(Ok(output)),
            Ok(std::task::Poll::Pending) => std::task::Poll::Pending,
            Err(panic) => std::task::Poll::Ready(Err(panic_message(panic))),
        }
    }
}

fn panic_message(panic: Box<dyn std::any::Any + Send>) -> String {
    panic
        .downcast_ref::<&str>()
        .map(|s| s.to_string())
        .or_else(|| panic.downcast_ref::<String>().cloned())
        .unwrap_or_else(|| "panic".to_string())
}

/// Run the (synchronous) `Application` off the async scheduler — file reads
/// and upstream proxying must never hold an async worker thread hostage,
/// where a handful of slow requests would stall every connection. Used by the
/// HTTP/1.1, HTTP/2 and HTTP/3 handlers alike. A handler panic is returned
/// as an error, costing that one request.
pub(crate) async fn execute_blocking<A: Application + Send + 'static>(
    app: A,
    request: Request,
    connection: ConnectionInfo,
) -> Result<(Request, Result<Response, String>), String> {
    tokio::task::spawn_blocking(move || {
        let result = app.execute(&request, &connection);
        (request, result)
    })
    .await
    .map_err(|e| e.to_string())
}

/// Feed a streamed request body (#31) from the connection into `tx`: what
/// is already in `pending`, then reads, decoding chunked framing. Every read
/// is bounded by the idle timer as usual. A client that stops, sends too
/// much or breaks the framing ends the stream with an error the handler
/// sees; a handler that drops the body ends it as abandoned.
async fn pump_body<S: AsyncRead + Unpin>(
    stream: &mut S,
    read_buffer: &mut [u8],
    pending: &mut Vec<u8>,
    mut idle: std::pin::Pin<&mut tokio::time::Sleep>,
    body_framing: framing::BodyFraming,
    max_body_size: u64,
    tx: crate::server::body_pump::BodySender,
) -> crate::server::body_pump::PumpOutcome {
    use crate::server::body_pump::PumpOutcome;
    use std::io::{Error, ErrorKind};
    let stopped = || Error::new(ErrorKind::UnexpectedEof, "the client stopped sending the request body");
    match body_framing {
        framing::BodyFraming::None => PumpOutcome::Complete,
        framing::BodyFraming::Length(declared_len) => {
            let mut remaining = declared_len;
            loop {
                if remaining > 0 && !pending.is_empty() {
                    let take = (pending.len() as u64).min(remaining) as usize;
                    let chunk: Vec<u8> = pending.drain(..take).collect();
                    remaining -= take as u64;
                    if tx.send(Ok(chunk)).await.is_err() {
                        return PumpOutcome::Abandoned;
                    }
                }
                if remaining == 0 {
                    return PumpOutcome::Complete;
                }
                if !read_more(stream, read_buffer, pending, idle.as_mut(), None).await {
                    let _ = tx.send(Err(stopped())).await;
                    return PumpOutcome::Failed;
                }
            }
        }
        framing::BodyFraming::Chunked => {
            let mut decoder = framing::ChunkedBody::new(max_body_size);
            loop {
                match decoder.feed(pending) {
                    Ok((consumed, done)) => {
                        pending.drain(..consumed);
                        let chunk = decoder.take_decoded();
                        if !chunk.is_empty() && tx.send(Ok(chunk)).await.is_err() {
                            return PumpOutcome::Abandoned;
                        }
                        if done {
                            return PumpOutcome::Complete;
                        }
                    }
                    Err(framing::ChunkError::TooLarge) => {
                        let message = format!("the request body exceeds the {} byte limit", max_body_size);
                        let _ = tx.send(Err(Error::new(ErrorKind::InvalidData, message))).await;
                        return PumpOutcome::Failed;
                    }
                    Err(framing::ChunkError::Malformed(message)) => {
                        let _ = tx.send(Err(Error::new(ErrorKind::InvalidData, message))).await;
                        return PumpOutcome::Failed;
                    }
                }
                if !read_more(stream, read_buffer, pending, idle.as_mut(), None).await {
                    let _ = tx.send(Err(stopped())).await;
                    return PumpOutcome::Failed;
                }
            }
        }
    }
}

/// One read (bounded by [`READ_TIMEOUT`], via the connection's `idle` timer,
/// and by `deadline` when given) appended to `pending`. `false` when the
/// peer closed, the read failed, or it timed out.
async fn read_more<S: AsyncRead + Unpin>(
    stream: &mut S,
    read_buffer: &mut [u8],
    pending: &mut Vec<u8>,
    mut idle: std::pin::Pin<&mut tokio::time::Sleep>,
    deadline: Option<tokio::time::Instant>,
) -> bool {
    let idle_until = tokio::time::Instant::now() + READ_TIMEOUT;
    idle.as_mut().reset(deadline.map_or(idle_until, |d| d.min(idle_until)));
    tokio::select! {
        read = stream.read(read_buffer) => match read {
            Ok(n) if n > 0 => {
                pending.extend_from_slice(&read_buffer[..n]);
                true
            }
            _ => false,
        },
        _ = idle => false,
    }
}

async fn write_and_flush<S: AsyncWrite + Unpin>(stream: &mut S, bytes: &[u8]) {
    if stream.write_all(bytes).await.is_ok() {
        let _ = stream.flush().await;
    }
}

/// Write a `stream_file` / `stream_pipe` response by running the existing
/// synchronous chunked writers (`Server::write_chunked_file`,
/// `Server::pipe_stream`) on the blocking pool, feeding their output to the
/// connection through a bounded channel — so a large file or a long SSE
/// stream never holds an async worker thread.
async fn write_streaming<S: AsyncWrite + Unpin>(
    stream: &mut S,
    mut response: Response,
    request: Request,
) -> Result<(), String> {
    let (tx, mut rx) = tokio::sync::mpsc::channel::<Vec<u8>>(8);
    // Written chunks go back to the producer for reuse: a fresh 64 KiB Vec
    // per chunk, freed on another thread, kept the allocator handing out new
    // pages — page faults and zeroing were an eighth of a TLS download's CPU.
    let (recycle_tx, recycle_rx) = std::sync::mpsc::channel::<Vec<u8>>();
    let writer = tokio::task::spawn_blocking(move || {
        let mut sink = ChannelWriter { chunks: tx, recycled: recycle_rx };
        if let Some(reader) = response.stream_pipe.take() {
            Server::pipe_stream(&mut sink, response, request, reader)
        } else {
            let filepath = response.stream_file.clone().unwrap_or_default();
            Server::write_chunked_file(&mut sink, response, request, &filepath)
        }
    });

    let mut client_error = None;
    while let Some(chunk) = rx.recv().await {
        let written = async {
            stream.write_all(&chunk).await?;
            stream.flush().await
        }
        .await;
        if let Err(e) = written {
            // Dropping the receiver makes the producer's next write fail.
            client_error = Some(e.to_string());
            break;
        }
        let _ = recycle_tx.send(chunk);
    }
    drop(rx);

    let produced = writer.await.map_err(|e| format!("streaming writer panicked: {}", e))?;
    match client_error {
        Some(e) => Err(e),
        None => produced,
    }
}

/// A `stream_file` GET with a known `Content-Length` over plain TCP: write
/// the head, then let the kernel copy the file to the socket (`sendfile`).
/// `Ok(false)` — nothing written — when that doesn't apply (TLS, HEAD, a
/// chunked or piped body, an unsupported OS, a file that won't open); the
/// caller then uses [`write_streaming`].
async fn write_file_zero_copy<S: AsyncWrite + Unpin + SendfileSocket>(
    stream: &mut S,
    response: &mut Response,
    request: &Request,
) -> Result<bool, String> {
    if !sendfile::SUPPORTED || response.stream_pipe.is_some() || request.method != crate::request::METHOD.get {
        return Ok(false);
    }
    let Some(path) = response.stream_file.clone() else { return Ok(false) };
    let Some(len) = response
        .headers
        .iter()
        .find(|h| h.name.eq_ignore_ascii_case(Header::_CONTENT_LENGTH))
        .and_then(|h| h.value.trim().parse::<u64>().ok())
    else {
        return Ok(false);
    };
    if stream.sendfile_socket().is_none() {
        return Ok(false);
    }
    // The static controller's open-file cache usually already holds this
    // file open: reuse that handle instead of opening it again by path.
    let Some(file) = crate::app::controller::static_resource::open_file_cache::open(&path, true) else {
        return Ok(false);
    };

    let (head, _) = Server::stream_file_head(std::mem::replace(response, Response::new()));
    let socket = stream.sendfile_socket().expect("checked above");
    sendfile::send_file(socket, &head, file.file(), len).await.map_err(|e| e.to_string())?;
    Ok(true)
}

/// Chunk size of [`write_file_inline`].
const INLINE_FILE_CHUNK: usize = 64 * 1024;

/// A `stream_file` GET with a known `Content-Length` where `sendfile`
/// doesn't apply (TLS): the same head, then the file read on this task, 64
/// KiB at a time into one buffer, from the open-file cache's handle — the way
/// nginx serves files without aio. Going through the blocking pool instead
/// cost a cross-thread hand-off per chunk, and with mimalloc each chunk
/// freed on another thread than it was allocated on: 1 MiB over TLS took
/// 20–50% more CPU than nginx's. A read can block this worker on a cold
/// cache, as it does nginx's. `Ok(false)` when it doesn't apply.
#[cfg(unix)]
async fn write_file_inline<S: AsyncWrite + Unpin>(
    stream: &mut S,
    response: &mut Response,
    request: &Request,
) -> Result<bool, String> {
    use std::os::unix::fs::FileExt;
    if response.stream_pipe.is_some() || request.method != crate::request::METHOD.get {
        return Ok(false);
    }
    let Some(path) = response.stream_file.clone() else { return Ok(false) };
    let Some(len) = response
        .headers
        .iter()
        .find(|h| h.name.eq_ignore_ascii_case(Header::_CONTENT_LENGTH))
        .and_then(|h| h.value.trim().parse::<u64>().ok())
    else {
        return Ok(false);
    };
    let Some(file) = crate::app::controller::static_resource::open_file_cache::open(&path, true) else {
        return Ok(false);
    };

    let (head, _) = Server::stream_file_head(std::mem::replace(response, Response::new()));
    stream.write_all(&head).await.map_err(|e| e.to_string())?;
    let chunk = INLINE_FILE_CHUNK.min(len as usize);
    let mut backing = vec![0u8; chunk + 4096];
    let start = mid_page_offset(backing.as_ptr());
    let buffer = &mut backing[start..start + chunk];
    let mut offset = 0u64;
    while offset < len {
        let want = buffer.len().min((len - offset) as usize);
        let read = file.file().read_at(&mut buffer[..want], offset).map_err(|e| e.to_string())?;
        if read == 0 {
            // The file shrank after the head promised `len` bytes.
            return Err(format!("{}: ended at {} of {} bytes", path, offset, len));
        }
        stream.write_all(&buffer[..read]).await.map_err(|e| e.to_string())?;
        offset += read as u64;
    }
    stream.flush().await.map_err(|e| e.to_string())?;
    Ok(true)
}

/// Where in an allocation at `base` a buffer handed to rustls should
/// start: half a page from a page boundary. rustls copies each 16 KiB record
/// of plaintext into a fresh buffer 5 bytes past its start; with our buffer
/// page-aligned too, source and destination are 5 bytes apart modulo 4 KiB,
/// which AMD Zen 3's `rep movsb` (glibc's memcpy at that size) runs at
/// 2.8 instead of 40 GB/s — two thirds of a 1 MiB TLS download's CPU on the
/// EPYC 7763 benchmark runners. Half a page apart is fast on every CPU.
fn mid_page_offset(base: *const u8) -> usize {
    (2048 + 4096 - (base as usize % 4096)) % 4096
}

#[cfg(not(unix))]
async fn write_file_inline<S: AsyncWrite + Unpin>(
    _stream: &mut S,
    _response: &mut Response,
    _request: &Request,
) -> Result<bool, String> {
    Ok(false)
}

/// `std::io::Write` that hands each buffer to the async writer task.
/// `head` then `body` with as few `writev`s as the socket takes.
async fn write_all_vectored<S: AsyncWrite + Unpin>(stream: &mut S, head: &[u8], body: &[u8]) -> std::io::Result<()> {
    use std::io::IoSlice;
    let total = head.len() + body.len();
    let mut written = 0;
    while written < total {
        let n = if written < head.len() {
            stream.write_vectored(&[IoSlice::new(&head[written..]), IoSlice::new(body)]).await?
        } else {
            stream.write(&body[written - head.len()..]).await?
        };
        if n == 0 {
            return Err(std::io::ErrorKind::WriteZero.into());
        }
        written += n;
    }
    Ok(())
}

struct ChannelWriter {
    chunks: tokio::sync::mpsc::Sender<Vec<u8>>,
    /// Buffers the connection task has written, for reuse.
    recycled: std::sync::mpsc::Receiver<Vec<u8>>,
}

impl Write for ChannelWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let mut chunk = self.recycled.try_recv().unwrap_or_default();
        chunk.clear();
        chunk.extend_from_slice(buf);
        self.chunks
            .blocking_send(chunk)
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "client connection closed"))?;
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::response::STATUS_CODE_REASON_PHRASE;

    #[tokio::test]
    async fn write_all_vectored_survives_short_writes() {
        // A 16-byte pipe: every write is short.
        let (mut client, mut server) = tokio::io::duplex(16);
        let head = b"HTTP/1.1 200 OK\r\nX: y\r\n\r\n".to_vec();
        let body: Vec<u8> = (0..1000u32).map(|i| i as u8).collect();
        let reader = tokio::spawn(async move {
            let mut all = Vec::new();
            client.read_to_end(&mut all).await.unwrap();
            all
        });
        write_all_vectored(&mut server, &head, &body).await.unwrap();
        drop(server);
        assert_eq!([head, body].concat(), reader.await.unwrap());
    }

    #[derive(Clone)]
    struct Big;

    impl Application for Big {
        fn execute(&self, _request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
            Ok(Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, &"z".repeat(VECTORED_MIN + 3000)))
        }
    }

    // Plain TCP has vectored writes: a body over VECTORED_MIN goes out with
    // writev beside the head — the client must still see one exact response.
    #[tokio::test]
    async fn large_body_over_tcp_is_sent_whole_with_writev() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            let (socket, peer) = listener.accept().await.unwrap();
            assert!(socket.is_write_vectored());
            serve(socket, peer, None, false, Big).await
        });
        let mut client = tokio::net::TcpStream::connect(address).await.unwrap();
        client.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
        let mut raw = Vec::new();
        client.read_to_end(&mut raw).await.unwrap();
        let text = String::from_utf8(raw).unwrap();
        let (head, body) = text.split_once("\r\n\r\n").unwrap();
        assert!(head.starts_with("HTTP/1.1 200 OK"), "{head}");
        assert!(head.contains(&format!("Content-Length: {}", VECTORED_MIN + 3000)), "{head}");
        assert_eq!("z".repeat(VECTORED_MIN + 3000), body);
        assert!(server.await.unwrap().is_ok());
    }

    #[test]
    fn channel_writer_reuses_returned_buffers() {
        let (tx, mut rx) = tokio::sync::mpsc::channel::<Vec<u8>>(8);
        let (recycle_tx, recycle_rx) = std::sync::mpsc::channel::<Vec<u8>>();
        let mut writer = ChannelWriter { chunks: tx, recycled: recycle_rx };
        writer.write_all(&[1u8; 1000]).unwrap();
        let first = rx.blocking_recv().unwrap();
        assert_eq!(vec![1u8; 1000], first);
        let address = first.as_ptr();
        recycle_tx.send(first).unwrap();
        writer.write_all(&[2u8; 10]).unwrap();
        let second = rx.blocking_recv().unwrap();
        assert_eq!(vec![2u8; 10], second, "contents are replaced, not appended");
        assert_eq!(address, second.as_ptr(), "the returned buffer is reused");
        drop(rx);
        assert!(writer.write_all(b"x").is_err(), "a closed connection fails the producer");
    }

    #[derive(Clone)]
    struct Hello;

    impl Application for Hello {
        fn execute(&self, request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
            if request.request_uri == "/panic" {
                panic!("handler bug");
            }
            Ok(Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "hello"))
        }
    }

    fn peer() -> SocketAddr {
        "127.0.0.1:5555".parse().unwrap()
    }

    async fn read_until_count(client: &mut tokio::io::DuplexStream, needle: &str, count: usize) -> String {
        let mut out = Vec::new();
        let mut buf = [0u8; 4096];
        while String::from_utf8_lossy(&out).matches(needle).count() < count {
            let n = client.read(&mut buf).await.unwrap();
            if n == 0 {
                break;
            }
            out.extend_from_slice(&buf[..n]);
        }
        String::from_utf8_lossy(&out).to_string()
    }

    #[tokio::test]
    async fn keep_alive_serves_several_requests_on_one_connection() {
        let (mut client, server) = tokio::io::duplex(64 * 1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        client.write_all(b"GET /a HTTP/1.1\r\nHost: x\r\n\r\n").await.unwrap();
        let first = read_until_count(&mut client, "hello", 1).await;
        client.write_all(b"GET /b HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
        let second = read_until_count(&mut client, "hello", 1).await;
        assert!(first.starts_with("HTTP/1.1 200") && first.contains("Connection: keep-alive"), "{first}");
        assert!(second.contains("Connection: close"), "{second}");
        assert!(task.await.unwrap().is_ok());
    }

    #[tokio::test]
    async fn connection_close_among_other_tokens_ends_the_session() {
        let (mut client, server) = tokio::io::duplex(64 * 1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        client.write_all(b"GET /a HTTP/1.1\r\nHost: x\r\nConnection: close, TE\r\nTE: trailers\r\n\r\n").await.unwrap();
        let mut raw = Vec::new();
        client.read_to_end(&mut raw).await.unwrap();
        let text = String::from_utf8_lossy(&raw);
        assert!(text.contains("Connection: close"), "{text}");
        assert!(task.await.unwrap().is_ok());
    }

    #[tokio::test]
    async fn a_request_without_host_is_400() {
        let (mut client, server) = tokio::io::duplex(64 * 1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        client.write_all(b"GET /a HTTP/1.1\r\nAccept: */*\r\n\r\n").await.unwrap();
        let mut raw = Vec::new();
        client.read_to_end(&mut raw).await.unwrap();
        assert!(String::from_utf8_lossy(&raw).starts_with("HTTP/1.1 400"), "{}", String::from_utf8_lossy(&raw));
        assert!(task.await.unwrap().is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn idle_connection_is_closed_after_the_read_timeout() {
        let (_client, server) = tokio::io::duplex(1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        // Paused clock: the timeout fires as soon as the runtime is idle.
        let result = tokio::time::timeout(READ_TIMEOUT + Duration::from_secs(1), task).await;
        assert!(result.expect("serve should return after READ_TIMEOUT").unwrap().is_ok());
    }

    #[tokio::test]
    async fn handler_panic_costs_the_request_not_the_runtime() {
        let (mut client, server) = tokio::io::duplex(1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        client.write_all(b"GET /panic HTTP/1.1\r\nHost: x\r\n\r\n").await.unwrap();
        let result = task.await.unwrap();
        assert!(result.unwrap_err().contains("panicked"));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn multi_thread_runtime_serves_and_survives_handler_panics() {
        // The rws binary runs a multi-threaded runtime.
        let (mut client, server) = tokio::io::duplex(64 * 1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        client.write_all(b"GET /a HTTP/1.1\r\nHost: x\r\n\r\n").await.unwrap();
        assert!(read_until_count(&mut client, "hello", 1).await.starts_with("HTTP/1.1 200"));
        client.write_all(b"GET /panic HTTP/1.1\r\nHost: x\r\n\r\n").await.unwrap();
        assert!(task.await.unwrap().unwrap_err().contains("panicked"));
    }

    #[tokio::test]
    async fn tls_responses_carry_hsts_and_alt_svc() {
        let (mut client, server) = tokio::io::duplex(64 * 1024);
        let task = tokio::spawn(serve(server, peer(), Some("x".to_string()), true, Hello));
        client.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
        let response = read_until_count(&mut client, "hello", 1).await;
        assert!(response.contains("Strict-Transport-Security"), "{response}");
        assert!(response.contains("Alt-Svc"), "{response}");
        task.await.unwrap().unwrap();
    }

    #[derive(Clone)]
    struct BigFile(String);

    impl Application for BigFile {
        fn execute(&self, _request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
            let mut response = Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "");
            response.content_range_list.clear();
            response.stream_file = Some(self.0.clone());
            Ok(response)
        }
    }

    #[tokio::test]
    async fn stream_file_is_sent_chunked_and_complete() {
        let file = crate::test_env::TempPath::new("http1_async_big.bin");
        let content: Vec<u8> = (0..300_000u32).map(|i| (i % 251) as u8).collect();
        std::fs::write(&*file, &content).unwrap();

        let (mut client, server) = tokio::io::duplex(16 * 1024);
        let task = tokio::spawn(serve(server, peer(), None, false, BigFile(file.as_str().to_string())));
        client.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
        let mut raw = Vec::new();
        client.read_to_end(&mut raw).await.unwrap();
        task.await.unwrap().unwrap();

        let head_end = raw.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
        assert!(String::from_utf8_lossy(&raw[..head_end]).contains("Transfer-Encoding: chunked"));
        let mut decoded = Vec::new();
        crate::server::body_stream::ChunkedDecoder::new(&raw[head_end..]).read_to_end_sync(&mut decoded);
        assert_eq!(content, decoded);
    }

    /// No `sendfile` on this stream (as over TLS): the file is read on the
    /// connection's task, sent whole, and the connection stays usable.
    #[tokio::test]
    async fn sized_file_without_sendfile_is_read_inline_and_keep_alive_continues() {
        let file = crate::test_env::TempPath::new("http1_async_sized.bin");
        let content: Vec<u8> = (0..300_000u32).map(|i| (i % 253) as u8).collect();
        std::fs::write(&*file, &content).unwrap();

        let (mut client, server) = tokio::io::duplex(16 * 1024);
        let app = SizedFile(file.as_str().to_string(), content.len());
        let task = tokio::spawn(serve(server, peer(), None, false, app));
        client.write_all(b"GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
        let mut raw = Vec::new();
        client.read_to_end(&mut raw).await.unwrap();
        task.await.unwrap().unwrap();

        let mut rest = &raw[..];
        for _ in 0..2 {
            let head_end = rest.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
            let head = String::from_utf8_lossy(&rest[..head_end]).to_string();
            assert!(head.contains("Content-Length: 300000"), "{head}");
            assert!(!head.contains("Transfer-Encoding"), "{head}");
            assert_eq!(&content[..], &rest[head_end..head_end + content.len()]);
            rest = &rest[head_end + content.len()..];
        }
        assert!(rest.is_empty(), "{} trailing bytes", rest.len());
    }

    #[test]
    fn inline_file_buffer_starts_mid_page() {
        for base in [0usize, 1, 5, 2048, 4095, 4096, 12_345] {
            let start = super::mid_page_offset(base as *const u8);
            assert!(start < 4096);
            assert_eq!(2048, (base + start) % 4096, "base {base}");
        }
    }

    /// A file shorter than the promised `Content-Length` ends the
    /// connection with an error rather than sending a short body as complete.
    #[tokio::test]
    async fn sized_file_that_shrank_ends_the_connection() {
        let file = crate::test_env::TempPath::new_in_cwd("http1_async_shrank.bin");
        std::fs::write(&*file, vec![7u8; 1000]).unwrap();

        let (mut client, server) = tokio::io::duplex(16 * 1024);
        let task = tokio::spawn(serve(server, peer(), None, false, SizedFile(file.as_str().to_string(), 5000)));
        client.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").await.unwrap();
        let mut raw = Vec::new();
        client.read_to_end(&mut raw).await.unwrap();
        let err = task.await.unwrap().unwrap_err();
        assert!(err.contains("ended at 1000 of 5000"), "{err}");
        let head_end = raw.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
        assert_eq!(1000, raw.len() - head_end);
    }

    trait ReadToEndSync {
        fn read_to_end_sync(self, out: &mut Vec<u8>);
    }
    impl<R: std::io::Read> ReadToEndSync for crate::server::body_stream::ChunkedDecoder<R> {
        fn read_to_end_sync(mut self, out: &mut Vec<u8>) {
            std::io::Read::read_to_end(&mut self, out).unwrap();
        }
    }

    #[derive(Clone)]
    struct Echo;

    impl Application for Echo {
        fn execute(&self, request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
            let text = format!("uri={} body={}", request.request_uri, String::from_utf8_lossy(&request.body));
            Ok(Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, &text))
        }
    }

    async fn read_to_end(client: &mut tokio::io::DuplexStream) -> String {
        let mut out = Vec::new();
        client.read_to_end(&mut out).await.unwrap();
        String::from_utf8_lossy(&out).to_string()
    }

    #[tokio::test]
    async fn large_request_head_spanning_reads_is_served() {
        let (mut client, server) = tokio::io::duplex(1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Echo));
        let mut raw = b"GET /big HTTP/1.1\r\nHost: x\r\nCookie: ".to_vec();
        raw.extend(std::iter::repeat(b'c').take(40_000));
        raw.extend_from_slice(b"\r\nConnection: close\r\n\r\n");
        let writer = tokio::spawn(async move {
            client.write_all(&raw).await.unwrap();
            client
        });
        let mut client = writer.await.unwrap();
        let out = read_to_end(&mut client).await;
        assert!(out.starts_with("HTTP/1.1 200") && out.contains("uri=/big"), "{out}");
        assert!(task.await.unwrap().is_ok());
    }

    #[tokio::test]
    async fn oversized_request_head_gets_431() {
        let (mut client, server) = tokio::io::duplex(256 * 1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Echo));
        let mut raw = b"GET / HTTP/1.1\r\nHost: x\r\nCookie: ".to_vec();
        raw.extend(std::iter::repeat(b'c').take(framing::MAX_HEAD_SIZE + 10));
        client.write_all(&raw).await.unwrap();
        let out = read_to_end(&mut client).await;
        assert!(out.starts_with("HTTP/1.1 431"), "{out}");
        assert!(task.await.unwrap().is_err());
    }

    #[tokio::test]
    async fn pipelined_and_chunked_requests_are_framed() {
        let (mut client, server) = tokio::io::duplex(64 * 1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Echo));
        client.write_all(
            b"POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 3\r\n\r\nabc\
              POST /b HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nhi\r\n0\r\n\r\n\
              GET /c HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",
        ).await.unwrap();
        let out = read_to_end(&mut client).await;
        assert!(out.contains("uri=/a body=abc"), "{out}");
        assert!(out.contains("uri=/b body=hi"), "{out}");
        assert!(out.contains("uri=/c body="), "{out}");
        assert!(task.await.unwrap().is_ok());
    }


    /// Answers inline (opted in) or not, and records the thread it ran on.
    #[derive(Clone)]
    struct ThreadProbe {
        inline: bool,
        ran_on: std::sync::Arc<std::sync::Mutex<Option<std::thread::ThreadId>>>,
    }

    impl Application for ThreadProbe {
        fn execute(&self, request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
            if request.request_uri == "/panic" {
                panic!("inline handler bug");
            }
            *self.ran_on.lock().unwrap() = Some(std::thread::current().id());
            Ok(Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "ok"))
        }

        fn runs_inline(&self, _request: &Request, _connection: &ConnectionInfo) -> bool {
            self.inline
        }
    }

    fn probe_connection() -> ConnectionInfo {
        ConnectionInfo {
            client: Address { ip: "127.0.0.1".to_string(), port: 1 },
            server: Address { ip: "127.0.0.1".to_string(), port: 2 },
            request_size: 1024,
            sni_hostname: None,
        }
    }

    fn probe_request(uri: &str) -> Request {
        Request { method: "GET".to_string(), request_uri: uri.to_string(), http_version: "HTTP/1.1".to_string(), headers: vec![], body: Default::default() }
    }

    #[tokio::test]
    async fn inline_requests_run_on_the_calling_thread_others_on_the_blocking_pool() {
        let here = std::thread::current().id();
        for (inline, expect_here) in [(true, true), (false, false)] {
            let probe = ThreadProbe { inline, ran_on: Default::default() };
            let (_, result) = execute_request(probe.clone(), probe_request("/"), &probe_connection()).await.unwrap();
            assert!(result.is_ok());
            let ran_on = probe.ran_on.lock().unwrap().unwrap();
            assert_eq!(ran_on == here, expect_here, "inline = {inline}");
        }
    }

    #[tokio::test]
    async fn inline_handler_panic_is_caught() {
        let probe = ThreadProbe { inline: true, ran_on: Default::default() };
        let err = execute_request(probe, probe_request("/panic"), &probe_connection()).await.unwrap_err();
        assert!(err.contains("inline handler bug"), "{err}");
    }


    /// A `stream_file` response with a known length — the static-file shape.
    #[derive(Clone)]
    struct SizedFile(String, usize);

    impl Application for SizedFile {
        fn execute(&self, _request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
            let mut response = Response::get_response(STATUS_CODE_REASON_PHRASE.n200_ok, None, None);
            response.headers.push(Header { name: Header::_CONTENT_LENGTH.into(), value: self.1.to_string().into() });
            response.stream_file = Some(self.0.clone());
            Ok(response)
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn plain_tcp_file_is_sent_with_sendfile_and_keep_alive_continues() {
        let file = crate::test_env::TempPath::new("http1_async_sendfile.bin");
        let content: Vec<u8> = (0..700_000u32).map(|i| (i % 249) as u8).collect();
        std::fs::write(&*file, &content).unwrap();

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let app = SizedFile(file.as_str().to_string(), content.len());
        let server = tokio::spawn(async move {
            let (tcp, peer_addr) = listener.accept().await.unwrap();
            serve(tcp, peer_addr, None, false, app).await
        });

        let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
        client.write_all(b"GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
        let mut raw = Vec::new();
        client.read_to_end(&mut raw).await.unwrap();
        assert!(server.await.unwrap().is_ok());

        // Two complete responses, each a head plus exactly Content-Length bytes.
        let mut rest = &raw[..];
        for _ in 0..2 {
            let head_end = rest.windows(4).position(|w| w == b"\r\n\r\n").expect("response head") + 4;
            let head = String::from_utf8_lossy(&rest[..head_end]).to_string();
            assert!(head.starts_with("HTTP/1.1 200"), "{head}");
            assert!(!head.contains("Transfer-Encoding"), "{head}");
            assert!(rest[head_end..].len() >= content.len());
            assert!(rest[head_end..head_end + content.len()] == content[..], "body differs from the file");
            rest = &rest[head_end + content.len()..];
        }
        assert!(rest.is_empty());
    }


    #[tokio::test(start_paused = true)]
    async fn idle_timer_restarts_on_every_body_read() {
        // Body pieces 20 s apart: 40 s in total, but never 30 s without a read.
        let (mut client, server) = tokio::io::duplex(1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Echo));
        client.write_all(b"POST /slow HTTP/1.1\r\nHost: x\r\nContent-Length: 6\r\nConnection: close\r\n\r\nab").await.unwrap();
        tokio::time::sleep(Duration::from_secs(20)).await;
        client.write_all(b"cd").await.unwrap();
        tokio::time::sleep(Duration::from_secs(20)).await;
        client.write_all(b"ef").await.unwrap();
        let out = read_to_end(&mut client).await;
        assert!(out.contains("uri=/slow body=abcdef"), "{out}");
        assert!(task.await.unwrap().is_ok());
    }

    #[tokio::test(start_paused = true)]
    async fn a_head_trickled_past_the_head_deadline_gets_408() {
        // Pieces 20 s apart never trip the idle timeout, but the head as a
        // whole must arrive within REQUEST_HEAD_TIMEOUT.
        let (mut client, server) = tokio::io::duplex(1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        client.write_all(b"GET /slow HTTP/1.1\r\n").await.unwrap();
        tokio::time::sleep(Duration::from_secs(20)).await;
        client.write_all(b"Host: x\r\n").await.unwrap();
        tokio::time::sleep(Duration::from_secs(20)).await;
        let _ = client.write_all(b"\r\n").await;
        let out = read_to_end(&mut client).await;
        assert!(out.starts_with("HTTP/1.1 408"), "{out}");
        assert!(task.await.unwrap().is_ok());
    }

    #[tokio::test(start_paused = true)]
    async fn waiting_for_the_next_request_is_not_a_head_timeout() {
        // 25 s idle between keep-alive requests: only the idle timeout applies.
        let (mut client, server) = tokio::io::duplex(1024);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        client.write_all(b"GET /a HTTP/1.1\r\nHost: x\r\n\r\n").await.unwrap();
        let first = read_until_count(&mut client, "hello", 1).await;
        assert!(first.starts_with("HTTP/1.1 200"));
        tokio::time::sleep(Duration::from_secs(25)).await;
        client.write_all(b"GET /b HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
        let second = read_until_count(&mut client, "hello", 1).await;
        assert!(second.starts_with("HTTP/1.1 200"), "{second}");
        assert!(task.await.unwrap().is_ok());
    }


    // ── #31: streamed request bodies ────────────────────────────────────────

    /// Streams `/stream*` bodies; reports bytes read, or the read error.
    /// `/stream/early` answers without reading. Each read's size goes to
    /// `reads` so a test can see the handler running mid-upload.
    #[derive(Clone)]
    struct StreamApp {
        reads: std::sync::mpsc::Sender<usize>,
    }

    impl Application for StreamApp {
        fn execute(&self, request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
            use std::io::Read;
            if request.request_uri == "/stream/early" {
                return Ok(Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "early"));
            }
            let mut body = &request.body;
            let mut total = 0;
            let mut buf = [0u8; 8192];
            loop {
                match body.read(&mut buf) {
                    Ok(0) => break,
                    Ok(n) => {
                        total += n;
                        let _ = self.reads.send(n);
                    }
                    Err(e) => return Ok(Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, &format!("error: {e}"))),
                }
            }
            Ok(Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, &format!("streamed {total} bytes, is_stream={}", request.body.is_stream())))
        }

        fn streams_body(&self, request: &Request, _connection: &ConnectionInfo) -> bool {
            request.request_uri.starts_with("/stream")
        }
    }

    fn stream_app() -> (StreamApp, std::sync::mpsc::Receiver<usize>) {
        let (reads, rx) = std::sync::mpsc::channel();
        (StreamApp { reads }, rx)
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_handler_reads_the_body_while_it_arrives() {
        let (app, reads) = stream_app();
        let (mut client, server) = tokio::io::duplex(1 << 20);
        let task = tokio::spawn(serve(server, peer(), None, false, app));
        client.write_all(b"PUT /stream HTTP/1.1\r\nHost: x\r\nContent-Length: 200000\r\n\r\n").await.unwrap();
        client.write_all(&vec![b'a'; 100_000]).await.unwrap();
        // The handler has read part of the body before the rest is sent.
        let first = tokio::task::spawn_blocking(move || {
            let n = reads.recv_timeout(std::time::Duration::from_secs(5)).expect("no read before the body ended");
            (n, reads)
        }).await.unwrap();
        assert!(first.0 > 0);
        client.write_all(&vec![b'b'; 100_000]).await.unwrap();
        let response = read_until_count(&mut client, "bytes", 1).await;
        assert!(response.contains("streamed 200000 bytes, is_stream=true"), "{response}");
        assert!(response.contains("Connection: keep-alive"), "{response}");
        // The connection carries the next request.
        client.write_all(b"POST /stream HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n5\r\nhello\r\n6\r\n world\r\n0\r\n\r\n").await.unwrap();
        let second = read_to_end(&mut client).await;
        assert!(second.contains("streamed 11 bytes"), "{second}");
        assert!(task.await.unwrap().is_ok());
        drop(first.1);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_client_that_stops_mid_body_is_an_error_for_the_handler() {
        let (app, _reads) = stream_app();
        let (mut client, server) = tokio::io::duplex(1 << 20);
        let task = tokio::spawn(serve(server, peer(), None, false, app));
        client.write_all(b"PUT /stream HTTP/1.1\r\nHost: x\r\nContent-Length: 1000\r\n\r\nonly this").await.unwrap();
        client.shutdown().await.unwrap();
        let response = read_to_end(&mut client).await;
        assert!(response.contains("error: the client stopped sending the request body"), "{response}");
        assert!(response.contains("Connection: close"), "{response}");
        let _ = task.await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_handler_that_stops_reading_closes_the_connection() {
        let (app, _reads) = stream_app();
        let (mut client, server) = tokio::io::duplex(1 << 20);
        let task = tokio::spawn(serve(server, peer(), None, false, app));
        client.write_all(b"PUT /stream/early HTTP/1.1\r\nHost: x\r\nContent-Length: 1000000\r\n\r\nabc").await.unwrap();
        let response = read_to_end(&mut client).await;
        assert!(response.contains("early"), "{response}");
        assert!(response.contains("Connection: close"), "{response}");
        assert!(task.await.unwrap().is_ok());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn broken_chunked_framing_is_an_error_for_the_handler() {
        let (app, _reads) = stream_app();
        let (mut client, server) = tokio::io::duplex(1 << 20);
        let _task = tokio::spawn(serve(server, peer(), None, false, app));
        client.write_all(b"PUT /stream HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\nzz\r\n").await.unwrap();
        let response = read_until_count(&mut client, "error", 1).await;
        assert!(response.contains("error:"), "{response}");
        assert!(response.contains("Connection: close"), "{response}");
    }

    #[tokio::test]
    async fn without_opting_in_the_body_is_read_first() {
        let (mut client, server) = tokio::io::duplex(1 << 20);
        let task = tokio::spawn(serve(server, peer(), None, false, Echo));
        client.write_all(b"POST /x HTTP/1.1\r\nHost: x\r\nContent-Length: 3\r\nConnection: close\r\n\r\nabc").await.unwrap();
        let response = read_to_end(&mut client).await;
        assert!(response.starts_with("HTTP/1.1 200"), "{response}");
        assert!(task.await.unwrap().is_ok());
    }

    fn peak_rss_kib() -> u64 {
        std::fs::read_to_string("/proc/self/status").ok()
            .and_then(|status| status.lines().find(|l| l.starts_with("VmHWM:"))
                .and_then(|l| l.split_whitespace().nth(1)?.parse().ok()))
            .unwrap_or(0)
    }

    /// The exit criterion of #31: a large upload in constant memory. Run
    /// alone (`--ignored --test-threads=1`): other tests in the process
    /// move the peak.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    #[ignore]
    async fn a_large_streamed_upload_runs_in_constant_memory() {
        const TOTAL: usize = 512 * 1024 * 1024;
        let (app, reads) = stream_app();
        std::thread::spawn(move || while reads.recv().is_ok() {});
        let before = peak_rss_kib();
        let (client, server) = tokio::io::duplex(1 << 16);
        let task = tokio::spawn(serve(server, peer(), None, false, app));
        let (mut read_half, mut write_half) = tokio::io::split(client);
        let writer = tokio::spawn(async move {
            write_half.write_all(format!("PUT /stream HTTP/1.1\r\nHost: x\r\nContent-Length: {TOTAL}\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap();
            let chunk = vec![b'z'; 1 << 20];
            for _ in 0..TOTAL / chunk.len() {
                write_half.write_all(&chunk).await.unwrap();
            }
        });
        let mut response = Vec::new();
        read_half.read_to_end(&mut response).await.unwrap();
        writer.await.unwrap();
        let response = String::from_utf8_lossy(&response);
        assert!(response.contains(&format!("streamed {TOTAL} bytes")), "{response}");
        let grown_mib = (peak_rss_kib().saturating_sub(before)) / 1024;
        assert!(grown_mib < 64, "peak memory grew {grown_mib} MiB for a 512 MiB upload");
        let _ = task.await;
    }

    /// Control for the test above: the same upload read into memory first
    /// does show up in the peak — so the measurement can tell.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    #[ignore]
    async fn a_large_buffered_upload_shows_in_peak_memory() {
        const TOTAL: usize = 256 * 1024 * 1024;
        let before = peak_rss_kib();
        let (client, server) = tokio::io::duplex(1 << 16);
        let task = tokio::spawn(serve(server, peer(), None, false, Hello));
        let (mut read_half, mut write_half) = tokio::io::split(client);
        let writer = tokio::spawn(async move {
            write_half.write_all(format!("PUT /x HTTP/1.1\r\nHost: x\r\nContent-Length: {TOTAL}\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap();
            let chunk = vec![b'z'; 1 << 20];
            for _ in 0..TOTAL / chunk.len() {
                write_half.write_all(&chunk).await.unwrap();
            }
        });
        let mut response = Vec::new();
        read_half.read_to_end(&mut response).await.unwrap();
        writer.await.unwrap();
        let grown_mib = (peak_rss_kib().saturating_sub(before)) / 1024;
        assert!(grown_mib >= 200, "a 256 MiB buffered body grew the peak only {grown_mib} MiB");
        let _ = task.await;
    }

    /// #31's exit criterion end to end: a multi-gigabyte HTTP upload, read
    /// by the handler as a stream and sent to S3 in parts, in constant
    /// memory. Needs an S3-compatible server (`RWS_S3_E2E_*`, as for
    /// `storage::s3` tests); run alone with `--ignored --test-threads=1`.
    #[cfg(feature = "storage-s3")]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    #[ignore]
    async fn a_multi_gigabyte_upload_streams_to_s3_in_constant_memory() {
        use crate::storage::{S3Config, S3Storage, Storage};
        const TOTAL: usize = 2 * 1024 * 1024 * 1024;
        let env = |name: &str| std::env::var(format!("RWS_S3_E2E_{name}")).expect("set RWS_S3_E2E_*");
        let storage = std::sync::Arc::new(S3Storage::new(S3Config {
            bucket: env("BUCKET"),
            region: "us-east-1".to_string(),
            access_key: env("ACCESS_KEY"),
            secret_key: env("SECRET_KEY"),
            endpoint: env("ENDPOINT"),
        }));

        #[derive(Clone)]
        struct ToS3(std::sync::Arc<S3Storage>);
        impl Application for ToS3 {
            fn execute(&self, request: &Request, _c: &ConnectionInfo) -> Result<Response, String> {
                let stored = self.0.put_stream("rws-e2e/huge.bin", &mut &request.body, "application/octet-stream");
                Ok(Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, &format!("{stored:?}")))
            }
            fn streams_body(&self, _r: &Request, _c: &ConnectionInfo) -> bool {
                true
            }
        }

        let before = peak_rss_kib();
        let (client, server) = tokio::io::duplex(1 << 16);
        let task = tokio::spawn(serve(server, peer(), None, false, ToS3(storage.clone())));
        let (mut read_half, mut write_half) = tokio::io::split(client);
        let writer = tokio::spawn(async move {
            write_half.write_all(format!("PUT /upload HTTP/1.1\r\nHost: x\r\nContent-Length: {TOTAL}\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap();
            let chunk = vec![b'u'; 1 << 20];
            for _ in 0..TOTAL / chunk.len() {
                write_half.write_all(&chunk).await.unwrap();
            }
        });
        let mut response = Vec::new();
        read_half.read_to_end(&mut response).await.unwrap();
        writer.await.unwrap();
        let response = String::from_utf8_lossy(&response).to_string();
        assert!(response.contains("Ok(\"rws-e2e/huge.bin\")"), "{response}");
        let grown_mib = peak_rss_kib().saturating_sub(before) / 1024;
        assert!(grown_mib < 96, "a 2 GiB upload to S3 grew the peak by {grown_mib} MiB");
        let _ = task.await;
        tokio::task::spawn_blocking(move || storage.delete("rws-e2e/huge.bin").unwrap()).await.unwrap();
    }
}