mini_static/

server.rs

use std::convert::Infallible;
use std::fs;
use std::io::Read;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::{Duration, SystemTime};

use bytes::Bytes;
use hyper::{Method, Response, StatusCode, Request};
use hyper::service::service_fn;
use http_body_util::Full;
use hyper::body::Incoming;
use hyper_util::rt::TokioExecutor;
use hyper_util::rt::TokioIo;
use hyper_util::server::conn::auto::Builder as AutoBuilder;
use tokio::fs::File;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, ReadBuf};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::time::timeout;

use crate::error::StaticError;
use crate::handler::{FileBody, ResponseBody};
use crate::reload::{self, SseBody};
use crate::resolve;
use crate::watcher::{start_watching, Broadcaster};

const ACCEPT_BACKOFF_INITIAL: Duration = Duration::from_millis(10);
const ACCEPT_BACKOFF_MAX: Duration = Duration::from_secs(1);

/// A source of accepted TCP connections. Abstracted so the accept-error backoff below
/// can be exercised against a listener that fails on demand, without needing to provoke
/// real OS-level accept errors (e.g. EMFILE) in tests.
trait TcpAccept {
    async fn accept(&self) -> std::io::Result<(TcpStream, SocketAddr)>;
}

impl TcpAccept for TcpListener {
    async fn accept(&self) -> std::io::Result<(TcpStream, SocketAddr)> {
        TcpListener::accept(self).await
    }
}

/// Exponential backoff for retrying `accept()` after an error, so a sustained failure
/// (e.g. the process is out of file descriptors) degrades into periodic retries instead
/// of a CPU-bound busy spin or, worse, silently ending the accept loop for good. Resets
/// to the initial delay as soon as an accept succeeds.
struct Backoff {
    delay: Duration,
}

impl Backoff {
    fn new() -> Self {
        Backoff { delay: ACCEPT_BACKOFF_INITIAL }
    }

    fn next_delay(&mut self) -> Duration {
        let delay = self.delay;
        self.delay = (self.delay * 2).min(ACCEPT_BACKOFF_MAX);
        delay
    }

    fn reset(&mut self) {
        self.delay = ACCEPT_BACKOFF_INITIAL;
    }
}

/// Accept a connection and reserve it a connection-limit permit, retrying transient
/// `accept()` errors with `Backoff` instead of ending the accept loop on the first one.
/// Returns `None` only if the semaphore itself has been closed (never happens in normal
/// operation, since nothing ever calls `close()` on it — handled so a caller can still
/// fail safely rather than panic).
async fn accept_and_permit<L: TcpAccept>(
    listener: &L,
    backoff: &mut Backoff,
    semaphore: &Arc<Semaphore>,
) -> Option<(TcpStream, OwnedSemaphorePermit)> {
    loop {
        let stream = match listener.accept().await {
            Ok((stream, _)) => {
                backoff.reset();
                stream
            }
            Err(_) => {
                tokio::time::sleep(backoff.next_delay()).await;
                continue;
            }
        };
        return match semaphore.clone().acquire_owned().await {
            Ok(permit) => Some((stream, permit)),
            Err(_) => None,
        };
    }
}

/// A static file server for serving files securely from a root directory.
///
/// `Server` canonicalizes the root directory once at creation time and uses the
/// canonical form for all subsequent requests, avoiding repeated filesystem calls.
///
/// # Security
///
/// The server protects against:
/// - Path traversal attacks (e.g., `../../etc/passwd`)
/// - Accessing files outside the root via symlinks
/// - Disclosing filesystem structure (traversal and missing files both return 404)
///
/// # Cloning
///
/// `Server` is cheap to clone (a `PathBuf` and a `usize`). Multiple clones can be used
/// concurrently in async tasks without synchronization overhead.
///
/// # Example
///
/// ```no_run
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// use mini_static::Server;
/// use std::path::Path;
/// use std::time::Duration;
///
/// let server = Server::new(Path::new("./public"))?;
/// let (port, _handle) = server.run(Duration::from_secs(30)).await?;
/// println!("Server running on port {}", port);
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct Server {
    root_canon: PathBuf,
    max_connections: usize,
    live_reload: bool,
    broadcaster: Option<Broadcaster>,
}

impl Server {
    /// Create a new server with the given root directory.
    ///
    /// Canonicalizes the root once at startup. All subsequent requests use the
    /// canonical root without re-canonicalizing it, making this suitable for long-lived servers.
    ///
    /// # Arguments
    ///
    /// * `root` - The root directory to serve files from.
    ///
    /// # Errors
    ///
    /// Returns `Err(StaticError::Io)` if the root cannot be canonicalized (e.g., doesn't exist,
    /// no read permissions).
    pub fn new(root: &Path) -> Result<Self, StaticError> {
        let root_canon = root.canonicalize().map_err(StaticError::Io)?;
        Ok(Server {
            root_canon,
            max_connections: DEFAULT_MAX_CONNECTIONS,
            live_reload: false,
            broadcaster: None,
        })
    }

    /// Set the maximum number of connections served concurrently (default 1024).
    ///
    /// Once this many connections are in flight, `run()`'s accept loop stops accepting
    /// new ones — without pausing the accept loop, a client that opens a connection and
    /// sends nothing (see the header-read timeout docs on `run()`) could otherwise be
    /// used, in enough parallel copies, to exhaust the process's file descriptors or
    /// memory with no bound at all.
    pub fn with_max_connections(mut self, max: usize) -> Self {
        self.max_connections = max;
        self
    }

    /// Enable live-reload for this server (disabled by default).
    ///
    /// Once enabled, `run()`/`run_all()`/`run_on()` start a background watcher (mtime
    /// polling, bounded 500ms interval — see [`crate::start_watching`]) over the
    /// server's root the first time the server actually starts accepting connections,
    /// and:
    ///
    /// - serve a live-reload SSE stream at [`crate::reload::LIVE_RELOAD_PATH`],
    ///   broadcasting a change event (with [`crate::ChangeType`]) whenever a served file
    ///   is added, modified, or removed;
    /// - inject a small `<script>` into every served `text/html` response that connects
    ///   to that stream and reloads the page (or hot-swaps stylesheet `<link>`s, for CSS
    ///   changes) — no manual client wiring required.
    ///
    /// This is meant for local development, not production: leave it disabled (the
    /// default) for any server serving real traffic. A typical call site gates it behind
    /// `#[cfg(debug_assertions)]` so a release build never pays for the watcher or the
    /// injected script.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use mini_static::Server;
    /// use std::path::Path;
    ///
    /// let server = Server::new(Path::new("./public"))?;
    /// #[cfg(debug_assertions)]
    /// let server = server.with_live_reload();
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_live_reload(mut self) -> Self {
        self.live_reload = true;
        self
    }

