mini-static 0.14.4

use std::convert::Infallible;
use std::fs;
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::{HeaderMap, Method, Request, Response, StatusCode};
use hyper::body::Incoming;
use hyper::http::response::Builder;
use hyper::service::service_fn;
use http_body_util::Full;
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::bundle;
use crate::error::StaticError;
use crate::handler::{FileBody, ResponseBody};
use crate::minify::{self, MinifyError};
use crate::minify_cache::{MinifyCache, DEFAULT_MINIFY_CACHE_CAPACITY};
use crate::reload::{self, ChangeType, 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);

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

/// 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
    }
}

/// Accept a connection and reserve it a connection-limit permit.
///
/// `backoff` retries a failed `accept()` after an exponentially growing delay (reset on
/// the next success, capped at `ACCEPT_BACKOFF_MAX`) instead of ending the accept loop,
/// so a sustained failure — the process being out of file descriptors, say — degrades
/// into periodic retries rather than a CPU-bound busy spin or a permanently dead server.
///
/// 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 Duration,
    semaphore: &Arc<Semaphore>,
) -> Option<(TcpStream, OwnedSemaphorePermit)> {
    loop {
        let stream = match listener.accept().await {
            Ok((stream, _)) => {
                *backoff = ACCEPT_BACKOFF_INITIAL;
                stream
            }
            Err(_) => {
                tokio::time::sleep(*backoff).await;
                *backoff = (*backoff * 2).min(ACCEPT_BACKOFF_MAX);
                continue;
            }
        };
        return semaphore.clone().acquire_owned().await.ok().map(|permit| (stream, permit));
    }
}

/// A predicate deciding whether a resolved file path should get an immutable cache
/// policy; see [`Server::with_immutable_assets`].
type ImmutablePredicate = Arc<dyn Fn(&Path) -> bool + Send + Sync>;

