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::{BufMut, Bytes, BytesMut};
use hyper::{Method, Response, StatusCode, Request};
use hyper::service::service_fn;
use http_body::{Body, Frame};
use http_body_util::{BodyExt, 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, ReadBuf};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::time::timeout;

use crate::error::StaticError;
use crate::handler::ResponseBody;
use crate::resolve;

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

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

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

    /// 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.
    ///
    /// # 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 server = self.clone();
        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.
    ///
    /// # 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.
    ///
    /// # 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.
    ///
    /// 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.
    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")))));
        }

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

                // HEAD must not return a body (RFC 9110); skip opening the read stream
                // entirely since we'd just discard every chunk.
                let body: ResponseBody = if *method == Method::HEAD {
                    into_response_body(Full::new(Bytes::new()))
                } else {
                    FileBody::new(file).boxed()
                };

                let content_type = mime_type_for_path(&path);
                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)))))
            }
        }
    }
}

/// Wires an accepted connection up to the hyper HTTP/1 service and drives it to
/// completion, bounded by `header_timeout`. Shared by every accept loop so the
/// framing/timeout setup is defined exactly once.
async fn serve_connection(stream: TcpStream, server: Server, header_timeout: Duration) {
    let io = TokioIo::new(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 _ = timeout(
        header_timeout,
        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<()>,
}

impl ServerHandle {
    /// Stop accepting new connections and wait for in-flight connections to finish.
    pub async fn shutdown(mut self) {
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(());
        }
        let _ = self.accept_task.await;
    }
}

fn into_response_body(body: Full<Bytes>) -> ResponseBody {
    body.map_err(|never| match never {}).boxed()
}

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

/// Chunk size for streaming file reads — 64 KB per frame.
const FILE_CHUNK_SIZE: usize = 65_536;

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

/// An `http_body::Body` that streams a `tokio::fs::File` to the client one chunk at a
/// time, instead of buffering the whole file before the response body is polled.
///
/// Each `poll_frame` call reads directly into `buf`'s spare (uninitialized) capacity via
/// `ReadBuf::uninit` and marks only the bytes the read syscall actually wrote as
/// initialized via `advance_mut` — there's no `resize`-driven zero-fill and no extra
/// copy: `split_to(n).freeze()` hands the just-filled bytes to the caller and leaves
/// `buf`'s already-reserved spare capacity in place for the next read.
struct FileBody {
    file: File,
    buf: BytesMut,
}

impl FileBody {
    fn new(file: File) -> Self {
        FileBody { file, buf: BytesMut::new() }
    }
}

impl Body for FileBody {
    type Data = Bytes;
    type Error = StaticError;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        let this = self.get_mut();

        if this.buf.capacity() - this.buf.len() < FILE_CHUNK_SIZE {
            this.buf.reserve(FILE_CHUNK_SIZE);
        }

        let mut read_buf = ReadBuf::uninit(this.buf.spare_capacity_mut());
        let file = Pin::new(&mut this.file);

        match file.poll_read(cx, &mut read_buf) {
            Poll::Ready(Ok(())) => {
                let n = read_buf.filled().len();
                if n == 0 {
                    return Poll::Ready(None);
                }
                // Safety: `poll_read` reported exactly `n` bytes filled into the spare
                // capacity we handed it via `ReadBuf::uninit`; advancing by that same
                // `n` only marks bytes the reader actually initialized.
                unsafe { this.buf.advance_mut(n) };
                let chunk = this.buf.split_to(n).freeze();
                Poll::Ready(Some(Ok(Frame::data(chunk))))
            }
            Poll::Ready(Err(e)) => Poll::Ready(Some(Err(StaticError::Io(e)))),
            Poll::Pending => Poll::Pending,
        }
    }
}

// `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 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"
        );
    }
}