    /// Resolve a request path under the server's root.
    ///
    /// This is a lower-level API for resolving paths without generating HTTP responses.
    /// For most use cases, prefer `handle_request_with_method()` or the `run()` methods.
    ///
    /// # Arguments
    ///
    /// * `request_path` - The HTTP request path (e.g., `/path/to/file.html`).
    ///
    /// # Returns
    ///
    /// - `Ok(PathBuf)` if the path resolves to a file within root.
    /// - `Err(StaticError)` if the path is invalid, missing, or attempts traversal.
    pub fn resolve(&self, request_path: &str) -> Result<PathBuf, StaticError> {
        resolve::resolve_with_canonical_root(&self.root_canon, request_path)
    }

    /// Handle an HTTP GET request for a resource path.
    ///
    /// Convenience method equivalent to `handle_request_with_method(&Method::GET, request_path)`.
    ///
    /// # Arguments
    ///
    /// * `request_path` - The HTTP request path (e.g., `/index.html`).
    pub fn handle_request(&self, request_path: &str) -> Response<ResponseBody> {
        self.handle_request_with_method(&Method::GET, request_path)
    }

    /// Handle an HTTP request with an explicit method.
    ///
    /// Only GET and HEAD methods are allowed. Other methods return 405 Method Not Allowed
    /// with an Allow header listing the permitted methods.
    ///
    /// # Arguments
    ///
    /// * `method` - The HTTP method (GET and HEAD are allowed; others return 405).
    /// * `request_path` - The HTTP request path (e.g., `/index.html`).
    pub fn handle_request_with_method(
        &self,
        method: &Method,
        request_path: &str,
    ) -> Response<ResponseBody> {
        self.handle_request_with_headers(method, request_path, None, None)
    }

    /// Run the server on a specific address with a configurable header-read timeout.
    ///
    /// Spawns the server in a background Tokio task and returns immediately with the
    /// assigned port number and a [`ServerHandle`]. Call `handle.shutdown().await` to
    /// stop accepting new connections and wait for in-flight connections to finish.
    ///
    /// # Header-Read Timeout
    ///
    /// Connections that don't send complete HTTP headers within `header_timeout` are closed.
    /// This prevents slowloris attacks and resource exhaustion from incomplete requests. The
    /// timeout applies only to the header-read phase — once a complete header block has been
    /// read, the connection is handed off with no further time bound, so long-lived response
    /// bodies (e.g. the live-reload SSE stream from [`Server::with_live_reload`]) are not cut
    /// off mid-stream.
    ///
    /// # Arguments
    ///
    /// * `addr` - Socket address to bind to (e.g., `127.0.0.1:0` for loopback ephemeral,
    ///   or `0.0.0.0:8080` to bind all interfaces on a fixed port).
    /// * `header_timeout` - Maximum time to wait for complete HTTP headers on each connection.
    ///
    /// # Returns
    ///
    /// - `Ok((u16, ServerHandle))` with the assigned port number and a handle for graceful shutdown.
    /// - `Err(StaticError::Io)` if binding to the socket fails.
    pub async fn run_on(&self, addr: SocketAddr, header_timeout: Duration) -> Result<(u16, ServerHandle), StaticError> {
        let listener = TcpListener::bind(addr)
            .await
            .map_err(StaticError::Io)?;
        let port = listener
            .local_addr()
            .map_err(StaticError::Io)?
            .port();

        let mut server = self.clone();
        if server.live_reload {
            let broadcaster = Broadcaster::new();
            start_watching(Arc::new(server.root_canon.clone()), broadcaster.clone());
            server.broadcaster = Some(broadcaster);
        }
        let semaphore = Arc::new(Semaphore::new(server.max_connections));
        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();

        let accept_task = tokio::spawn(async move {
            let mut backoff = Backoff::new();
            let mut join_set: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
            let mut shutdown_pin = std::pin::pin!(shutdown_rx);
            let mut shutting_down = false;

            loop {
                if !shutting_down {
                    // The accept-and-permit step and the shutdown signal race in a single
                    // `select!` so shutdown can preempt a pending accept or a permit wait
                    // cleanly, at any point — not just between loop iterations.
                    tokio::select! {
                        accepted = accept_and_permit(&listener, &mut backoff, &semaphore) => {
                            match accepted {
                                Some((stream, permit)) => {
                                    let server = server.clone();
                                    join_set.spawn(async move {
                                        let _permit = permit;
                                        serve_connection(stream, server, header_timeout).await;
                                    });
                                }
                                None => shutting_down = true,
                            }
                        }
                        _ = shutdown_pin.as_mut() => {
                            shutting_down = true;
                        }
                    }
                    continue;
                }

                // Stop accepting; drain already-spawned connections before returning.
                match join_set.join_next().await {
                    Some(_) => continue,
                    None => break,
                }
            }
        });

        Ok((port, ServerHandle { shutdown_tx: Some(shutdown_tx), accept_task }))
    }