/// 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`, a couple of primitives, and an `Arc`'d
/// predicate closure. 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,
    bundle_roots: Vec<PathBuf>,
    max_connections: usize,
    live_reload: bool,
    broadcaster: Option<Broadcaster>,
    immutable_predicate: Option<ImmutablePredicate>,
    minify_cache: Option<Arc<MinifyCache>>,
    bundle_css: bool,
}

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.
    ///
    /// # 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,
            bundle_roots: Vec::new(),
            max_connections: DEFAULT_MAX_CONNECTIONS,
            live_reload: false,
            broadcaster: None,
            immutable_predicate: None,
            minify_cache: None,
            bundle_css: false,
        })
    }

    /// 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 [`Server::run_on`]) 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, the `run*` methods 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::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
    }

    /// Serve files matching `predicate` with a long-lived, immutable cache policy
    /// instead of the default `Cache-Control: no-cache`.
    ///
    /// `predicate` is evaluated against each resolved file's path; a match sends
    /// `Cache-Control: public, max-age=31536000, immutable` on that file's 200 and 304
    /// responses. This is correct only for fingerprinted assets (e.g.
    /// `main.a1b2c3.js`) where a content change always produces a new filename —
    /// caching a mutable filename indefinitely would serve stale content to every
    /// client that already has it cached.
    ///
    /// # 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"))?
    ///     .with_immutable_assets(|path| {
    ///         path.file_name()
    ///             .and_then(|name| name.to_str())
    ///             .is_some_and(|name| name.contains(".fingerprint."))
    ///     });
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_immutable_assets<F>(mut self, predicate: F) -> Self
    where
        F: Fn(&Path) -> bool + Send + Sync + 'static,
    {
        self.immutable_predicate = Some(Arc::new(predicate));
        self
    }

    /// The `Cache-Control` header value for a resolved file path: the immutable policy
    /// if `with_immutable_assets`'s predicate matches, `no-cache` otherwise.
    fn cache_control_for(&self, path: &Path) -> &'static str {
        match &self.immutable_predicate {
            Some(predicate) if predicate(path) => "public, max-age=31536000, immutable",
            _ => "no-cache",
        }
    }

    /// Enable in-memory CSS/JS minification for this server (disabled by default).
    ///
    /// A `.css`/`.js`/`.mjs` response is minified at most once per source mtime: a hit
    /// serves cached bytes, a miss reads and minifies the file and caches the result
    /// (see [`crate::minify`]). Files matching `*.min.css`/`*.min.js`
    /// are served as-is — minifying already-minified input is wasted work at best and
    /// a correctness risk at worst. If a precompressed sidecar (see
    /// [`Server::run_on`]'s docs) matches the request, its bytes are served directly
    /// and minification is skipped, since a sidecar already represents whatever a
    /// build step decided the final bytes should be. A file that fails to minify (rare
    /// malformed CSS/JS) is served unminified rather than failing the request.
    ///
    /// When `with_live_reload()` is also enabled, the cache drops an entry as soon as the
    /// same file-change event that drives live-reload arrives, instead of only noticing the
    /// change reactively on that file's next request.
    ///
    /// # 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"))?.with_minify();
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_minify(mut self) -> Self {
        self.minify_cache = Some(Arc::new(MinifyCache::new(DEFAULT_MINIFY_CACHE_CAPACITY)));
        self
    }

    /// Enable `@import` bundling for CSS files (disabled by default).
    ///
    /// Every `.css` response becomes a bundle entry point: `@import` statements are
    /// resolved and inlined (recursively) before minification. A `.css` file with no
    /// `@import` statements bundles to itself — identical output to plain `with_minify()`
    /// alone.
    ///
    /// Import resolution is bounded by the served root and any roots registered via
    /// [`Self::with_bundle_root`]: `@import "../../etc/passwd"` is rejected if the
    /// resolved path escapes all allowed roots.
    ///
    /// Requires `with_minify()` to also be enabled. If called without it, bundling is silently
    /// disabled — a no-op for v1 while we validate the feature. A future version may decouple
    /// bundling from minification.
    ///
    /// # 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"))?
    ///     .with_minify()
    ///     .with_css_bundling();
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_css_bundling(mut self) -> Self {
        self.bundle_css = true;
        self
    }

    /// Allow CSS `@import` resolution to reach files outside the served root.
    ///
    /// When CSS bundling is enabled, `@import` statements may resolve to files under `path`,
    /// in addition to files under the served root. This is useful for build pipelines where
    /// source partials live in a separate directory tree from the final served output.
    ///
    /// Files under `path` are never directly HTTP-servable: `Server::resolve` and the
    /// request-handling path never consult bundle roots, only the bundler's `@import`
    /// resolution does. This is purely an `@import` resolution boundary, not a second
    /// served root.
    ///
    /// This method is fallible and canonicalizes the path once at call time, matching
    /// `Server::new`'s canonicalize-once policy. Call it multiple times to register
    /// more than one external source tree.
    ///
    /// # Errors
    ///
    /// Returns `Err(StaticError::Io)` if the path cannot be canonicalized.
    pub fn with_bundle_root(mut self, path: &Path) -> Result<Self, StaticError> {
        let canon = path.canonicalize().map_err(StaticError::Io)?;
        self.bundle_roots.push(canon);
        Ok(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 [`Server::handle_request`] or the `run*` methods.
    ///
    /// # 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)
    }

    /// 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.
    /// Dropping the handle instead leaves the server running for the life of the process.
    ///
    /// # 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.
    ///
    /// # Precompressed Sidecars
    ///
    /// If a request's `Accept-Encoding` allows `br` or `gzip` (preferring `br`) and a
    /// sibling `<path>.br`/`<path>.gz` exists next to the resolved file, its bytes are
    /// served instead with a matching `Content-Encoding`. Every file response carries
    /// `Vary: Accept-Encoding` so intermediate caches don't serve the wrong variant to a
    /// differently-capable client.
    ///
    /// # 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());
            for bundle_root in &server.bundle_roots {
                start_watching(Arc::new(bundle_root.clone()), broadcaster.clone());
            }
            if let Some(cache) = &server.minify_cache {
                Arc::clone(cache).subscribe_to_invalidation(&broadcaster);
            }
            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 = ACCEPT_BACKOFF_INITIAL;
            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), binding an ephemeral port.
    ///
    /// Thin wrapper around [`Server::run_on`] — see it for the header-read timeout and
    /// sidecar semantics, and for what the returned [`ServerHandle`] does.
    pub async fn run(&self, header_timeout: Duration) -> Result<(u16, ServerHandle), StaticError> {
        self.run_on(([127, 0, 0, 1], 0).into(), header_timeout).await
    }

    /// Run the server on all interfaces (0.0.0.0) at `port` (0 for an ephemeral port).
    ///
    /// Useful for containerized deployments and reverse-proxy setups. Thin wrapper
    /// around [`Server::run_on`] — see it for the header-read timeout and sidecar
    /// semantics, and for what the returned [`ServerHandle`] does.
    pub async fn run_all(&self, port: u16, header_timeout: Duration) -> Result<(u16, ServerHandle), StaticError> {
        self.run_on(([0, 0, 0, 0], port).into(), header_timeout).await
    }

    /// Run the server on loopback with the default 30-second header-read timeout.
    ///
    /// The recommended entry point for tests and lightweight services that don't need a
    /// custom timeout. Thin wrapper around [`Server::run`].
    ///
    /// # 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(DEFAULT_HEADER_TIMEOUT).await
    }

    /// Produce the HTTP response for a request, streaming file bodies to the client.
    ///
    /// This is the crate's single request-handling path: the `run*` accept loop calls it,
    /// and so should any async server embedding `mini-static` as a fallback route (e.g.
    /// `mini-unified`). It 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 hands hyper one 64 KB chunk at a
    /// time as `poll_frame` is driven: memory use stays bounded to one chunk per in-flight
    /// response regardless of file size.
    ///
    /// `headers` are the request's headers; `If-None-Match` (304 on a matching ETag) and
    /// `Accept-Encoding` (precompressed sidecar selection, see [`Server::run_on`]) are the
    /// ones read today. Only `GET` and `HEAD` are allowed; anything else gets a 405 with an
    /// `Allow` header. Missing files and traversal attempts both get an identical 404, so a
    /// response never discloses whether a path exists outside the root.
    pub async fn handle_request(
        &self,
        method: &Method,
        request_path: &str,
        headers: &HeaderMap,
    ) -> Response<ResponseBody> {
        if method != Method::GET && method != Method::HEAD {
            return text(
                response(StatusCode::METHOD_NOT_ALLOWED).header("Allow", "GET, HEAD"),
                "method not allowed\n",
            );
        }

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

        // `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 path = match resolved {
            Err(_) => return internal_error_response(),
            Ok(Err(e)) => return text(response(StatusCode::NOT_FOUND), format!("{}\n", e.user_message())),
            Ok(Ok(path)) => path,
        };

        // A directory served via its `index.html` needs a trailing slash to establish the
        // correct base for the page's 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")
        {
            // `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.
            let location = format!("{}/", request_path.trim_end_matches('/'));
            return text(
                response(StatusCode::MOVED_PERMANENTLY).header("Location", location),
                "moved\n",
            );
        }

        let Ok(file) = File::open(&path).await else {
            return internal_error_response();
        };
        let Ok(metadata) = file.metadata().await else {
            return internal_error_response();
        };

        let content_type = mime_type_for_path(&path);
        // Live-reload HTML injection needs the original, uncompressed bytes to splice the
        // reload script into — never substitute a precompressed sidecar on this path.
        let html_injection = self.broadcaster.is_some() && content_type.starts_with("text/html");

        let accept_encoding = header_str(headers, "accept-encoding");
        let sidecar = if html_injection {
            None
        } else {
            select_precompressed_sidecar(&path, accept_encoding).await
        };
        let (mut file, metadata, content_encoding) = match sidecar {
            Some((sidecar_file, sidecar_metadata, encoding)) => (sidecar_file, sidecar_metadata, Some(encoding)),
            None => (file, metadata, None),
        };

        // Minification and bundling are skipped for a served precompressed sidecar (already final bytes
        // from a build step) and for the live-reload HTML injection path (needs the
        // original text to splice into).
        let change_type = ChangeType::from_path(&path);
        let should_minify = self.minify_cache.as_ref().is_some()
            && content_encoding.is_none()
            && !html_injection
            && matches!(change_type, ChangeType::Css | ChangeType::Script)
            && !minify::is_already_minified(&path);

        let etag = generate_etag(&metadata, if should_minify { "-min" } else { "" });
        let cache_control = self.cache_control_for(&path);

        if header_str(headers, "if-none-match").is_some_and(|value| is_etag_match(value, &etag)) {
            return finish(
                Response::builder()
                    .status(StatusCode::NOT_MODIFIED)
                    .header("Cache-Control", cache_control)
                    .header("Vary", "Accept-Encoding")
                    .header("ETag", etag)
                    .body(ResponseBody::Buffered(Full::new(Bytes::new()))),
            );
        }

        // `Some` when the served representation differs from the file's raw bytes and had
        // to be built in memory; `None` means stream the open file as-is. Computed before
        // the HEAD check below because RFC 9110 requires a HEAD response's headers —
        // `Content-Length` included — to match what a GET would send, even though the body
        // itself is dropped.
        let transformed: Option<Bytes> = if should_minify {
            let cache = self.minify_cache.as_ref().unwrap();
            let bundle_enabled = self.bundle_css && matches!(change_type, ChangeType::Css);

            if bundle_enabled {
                let mut allowed_roots = Vec::with_capacity(1 + self.bundle_roots.len());
                allowed_roots.push(self.root_canon.clone());
                allowed_roots.extend(self.bundle_roots.iter().cloned());
                let entry_path = path.clone();
                match cache
                    .get_or_bundle_css(&path, bundle::bundle_and_minify_css(&allowed_roots, &entry_path))
                    .await
                {
                    Ok(bytes) => Some(bytes),
                    Err(e) => {
                        log_bundle_error(&e);
                        let mut raw = Vec::new();
                        if file.read_to_end(&mut raw).await.is_err() {
                            return internal_error_response();
                        }
                        Some(Bytes::from(raw))
                    }
                }
            } else {
                let mtime = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH);
                match minified_or_raw(cache, &path, mtime, change_type, &mut file).await {
                    Ok(bytes) => Some(bytes),
                    Err(_) => return internal_error_response(),
                }
            }
        } else if html_injection {
            let mut html = Vec::with_capacity(metadata.len() as usize);
            if file.read_to_end(&mut html).await.is_err() {
                return internal_error_response();
            }
            reload::inject_reload_script(&mut html);
            Some(Bytes::from(html))
        } else {
            None
        };

        let file_size = transformed.as_ref().map_or(metadata.len(), |bytes| bytes.len() as u64);

        // HEAD must not return a body (RFC 9110).
        let body = if *method == Method::HEAD {
            ResponseBody::Buffered(Full::new(Bytes::new()))
        } else {
            match transformed {
                Some(bytes) => ResponseBody::Buffered(Full::new(bytes)),
                None => ResponseBody::Streamed(FileBody::new(file)),
            }
        };

        let mut builder = response(StatusCode::OK)
            .header("Content-Type", content_type)
            .header("Content-Length", file_size.to_string())
            .header("Cache-Control", cache_control)
            .header("Vary", "Accept-Encoding")
            .header("ETag", etag);
        if let Some(encoding) = content_encoding {
            builder = builder.header("Content-Encoding", encoding);
        }
        finish(builder.body(body))
    }
}