    /// Run the server on loopback (127.0.0.1) with a configurable header-read timeout.
    ///
    /// Binds to an ephemeral port and spawns the server in a background Tokio task.
    /// Returns immediately with the assigned port number and a [`ServerHandle`]. Dropping
    /// the handle without calling `shutdown()` leaves the server running in the
    /// background for the life of the process — the same behavior `run()` always had.
    /// Call `handle.shutdown().await` to stop accepting new connections and wait for
    /// in-flight connections to finish.
    ///
    /// # Header-Read Timeout
    ///
    /// Connections that don't send complete HTTP headers within `header_timeout` are closed.
    /// This prevents slowloris attacks and resource exhaustion from incomplete requests. The
    /// timeout applies only to the header-read phase — once a complete header block has been
    /// read, the connection is handed off with no further time bound, so long-lived response
    /// bodies (e.g. the live-reload SSE stream from [`Server::with_live_reload`]) are not cut
    /// off mid-stream.
    ///
    /// # Arguments
    ///
    /// * `header_timeout` - Maximum time to wait for complete HTTP headers on each connection.
    ///
    /// # Returns
    ///
    /// - `Ok((u16, ServerHandle))` with the ephemeral port number assigned by the OS and
    ///   a handle for graceful shutdown.
    /// - `Err(StaticError::Io)` if binding to the socket fails.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use mini_static::Server;
    /// use std::path::Path;
    /// use std::time::Duration;
    ///
    /// let server = Server::new(Path::new("./public"))?;
    /// let (port, handle) = server.run(Duration::from_secs(30)).await?;
    /// println!("Server running on http://127.0.0.1:{}", port);
    /// // ... later, to stop it gracefully:
    /// handle.shutdown().await;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn run(&self, header_timeout: Duration) -> Result<(u16, ServerHandle), StaticError> {
        let addr: SocketAddr = ([127, 0, 0, 1], 0).into();
        self.run_on(addr, header_timeout).await
    }

    /// Run the server on all interfaces (0.0.0.0) with a configurable header-read timeout.
    ///
    /// Binds to a specified port on all network interfaces. Useful for containerized
    /// deployments, reverse-proxy setups, or services that need to accept connections
    /// from anywhere. Spawns the server in a background Tokio task and returns immediately
    /// with the assigned port and a [`ServerHandle`].
    ///
    /// # Header-Read Timeout
    ///
    /// Connections that don't send complete HTTP headers within `header_timeout` are closed.
    /// This prevents slowloris attacks and resource exhaustion from incomplete requests. The
    /// timeout applies only to the header-read phase — once a complete header block has been
    /// read, the connection is handed off with no further time bound, so long-lived response
    /// bodies (e.g. the live-reload SSE stream from [`Server::with_live_reload`]) are not cut
    /// off mid-stream.
    ///
    /// # Arguments
    ///
    /// * `port` - Port number to bind to (0 for ephemeral port assignment).
    /// * `header_timeout` - Maximum time to wait for complete HTTP headers on each connection.
    ///
    /// # Returns
    ///
    /// - `Ok((u16, ServerHandle))` with the assigned port number and a handle for graceful shutdown.
    /// - `Err(StaticError::Io)` if binding to the socket fails.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use mini_static::Server;
    /// use std::path::Path;
    /// use std::time::Duration;
    ///
    /// let server = Server::new(Path::new("./public"))?;
    /// let (_port, handle) = server.run_all(8080, Duration::from_secs(30)).await?;
    /// println!("Server listening on 0.0.0.0:8080");
    /// handle.shutdown().await;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn run_all(&self, port: u16, header_timeout: Duration) -> Result<(u16, ServerHandle), StaticError> {
        let addr: SocketAddr = ([0, 0, 0, 0], port).into();
        self.run_on(addr, header_timeout).await
    }

    /// Run the server on loopback (127.0.0.1) with a default header-read timeout.
    ///
    /// Convenience wrapper around `run()` that uses a default 30-second header-read timeout.
    /// Returns immediately with the ephemeral port number and a [`ServerHandle`]; the server
    /// continues in a background Tokio task until the handle's `shutdown()` is awaited or
    /// the Tokio runtime shuts down.
    ///
    /// This is the recommended method for tests and lightweight services that don't require
    /// custom timeout configuration.
    ///
    /// # Returns
    ///
    /// - `Ok((u16, ServerHandle))` with the ephemeral port number assigned by the OS and
    ///   a handle for graceful shutdown.
    /// - `Err(StaticError::Io)` if binding to the socket fails.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use mini_static::Server;
    /// use std::path::Path;
    ///
    /// let server = Server::new(Path::new("./public"))?;
    /// let (port, handle) = server.run_ephemeral().await?;
    /// println!("Server ready on http://127.0.0.1:{}", port);
    /// handle.shutdown().await;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn run_ephemeral(&self) -> Result<(u16, ServerHandle), StaticError> {
        self.run(Duration::from_secs(30)).await
    }

    /// Handle an HTTP request asynchronously, streaming file bodies to the client.
    ///
    /// This is the method to call when embedding `mini-static` inside another async
    /// server's request-handling path (e.g. as a catch-all fallback route). Unlike
    /// [`Server::handle_request`] and its synchronous siblings, this method never blocks
    /// the calling task: path resolution runs on Tokio's blocking-thread pool via
    /// `spawn_blocking`, and the file is read via async I/O.
    ///
    /// File responses are backed by `FileBody`, which reads and hands off one 64 KB
    /// chunk to hyper at a time as `poll_frame` is driven — memory use stays bounded to
    /// one chunk per in-flight response regardless of file size, and no chunk is copied
    /// or zero-filled beyond what the read syscall itself writes.
    ///
    /// Conditional requests (If-None-Match, If-Modified-Since) are honored: if the
    /// request includes a validator that matches the file's ETag, returns 304 Not Modified.
    pub async fn handle_request_async(
        &self,
        method: &Method,
        request_path: &str,
        if_none_match: Option<&str>,
        if_modified_since: Option<&str>,
    ) -> Response<ResponseBody> {
        // Gate on HTTP method
        if method != Method::GET && method != Method::HEAD {
            return finish(Response::builder()
                .status(StatusCode::METHOD_NOT_ALLOWED)
                .header("Allow", "GET, HEAD")
                .header("X-Content-Type-Options", "nosniff")
                .body(into_response_body(Full::new(Bytes::from("method not allowed\n")))));
        }

        // Live-reload SSE stream — only reachable when `with_live_reload()` was called
        // and the server was started via `run()`/`run_all()`/`run_on()` (those are the
        // only paths that populate `broadcaster`).
        if *method == Method::GET && request_path == reload::LIVE_RELOAD_PATH {
            if let Some(broadcaster) = &self.broadcaster {
                let rx = broadcaster.subscribe();
                return finish(Response::builder()
                    .status(StatusCode::OK)
                    .header("Content-Type", "text/event-stream")
                    .header("Cache-Control", "no-cache")
                    .header("Connection", "keep-alive")
                    .header("X-Content-Type-Options", "nosniff")
                    .body(ResponseBody::Sse(SseBody::new(rx))));
            }
        }

        // `resolve()` does blocking filesystem syscalls (`canonicalize()`, up to two per
        // request). Running those directly in this `async fn` would block whichever
        // Tokio worker thread happens to be driving it, stalling every other task
        // scheduled on that thread for the duration of the syscalls. `spawn_blocking`
        // moves the work onto Tokio's dedicated blocking thread pool instead.
        let server = self.clone();
        let owned_request_path = request_path.to_string();
        let resolved = tokio::task::spawn_blocking(move || server.resolve(&owned_request_path)).await;
        let resolved = match resolved {
            Ok(r) => r,
            Err(_) => return internal_error_response(),
        };

        match resolved {
            Ok(path) => {
                let decoded_request_path = resolve::decode_request_path(request_path);
                if path.file_name().is_some_and(|name| name == "index.html")
                    && !decoded_request_path.ends_with('/')
                    && !decoded_request_path.ends_with("index.html")
                {
                    let location = format!("{}/", request_path.trim_end_matches('/'));
                    // `location` is built from the (attacker-controlled) request path;
                    // `finish()` degrades to 400 instead of panicking if it ever contains
                    // bytes invalid in a header value.
                    return finish(Response::builder()
                        .status(StatusCode::MOVED_PERMANENTLY)
                        .header("Location", location)
                        .header("X-Content-Type-Options", "nosniff")
                        .body(into_response_body(Full::new(Bytes::from("moved\n")))));
                }

                // Use async file operations for streaming
                let file = match File::open(&path).await {
                    Ok(f) => f,
                    Err(_) => return internal_error_response(),
                };

                let metadata = match file.metadata().await {
                    Ok(m) => m,
                    Err(_) => return internal_error_response(),
                };

                let mut file_size = metadata.len();
                let etag = generate_etag(&metadata);

                // Check If-None-Match (ETag) for 304 Not Modified
                if let Some(if_none_match) = if_none_match {
                    if is_etag_match(if_none_match, &etag) {
                        return finish(Response::builder()
                            .status(StatusCode::NOT_MODIFIED)
                            .header("ETag", etag)
                            .body(into_response_body(Full::new(Bytes::new()))));
                    }
                }

                // Check If-Modified-Since (mtime) for 304 Not Modified
                if let Some(if_modified_since) = if_modified_since {
                    if is_not_modified_since(if_modified_since, &metadata) {
                        return finish(Response::builder()
                            .status(StatusCode::NOT_MODIFIED)
                            .header("ETag", etag)
                            .body(into_response_body(Full::new(Bytes::new()))));
                    }
                }

                let content_type = mime_type_for_path(&path);

                // HEAD must not return a body (RFC 9110); skip opening the read stream
                // entirely since we'd just discard every chunk.
                //
                // Live-reload HTML injection reads the whole file into memory instead of
                // streaming it — acceptable only because it's gated on `broadcaster`
                // being set, i.e. `with_live_reload()` was called for local development;
                // the streamed path remains untouched for every production response.
                let body: ResponseBody = if *method == Method::HEAD {
                    into_response_body(Full::new(Bytes::new()))
                } else if self.broadcaster.is_some() && content_type.starts_with("text/html") {
                    let mut html = Vec::with_capacity(file_size as usize);
                    let mut file = file;
                    if file.read_to_end(&mut html).await.is_err() {
                        return internal_error_response();
                    }
                    reload::inject_reload_script(&mut html);
                    file_size = html.len() as u64;
                    into_response_body(Full::new(Bytes::from(html)))
                } else {
                    ResponseBody::Streamed(FileBody::new(file))
                };

                finish(Response::builder()
                    .status(StatusCode::OK)
                    .header("X-Content-Type-Options", "nosniff")
                    .header("Content-Type", content_type)
                    .header("Content-Length", file_size.to_string())
                    .header("ETag", etag)
                    .body(body))
            }
            Err(e) => {
                let message = e.user_message();
                let body = format!("{}\n", message);

                finish(Response::builder()
                    .status(StatusCode::NOT_FOUND)
                    .header("X-Content-Type-Options", "nosniff")
                    .body(into_response_body(Full::new(Bytes::from(body)))))
            }
        }
    }

    /// Handle an HTTP request with method and optional Range/If-Range headers (synchronous API).
    ///
    /// This is the synchronous version of request handling used internally by the
    /// async server loop. For most use cases, prefer using `run()` or `run_ephemeral()`
    /// which handle the full async lifecycle.
    ///
    /// Only GET and HEAD methods are allowed; other methods return 405 Method Not Allowed.
    /// All errors (missing files, traversal attempts, I/O failures) are returned as 404
    /// to avoid leaking filesystem structure information.
    ///
    /// # Range Request Handling
    ///
    /// mini-static does not yet serve `206 Partial Content` — every request,
    /// ranged or not, gets the full body with `200`. This is RFC 9110-correct behavior
    /// (as opposed to incorrectly answering `416`), but partial-content serving is
    /// deferred to a later phase.
    ///
    /// # Arguments
    ///
    /// * `method` - The HTTP method (GET and HEAD only).
    /// * `request_path` - The HTTP request path (e.g., `/index.html`).
    /// * `_range_header` - Optional Range header (currently unused).
    /// * `_if_range_header` - Optional If-Range header (currently unused).
    pub fn handle_request_with_headers(
        &self,
        method: &Method,
        request_path: &str,
        _range_header: Option<&str>,
        _if_range_header: Option<&str>,
    ) -> Response<ResponseBody> {
        // Gate on HTTP method
        if method != Method::GET && method != Method::HEAD {
            return finish(Response::builder()
                .status(StatusCode::METHOD_NOT_ALLOWED)
                .header("Allow", "GET, HEAD")
                .header("X-Content-Type-Options", "nosniff")
                .body(into_response_body(Full::new(Bytes::from("method not allowed\n")))));
        }

        // Method is allowed; resolve the path
        match self.resolve(request_path) {
            Ok(path) => {
                // Check if resolved path is index.html but request_path doesn't end with /
                // If so, redirect to path/ to establish correct base for relative links.
                // Compare against the *decoded* request path so a percent-encoded explicit
                // request for index.html (e.g. `/docs/index.htm%6c`) is recognized as such
                // instead of producing a redirect to a still-encoded, broken Location.
                let decoded_request_path = resolve::decode_request_path(request_path);
                if path.file_name().is_some_and(|name| name == "index.html")
                    && !decoded_request_path.ends_with('/')
                    && !decoded_request_path.ends_with("index.html")
                {
                    let location = format!("{}/", request_path.trim_end_matches('/'));

                    // Location is built from the (attacker-controlled) request path;
                    // `finish()` degrades to 400 instead of panicking if it ever contains
                    // bytes invalid in a header value.
                    return finish(Response::builder()
                        .status(StatusCode::MOVED_PERMANENTLY)
                        .header("Location", location)
                        .header("X-Content-Type-Options", "nosniff")
                        .body(into_response_body(Full::new(Bytes::from("moved\n")))));
                }

                let file = match fs::File::open(&path) {
                    Ok(f) => f,
                    Err(_) => return internal_error_response(),
                };
                let metadata = match file.metadata() {
                    Ok(m) => m,
                    Err(_) => return internal_error_response(),
                };
                let file_size = metadata.len();
                let etag = generate_etag(&metadata);

                // HEAD must not return a body (RFC 9110); avoid reading file content we'd
                // just discard.
                let body_bytes = if *method == Method::HEAD {
                    Bytes::new()
                } else {
                    let mut buf = Vec::with_capacity(file_size as usize);
                    let mut file = file;
                    if file.read_to_end(&mut buf).is_err() {
                        return internal_error_response();
                    }
                    Bytes::from(buf)
                };

                finish(Response::builder()
                    .status(StatusCode::OK)
                    .header("X-Content-Type-Options", "nosniff")
                    .header("Content-Length", file_size.to_string())
                    .header("ETag", etag)
                    .body(into_response_body(Full::new(body_bytes))))
            }
            Err(e) => {
                let message = e.user_message();
                let body = format!("{}\n", message);

                finish(Response::builder()
                    .status(StatusCode::NOT_FOUND)
                    .header("X-Content-Type-Options", "nosniff")
                    .body(into_response_body(Full::new(Bytes::from(body)))))
            }
        }
    }
}

/// Ceiling on how many bytes `read_header_prefix` buffers before giving up. Without this,
/// a client that trickles bytes forever without ever sending the terminating blank line
/// could grow the buffer without limit — the header-read timeout alone doesn't bound
/// memory, only wall-clock time, and a sufficiently patient sender could still send
/// unbounded data before the deadline fires.
const MAX_HEADER_BYTES: usize = 64 * 1024;

/// Why `read_header_prefix` gave up before seeing a complete header block. Every variant
/// is a legitimate reason to drop the connection — none is treated specially by the
/// caller today, but the distinction is worth preserving for anyone debugging this later.
#[derive(Debug)]
enum HeaderReadError {
    /// The client closed the connection (or shut down its write half) before sending a
    /// complete header block.
    ConnectionClosed,
    /// More than `MAX_HEADER_BYTES` were buffered without seeing `\r\n\r\n`.
    TooLarge,
    /// The underlying socket read failed. No caller inspects the wrapped error today —
    /// every outcome here just drops the connection — but it's kept rather than
    /// discarded so a future `log` feature (see the crate's planned Cargo feature of the
    /// same name) has the real I/O error to report instead of an opaque unit variant.
    #[allow(dead_code)]
    Io(std::io::Error),
}

/// Reads from `stream` until a complete HTTP header block (`\r\n\r\n`) has been buffered,
/// returning every byte read so far — which may include bytes past the header block
/// (request body, or a second pipelined request) if the client sent them in the same
/// read. Callers pair this with `tokio::time::timeout` to bound how long the header phase
/// itself may take; this function has no timeout of its own, only the size ceiling in
/// `MAX_HEADER_BYTES`.
async fn read_header_prefix(stream: &mut TcpStream) -> Result<Vec<u8>, HeaderReadError> {
    let mut buf = Vec::new();
    let mut chunk = [0u8; 4096];

    loop {
        let n = stream.read(&mut chunk).await.map_err(HeaderReadError::Io)?;
        if n == 0 {
            return Err(HeaderReadError::ConnectionClosed);
        }
        buf.extend_from_slice(&chunk[..n]);

        if buf.len() > MAX_HEADER_BYTES {
            return Err(HeaderReadError::TooLarge);
        }
        if buf.windows(4).any(|w| w == b"\r\n\r\n") {
            return Ok(buf);
        }
    }
}