/// Minified bytes for `path`, falling back to the file's raw bytes if the source is
/// malformed — a rare but real possibility (a hand-edited file, a build tool's bug). A
/// broken minify step shouldn't take down an otherwise-servable file.
///
/// # Errors
///
/// Returns `Err` only if the fallback read of `file` itself fails.
async fn minified_or_raw(
    cache: &MinifyCache,
    path: &Path,
    mtime: SystemTime,
    change_type: ChangeType,
    file: &mut File,
) -> Result<Bytes, MinifyError> {
    if let Ok(minified) = cache.get_or_minify(path, mtime, change_type, minify::minify).await {
        return Ok(minified);
    }
    let mut raw = Vec::new();
    file.read_to_end(&mut raw).await.map_err(MinifyError::Io)?;
    Ok(Bytes::from(raw))
}

/// 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. Kept rather than discarded so a future `log`
    /// feature 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);
        }
        // Only the tail can hold a terminator this read completed: the `n` new bytes plus
        // the 3 before them. Rescanning the whole buffer every time would make the header
        // read quadratic in the bytes received.
        let scan_from = buf.len().saturating_sub(n + 3);
        if buf[scan_from..].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 resp = server
                .handle_request(req.method(), req.uri().path(), req.headers())
                .await;
            Ok::<_, Infallible>(resp)
        }
    });
    let _ = AutoBuilder::new(TokioExecutor::new()).serve_connection(io, svc).await;
}