/// Wraps an accepted `TcpStream` whose header block has already been drained into
/// `prefix` (by `read_header_prefix`, ahead of the connection being handed to hyper).
/// Reads replay `prefix` first, then fall through to the live socket — so hyper sees
/// exactly the byte stream it would have seen without the pre-read, just sourced from two
/// buffers back-to-back instead of one continuous one. Writes pass straight through.
struct PrefixedIo {
    prefix: Bytes,
    prefix_pos: usize,
    inner: TcpStream,
}

impl PrefixedIo {
    fn new(prefix: Vec<u8>, inner: TcpStream) -> Self {
        PrefixedIo {
            prefix: Bytes::from(prefix),
            prefix_pos: 0,
            inner,
        }
    }
}

impl AsyncRead for PrefixedIo {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let this = self.get_mut();
        if this.prefix_pos < this.prefix.len() {
            let remaining = &this.prefix[this.prefix_pos..];
            let n = remaining.len().min(buf.remaining());
            buf.put_slice(&remaining[..n]);
            this.prefix_pos += n;
            return Poll::Ready(Ok(()));
        }
        Pin::new(&mut this.inner).poll_read(cx, buf)
    }
}

impl AsyncWrite for PrefixedIo {
    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
        Pin::new(&mut self.get_mut().inner).poll_write(cx, buf)
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.get_mut().inner).poll_flush(cx)
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.get_mut().inner).poll_shutdown(cx)
    }
}