/// Default header-read timeout used by [`Server::run_ephemeral`].
const DEFAULT_HEADER_TIMEOUT: Duration = Duration::from_secs(30);

/// 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);

/// A handle to a server started by one of the `Server::run*` methods.
///
/// Dropping this handle without calling `shutdown()` leaves the server running in the
/// background for the life of the process. 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<()>,
}

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();
        }
    }
}

/// Read a request header as a `&str`, or `None` if it's absent or not valid ASCII.
fn header_str<'h>(headers: &'h HeaderMap, name: &str) -> Option<&'h str> {
    headers.get(name).and_then(|value| value.to_str().ok())
}

/// Start a response carrying the baseline security header every response in this crate
/// sends. The 304 path is the one exception and builds its own — a 304 repeats only the
/// caching validators, not the full header set.
fn response(status: StatusCode) -> Builder {
    Response::builder()
        .status(status)
        .header("X-Content-Type-Options", "nosniff")
}

/// Finish `builder` with a plain-text body. `&'static str` bodies borrow rather than
/// allocate; `String` bodies (the 404 message) are moved in.
fn text(builder: Builder, body: impl Into<Bytes>) -> Response<ResponseBody> {
    finish(builder.body(ResponseBody::Buffered(Full::new(body.into()))))
}

/// 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())
}

// `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(StatusCode::INTERNAL_SERVER_ERROR)
        .body(ResponseBody::Buffered(Full::new(Bytes::from_static(
            b"internal server error\n",
        ))))
        .unwrap()
}

fn bad_request_response() -> Response<ResponseBody> {
    response(StatusCode::BAD_REQUEST)
        .body(ResponseBody::Buffered(Full::new(Bytes::from_static(b"bad request\n"))))
        .unwrap()
}

/// Log a bundle error at error level. Traversal/cycle/depth/count violations are
/// misconfiguration and should be visible in logs even though the response falls back
/// to raw-file serve (never a 500). Parse/IO failures are rarer and worth logging too.
fn log_bundle_error(error: &bundle::BundleError) {
    eprintln!("bundle error: {:?}", error);
}

/// `Content-Encoding` name and sidecar file extension for each supported precompressed
/// variant, in preference order — brotli wins when a client accepts both and both
/// sidecars exist.
const SIDECAR_ENCODINGS: [(&str, &str); 2] = [("br", ".br"), ("gzip", ".gz")];

/// Whether `accept_encoding` allows `encoding`.
///
/// Matches by substring rather than parsing `q`-value weights or the `identity`/`*`
/// directives — a lighter-weight negotiation than a general HTTP client would need,
/// sufficient for deciding between two static sidecar files.
fn accepts_encoding(accept_encoding: Option<&str>, encoding: &str) -> bool {
    accept_encoding.is_some_and(|header| header.contains(encoding))
}