/// Wires an accepted connection up to the hyper HTTP/1 service.
///
/// `header_timeout` bounds only the header-read phase (`read_header_prefix`, run before
/// hyper ever sees the connection). Once a complete header block has been read, the
/// connection is handed to hyper with no further time bound — deliberately, since a
/// response body may legitimately outlive `header_timeout` by design (the live-reload SSE
/// stream is the motivating case: it stays open until a watched file changes, which may
/// be minutes or hours after the request). Wrapping the whole connection lifetime in
/// `header_timeout` — the prior implementation — silently truncated exactly that stream
/// once `header_timeout` elapsed, aborting the response mid-write after headers had
/// already been sent (the client observes this as a chunked-encoding error, not a clean
/// close). The connection-count ceiling (`Server::with_max_connections`) is what bounds
/// resource use from connections held open indefinitely, not this timeout.
async fn serve_connection(mut stream: TcpStream, server: Server, header_timeout: Duration) {
    let prefix = match timeout(header_timeout, read_header_prefix(&mut stream)).await {
        Ok(Ok(prefix)) => prefix,
        Ok(Err(_)) | Err(_) => return,
    };

    let io = TokioIo::new(PrefixedIo::new(prefix, stream));
    let svc = service_fn(move |req: Request<Incoming>| {
        let server = server.clone();
        async move {
            let method = req.method().clone();
            let path = req.uri().path().to_string();
            let if_none_match = req.headers().get("if-none-match").and_then(|v| v.to_str().ok());
            let if_modified_since = req.headers().get("if-modified-since").and_then(|v| v.to_str().ok());
            let resp = server.handle_request_async(&method, &path, if_none_match, if_modified_since).await;
            Ok::<_, Infallible>(resp)
        }
    });
    let _ = AutoBuilder::new(TokioExecutor::new()).serve_connection(io, svc).await;
}

/// A handle to a server started by `Server::run()` or `Server::run_ephemeral()`.
///
/// Dropping this handle without calling `shutdown()` leaves the server running in the
/// background for the life of the process — the same behavior `run()` always had before
/// this handle existed. Call `shutdown()` to stop accepting new connections and wait for
/// already-accepted connections to finish before returning.
pub struct ServerHandle {
    shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
    accept_task: tokio::task::JoinHandle<()>,
}

/// Default grace period `ServerHandle::shutdown()` waits for in-flight connections to
/// finish on their own before aborting whatever is left. A connection with no
/// self-imposed end — an open live-reload SSE stream, or any keep-alive connection whose
/// peer simply never closes it — would otherwise let `shutdown()` hang forever waiting
/// for it to finish naturally. Every wait in this crate has a stated upper bound;
/// shutdown is no exception.
const DEFAULT_SHUTDOWN_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);

impl ServerHandle {
    /// Stop accepting new connections and wait up to `DEFAULT_SHUTDOWN_DRAIN_TIMEOUT`
    /// (5s) for in-flight connections to finish on their own. Equivalent to
    /// `shutdown_with_timeout(DEFAULT_SHUTDOWN_DRAIN_TIMEOUT)` — see that method for what
    /// happens to connections still open once the grace period elapses.
    pub async fn shutdown(self) {
        self.shutdown_with_timeout(DEFAULT_SHUTDOWN_DRAIN_TIMEOUT).await;
    }

    /// Stop accepting new connections and wait up to `drain_timeout` for in-flight
    /// connections to finish on their own.
    ///
    /// Connections still open once `drain_timeout` elapses are aborted rather than
    /// waited on further: dropping the accept task drops its `JoinSet`, which aborts
    /// every task still tracked in it (see `tokio::task::JoinSet`'s own drop behavior) —
    /// which in turn drops each connection's socket, closing it. This is what bounds
    /// shutdown when a connection has no natural end of its own (the live-reload SSE
    /// stream is the motivating case: it stays open until a watched file changes, which
    /// may never happen before the process needs to exit).
    pub async fn shutdown_with_timeout(mut self, drain_timeout: Duration) {
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(());
        }
        if timeout(drain_timeout, &mut self.accept_task).await.is_err() {
            self.accept_task.abort();
        }
    }
}

fn into_response_body(body: Full<Bytes>) -> ResponseBody {
    ResponseBody::Buffered(body)
}

/// Finishes building a response, degrading to a generic 400 instead of panicking if any
/// header value turns out to be invalid for use as an HTTP header value.
///
/// Every header value that reaches `Response::builder()` in this module is either a
/// static string or formatted from internal, already-validated data (a byte count, an
/// mtime, a fixed method list) — none of it can actually fail today. But `.unwrap()`
/// on that assumption is exactly the kind of thing that turns "can't happen" into a
/// production panic the day someone adds a header built from new input without
/// re-deriving that guarantee. Routing every response through this one fallible path
/// means that mistake fails safe instead of panicking.
fn finish(built: Result<Response<ResponseBody>, hyper::http::Error>) -> Response<ResponseBody> {
    built.unwrap_or_else(|_| bad_request_response())
}

/// Default maximum concurrent connections, overridable via `Server::with_max_connections`.
const DEFAULT_MAX_CONNECTIONS: usize = 1024;

// `internal_error_response()` and `bad_request_response()` are the fallback responses
// `finish()` itself degrades to — every header and body here is a fixed string with no
// external input, so `.body(...)` cannot fail. They can't be routed through `finish()`
// without it degrading to itself on failure.
fn internal_error_response() -> Response<ResponseBody> {
    Response::builder()
        .status(StatusCode::INTERNAL_SERVER_ERROR)
        .header("X-Content-Type-Options", "nosniff")
        .body(into_response_body(Full::new(Bytes::from(
            "internal server error\n",
        ))))
        .unwrap()
}

fn bad_request_response() -> Response<ResponseBody> {
    Response::builder()
        .status(StatusCode::BAD_REQUEST)
        .header("X-Content-Type-Options", "nosniff")
        .body(into_response_body(Full::new(Bytes::from("bad request\n"))))
        .unwrap()
}

/// Generate an ETag for a file based on modification time and size.
///
/// Format: `"<size>-<mtime_secs>"`
fn generate_etag(metadata: &fs::Metadata) -> String {
    let size = metadata.len();
    let mtime = metadata
        .modified()
        .ok()
        .and_then(|t| t.duration_since(SystemTime::UNIX_EPOCH).ok())
        .map(|d| d.as_secs())
        .unwrap_or(0);
    format!("\"{}-{}\"", size, mtime)
}

/// Determine MIME type from file path extension.
fn mime_type_for_path(path: &Path) -> &'static str {
    path.extension()
        .and_then(|ext| ext.to_str())
        .and_then(|ext| match ext.to_lowercase().as_str() {
            "html" | "htm" => Some("text/html; charset=utf-8"),
            "css" => Some("text/css; charset=utf-8"),
            "js" => Some("application/javascript; charset=utf-8"),
            "json" => Some("application/json; charset=utf-8"),
            "svg" => Some("image/svg+xml"),
            "png" => Some("image/png"),
            "jpg" | "jpeg" => Some("image/jpeg"),
            "gif" => Some("image/gif"),
            "webp" => Some("image/webp"),
            "ico" => Some("image/x-icon"),
            "woff" => Some("font/woff"),
            "woff2" => Some("font/woff2"),
            "ttf" => Some("font/ttf"),
            "md" | "markdown" => Some("text/markdown; charset=utf-8"),
            "txt" => Some("text/plain; charset=utf-8"),
            "xml" => Some("application/xml"),
            "pdf" => Some("application/pdf"),
            "zip" => Some("application/zip"),
            _ => None,
        })
        .unwrap_or("application/octet-stream")
}