/// Look for a precompressed sidecar (`<path>.br` / `<path>.gz`) matching the client's
/// `Accept-Encoding`, and return its open file, metadata, and encoding name if found.
///
/// `path` must already be the fully resolved, canonicalized path `resolve()` produced.
/// The sidecar path is built by appending an extension to it — never by re-resolving a
/// modified request path — so this lookup can't become a second traversal surface: any
/// path this function reads is provably a sibling of a path `resolve()` already cleared.
async fn select_precompressed_sidecar(
    path: &Path,
    accept_encoding: Option<&str>,
) -> Option<(File, fs::Metadata, &'static str)> {
    for (encoding, ext) in SIDECAR_ENCODINGS {
        if !accepts_encoding(accept_encoding, encoding) {
            continue;
        }
        let mut sidecar = path.as_os_str().to_os_string();
        sidecar.push(ext);
        let sidecar_path = PathBuf::from(sidecar);

        // Tripwire for the traversal boundary: a sidecar path built by appending a suffix
        // must stay in the same directory as `path` (which `resolve()` already proved is
        // inside root). `ext` is always one of the two static literals in
        // `SIDECAR_ENCODINGS`, never derived from request input, so this can only fire if
        // a future change starts deriving `sidecar` some other way.
        debug_assert_eq!(
            sidecar_path.parent(),
            path.parent(),
            "sidecar path must stay in the same directory as the already-resolved path"
        );

        if let Ok(sidecar_file) = File::open(&sidecar_path).await {
            if let Ok(sidecar_metadata) = sidecar_file.metadata().await {
                return Some((sidecar_file, sidecar_metadata, encoding));
            }
        }
    }
    None
}

/// Generate an ETag for a file based on modification time and size.
///
/// `variant_suffix` distinguishes a served representation that differs from the raw
/// source bytes without needing to read/transform the file just to compute a tag: pass
/// `"-min"` when the response will be minified, `""` otherwise. Without this, turning
/// `with_minify()` on for an already-served, already-cached file wouldn't change its
/// ETag at all (the source file's size and mtime are unchanged) — a client that cached
/// the unminified `200` would keep matching on `If-None-Match` and get `304`s forever,
/// never seeing the now-minified bytes until the source file's mtime actually changes.
///
/// Format: `"<size>-<mtime_secs><variant_suffix>"`
fn generate_etag(metadata: &fs::Metadata, variant_suffix: &str) -> String {
    let mtime = metadata
        .modified()
        .ok()
        .and_then(|t| t.duration_since(SystemTime::UNIX_EPOCH).ok())
        .map(|d| d.as_secs())
        .unwrap_or(0);
    format!("\"{}-{}{}\"", metadata.len(), mtime, variant_suffix)
}