/// Check if the If-None-Match header matches the current ETag.
/// Handles both exact match and wildcard (*) comparison per RFC 9110.
fn is_etag_match(if_none_match: &str, etag: &str) -> bool {
    if if_none_match == "*" {
        return true;
    }
    if_none_match.split(',').any(|tag| tag.trim() == etag)
}

/// Check if If-Modified-Since indicates the file hasn't been modified.
/// Returns true if the file's mtime is before/equal to the If-Modified-Since timestamp.
fn is_not_modified_since(if_modified_since: &str, metadata: &fs::Metadata) -> bool {
    let file_mtime = metadata
        .modified()
        .ok()
        .and_then(|t| t.duration_since(SystemTime::UNIX_EPOCH).ok())
        .map(|d| d.as_secs())
        .unwrap_or(0);

    // Parse the If-Modified-Since header as an HTTP-date (RFC 9110 Section 5.6.7).
    // For simplicity, try to parse as a simple Unix timestamp first, then fall back to
    // a basic string comparison. A production implementation would use a proper
    // RFC 2822 / RFC 9110 date parser, but for testing we can be lenient.
    if let Ok(client_time) = if_modified_since.parse::<u64>() {
        return file_mtime <= client_time;
    }

    // Fallback: if parsing fails, be conservative and don't return 304.
    false
}

#[cfg(test)]
mod file_body_tests {
    use super::*;
    use crate::handler::FILE_CHUNK_SIZE;
    use http_body_util::BodyExt;

    // Disproves the prior implementation, which read every chunk into a `Vec` and
    // only wrapped the whole result in a single `Full` frame at the end — that
    // implementation would fail this test with `frame_count == 1` and
    // `max_frame_len == file size`, regardless of `FILE_CHUNK_SIZE`.
    #[tokio::test]
    async fn file_body_yields_multiple_bounded_chunks_not_one_buffered_frame() {
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("big.bin");
        let content = vec![7u8; FILE_CHUNK_SIZE * 3 + 12_345];
        fs::write(&path, &content).unwrap();

        let file = File::open(&path).await.unwrap();
        let mut body = FileBody::new(file);

        let mut frame_count = 0usize;
        let mut max_frame_len = 0usize;
        let mut reassembled = Vec::new();

        while let Some(frame) = body.frame().await {
            let frame = frame.unwrap();
            let data = frame.into_data().unwrap();
            frame_count += 1;
            max_frame_len = max_frame_len.max(data.len());
            reassembled.extend_from_slice(&data);
        }

        assert!(
            frame_count > 1,
            "expected the file to be delivered as multiple frames, got {frame_count}"
        );
        assert!(
            max_frame_len <= FILE_CHUNK_SIZE,
            "no single frame should exceed the chunk size ({FILE_CHUNK_SIZE}), got {max_frame_len}"
        );
        assert_eq!(reassembled, content, "reassembled chunks must match original file content exactly");
    }
}

#[cfg(test)]
mod accept_tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Mutex;

    #[test]
    fn backoff_doubles_up_to_max() {
        let mut backoff = Backoff::new();
        let mut last = backoff.next_delay();
        assert_eq!(last, ACCEPT_BACKOFF_INITIAL);

        // Double repeatedly; it must stop growing once it hits the cap rather than
        // continuing to double forever (a fixed upper bound, not an unbounded retry).
        for _ in 0..20 {
            last = backoff.next_delay();
        }
        assert_eq!(last, ACCEPT_BACKOFF_MAX);
    }

    #[test]
    fn backoff_reset_returns_to_initial_delay() {
        let mut backoff = Backoff::new();
        backoff.next_delay();
        backoff.next_delay();
        backoff.reset();
        assert_eq!(backoff.next_delay(), ACCEPT_BACKOFF_INITIAL);
    }

    /// Fails `accept()` a fixed number of times, recording the (paused, virtual)
    /// instant of each attempt, before delegating to a real listener so the caller can
    /// eventually succeed.
    struct FlakyListener {
        inner: TcpListener,
        remaining_failures: AtomicUsize,
        attempts: Mutex<Vec<tokio::time::Instant>>,
    }

    impl TcpAccept for FlakyListener {
        async fn accept(&self) -> std::io::Result<(TcpStream, SocketAddr)> {
            self.attempts.lock().unwrap().push(tokio::time::Instant::now());
            if self.remaining_failures.fetch_sub(1, Ordering::SeqCst) > 0 {
                Err(std::io::Error::other("simulated accept error"))
            } else {
                TcpAccept::accept(&self.inner).await
            }
        }
    }

    // Disproves the prior implementation, which broke out of the accept loop entirely
    // on the first `accept()` error — permanently ending the server. This test would
    // also fail against a naive `continue`-only fix (no backoff): the recorded gaps
    // between attempts would collapse to ~0 (a busy spin) instead of the expected
    // exponentially growing delays.
    #[tokio::test(start_paused = true)]
    async fn accept_loop_backs_off_between_repeated_errors_instead_of_busy_spinning() {
        let inner = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
        let addr = inner.local_addr().unwrap();

        let flaky = FlakyListener {
            inner,
            remaining_failures: AtomicUsize::new(5),
            attempts: Mutex::new(Vec::new()),
        };

        tokio::spawn(async move {
            let _ = TcpStream::connect(addr).await;
        });

        let semaphore = Arc::new(Semaphore::new(1));
        let mut backoff = Backoff::new();
        let result = accept_and_permit(&flaky, &mut backoff, &semaphore).await;
        assert!(result.is_some(), "accept should eventually succeed once the flaky listener stops failing");

        let recorded = flaky.attempts.lock().unwrap();
        assert_eq!(recorded.len(), 6, "5 failures then 1 success");

        let expected_gaps = [
            ACCEPT_BACKOFF_INITIAL,
            ACCEPT_BACKOFF_INITIAL * 2,
            ACCEPT_BACKOFF_INITIAL * 4,
            ACCEPT_BACKOFF_INITIAL * 8,
            ACCEPT_BACKOFF_INITIAL * 16,
        ];
        for (i, expected) in expected_gaps.iter().enumerate() {
            let gap = recorded[i + 1] - recorded[i];
            assert_eq!(
                gap, *expected,
                "gap between attempt {i} and {} should reflect the backoff delay, not a busy spin",
                i + 1
            );
        }
    }
}