/// Determine MIME type from file path extension.
fn mime_type_for_path(path: &Path) -> &'static str {
    let ext = path
        .extension()
        .and_then(|ext| ext.to_str())
        .unwrap_or_default()
        .to_lowercase();

    match ext.as_str() {
        "html" | "htm" => "text/html; charset=utf-8",
        "css" => "text/css; charset=utf-8",
        "js" => "application/javascript; charset=utf-8",
        "json" => "application/json; charset=utf-8",
        "svg" => "image/svg+xml",
        "png" => "image/png",
        "jpg" | "jpeg" => "image/jpeg",
        "gif" => "image/gif",
        "webp" => "image/webp",
        "ico" => "image/x-icon",
        "woff" => "font/woff",
        "woff2" => "font/woff2",
        "ttf" => "font/ttf",
        "md" | "markdown" => "text/markdown; charset=utf-8",
        "txt" => "text/plain; charset=utf-8",
        "xml" => "application/xml",
        "pdf" => "application/pdf",
        "zip" => "application/zip",
        _ => "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)
}

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

    // `select_precompressed_sidecar` only ever appends a static extension literal
    // (".br"/".gz") to the `path` it's given — it never re-joins against `root` or
    // re-parses a request-path string, so it structurally cannot become a second
    // traversal surface the way re-running `resolve()` on modified input could. This
    // test locks that in by construction: the sidecar it finds must live in exactly
    // the same directory as the resolved file, for every encoding preference branch.
    #[tokio::test]
    async fn sidecar_never_leaves_the_resolved_files_directory() {
        let root = tempfile::TempDir::new().unwrap();
        let sub = root.path().join("assets");
        fs::create_dir(&sub).unwrap();
        let resolved = sub.join("app.js");
        fs::write(&resolved, b"plain").unwrap();
        fs::write(sub.join("app.js.br"), b"brotli-bytes").unwrap();
        fs::write(sub.join("app.js.gz"), b"gzip-bytes").unwrap();

        let (_, _, encoding) = select_precompressed_sidecar(&resolved, Some("br, gzip"))
            .await
            .expect("both sidecars present, br should be preferred");
        assert_eq!(encoding, "br", "br must be preferred over gzip when both are accepted");

        let (_, _, encoding) = select_precompressed_sidecar(&resolved, Some("gzip"))
            .await
            .expect("gzip sidecar present");
        assert_eq!(encoding, "gzip");

        assert!(
            select_precompressed_sidecar(&resolved, None).await.is_none(),
            "no Accept-Encoding header should never select a sidecar"
        );
    }

    #[test]
    fn accepts_encoding_matches_only_listed_directives() {
        assert!(!accepts_encoding(None, "br"));
        assert!(!accepts_encoding(Some("identity"), "br"));
        assert!(!accepts_encoding(Some("identity"), "gzip"));
        assert!(accepts_encoding(Some("gzip, br"), "br"));
        assert!(accepts_encoding(Some("gzip"), "gzip"));
        assert!(!accepts_encoding(Some("gzip"), "br"));
    }
}

#[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;

    /// 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 = ACCEPT_BACKOFF_INITIAL;
        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
            );
        }

        // The delay must stop doubling at the cap rather than growing without bound.
        let mut capped = ACCEPT_BACKOFF_MAX;
        capped = (capped * 2).min(ACCEPT_BACKOFF_MAX);
        assert_eq!(capped, ACCEPT_BACKOFF_MAX);
    }

    // A successful accept must clear the accumulated delay, so an isolated error later
    // on doesn't inherit a second-long wait from an unrelated earlier failure.
    #[tokio::test(start_paused = true)]
    async fn a_successful_accept_resets_the_backoff() {
        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(3),
            attempts: Mutex::new(Vec::new()),
        };
        tokio::spawn(async move {
            let _ = TcpStream::connect(addr).await;
        });

        let semaphore = Arc::new(Semaphore::new(1));
        let mut backoff = ACCEPT_BACKOFF_INITIAL * 32;
        accept_and_permit(&flaky, &mut backoff, &semaphore).await;

        assert_eq!(
            backoff, ACCEPT_BACKOFF_INITIAL,
            "the delay must return to its initial value once an accept succeeds"
        );
    }
}

#[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(ResponseBody::Buffered(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. Also pins the tail-only scan in
    // `read_header_prefix` — a terminator straddling two reads must still be seen.
    #[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\nHost: localhost\r").await.unwrap();
        client.write_all(b"\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 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");
    }
}