#[cfg(test)]
mod finish_tests {
    use super::*;

    // Disproves a bare `.unwrap()` on the same builder: CR/LF is not a legal header
    // value byte (it would enable header/response splitting), so this construction is
    // guaranteed to make `.body(...)` return `Err`. Every real call site in this module
    // only ever builds header values from static strings or internally-formatted
    // numbers, so this test can't happen through normal use — it exists to prove
    // `finish()`'s fallback path actually works, not to exercise a reachable case.
    #[test]
    fn finish_degrades_to_400_on_invalid_header_value_instead_of_panicking() {
        let built = Response::builder()
            .status(StatusCode::OK)
            .header("X-Test", "invalid\r\nvalue")
            .body(into_response_body(Full::new(Bytes::new())));
        assert!(built.is_err(), "CR/LF in a header value should be rejected by the builder");

        let response = finish(built);
        assert_eq!(
            response.status(),
            StatusCode::BAD_REQUEST,
            "finish() should degrade to 400 rather than panicking on an invalid header value"
        );
    }
}

#[cfg(test)]
mod header_prefix_tests {
    use super::*;
    use tokio::io::AsyncWriteExt;

    /// Binds an ephemeral listener, connects a client to it, and returns both ends —
    /// `(server_side, client_side)` — so a test can drive `read_header_prefix` against a
    /// real socket without a full `Server`/`serve_connection` in the loop.
    async fn connected_pair() -> (TcpStream, TcpStream) {
        let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
        let addr = listener.local_addr().unwrap();
        let client = TcpStream::connect(addr).await.unwrap();
        let (server_side, _) = listener.accept().await.unwrap();
        (server_side, client)
    }

    #[tokio::test]
    async fn reads_exactly_up_to_and_including_the_terminating_blank_line() {
        let (mut server_side, mut client) = connected_pair().await;

        client
            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
            .await
            .unwrap();

        let prefix = read_header_prefix(&mut server_side).await.unwrap_or_else(|_| {
            panic!("expected a complete header block to be read");
        });

        assert_eq!(prefix, b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n");
    }

    // Disproves a version that only inspects the newest chunk for `\r\n\r\n`: writing the
    // blank line in a separate write (and thus, almost always, a separate read) after the
    // rest of the headers would make that version wait forever, since the terminator
    // never appears within a single chunk.
    #[tokio::test]
    async fn assembles_a_header_block_split_across_multiple_writes() {
        let (mut server_side, mut client) = connected_pair().await;

        client.write_all(b"GET /page HTTP/1.1\r\n").await.unwrap();
        client.write_all(b"Host: localhost\r\n").await.unwrap();
        client.write_all(b"\r\n").await.unwrap();

        let prefix = read_header_prefix(&mut server_side).await.unwrap_or_else(|_| {
            panic!("expected a complete header block to be read across multiple writes");
        });

        assert_eq!(prefix, b"GET /page HTTP/1.1\r\nHost: localhost\r\n\r\n");
    }

    // Bytes past the header block (a pipelined second request, here) must be preserved
    // verbatim in the returned prefix — `PrefixedIo` depends on this to replay them to
    // hyper untouched.
    #[tokio::test]
    async fn preserves_bytes_sent_past_the_header_block() {
        let (mut server_side, mut client) = connected_pair().await;

        let first = b"GET /a HTTP/1.1\r\nHost: localhost\r\n\r\n";
        let second = b"GET /b HTTP/1.1\r\nHost: localhost\r\n\r\n";
        let mut sent = Vec::new();
        sent.extend_from_slice(first);
        sent.extend_from_slice(second);
        client.write_all(&sent).await.unwrap();

        let prefix = read_header_prefix(&mut server_side).await.unwrap_or_else(|_| {
            panic!("expected a complete header block to be read");
        });

        assert_eq!(&prefix, &sent, "pipelined bytes past the first header block must survive intact");
    }

    #[tokio::test]
    async fn errors_with_connection_closed_when_client_disconnects_before_headers_complete() {
        let (mut server_side, client) = connected_pair().await;
        drop(client);

        match read_header_prefix(&mut server_side).await {
            Err(HeaderReadError::ConnectionClosed) => {}
            Err(_) => panic!("expected ConnectionClosed, got a different error variant"),
            Ok(_) => panic!("expected an error, got a complete header block from a closed connection"),
        }
    }

    // Disproves an unbounded buffer: without the `MAX_HEADER_BYTES` check, this would
    // hang consuming memory forever instead of erroring, since the client never sends the
    // terminating blank line.
    #[tokio::test]
    async fn errors_with_too_large_once_max_header_bytes_is_exceeded_without_a_terminator() {
        let (mut server_side, mut client) = connected_pair().await;

        let garbage = vec![b'a'; MAX_HEADER_BYTES + 1];
        client.write_all(&garbage).await.unwrap();

        match read_header_prefix(&mut server_side).await {
            Err(HeaderReadError::TooLarge) => {}
            Err(_) => panic!("expected TooLarge, got a different error variant"),
            Ok(_) => panic!("expected an error, got a complete header block from unterminated garbage"),
        }
    }

    #[tokio::test]
    async fn prefixed_io_replays_the_prefix_before_reading_from_the_live_socket() {
        let (server_side, mut client) = connected_pair().await;
        let mut io = PrefixedIo::new(b"buffered-prefix".to_vec(), server_side);

        client.write_all(b"-live-bytes").await.unwrap();

        let mut collected = Vec::new();
        let mut chunk = [0u8; 8];
        while collected.len() < b"buffered-prefix-live-bytes".len() {
            let n = io.read(&mut chunk).await.unwrap();
            assert!(n > 0, "read returned 0 before all expected bytes arrived");
            collected.extend_from_slice(&chunk[..n]);
        }

        assert_eq!(collected, b"buffered-prefix-live-bytes");
    }
}