mini_serve/

app.rs

use std::future::Future;
use std::net::SocketAddr;
use std::io::Write as _;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use hyper::header::{HeaderName, HeaderValue};
use hyper::body::Body as HttpBody;
use hyper::body::Bytes;
use hyper::body::Incoming;
use hyper::server::conn::http1;
use hyper::{Method, Request, Response, StatusCode};
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use hyper_util::rt::TokioTimer;
use http_body_util::combinators::BoxBody;
use http_body_util::{BodyExt, Empty, Full};
use tokio::net::{TcpListener, TcpStream};

use crate::body::{MaxBodySize, DEFAULT_MAX_BODY_SIZE};
use crate::cors::CorsConfig;
use crate::error::ServeError;
use crate::handler::{Handler, Middleware, OnUpgrade, ResponseBody};
use crate::router::{QueryParams, Router};
use crate::state::State;

/// The transport-layer peer address a request arrived from. Inserted into
/// request extensions by [`App::route_with_peer`] — retrieve it with
/// `req.extensions().get::<PeerAddr>()`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PeerAddr(pub SocketAddr);

const MAX_PATH_LEN: usize = 8_192;
const MAX_QUERY_LEN: usize = 4_096;
const DEFAULT_HEADER_READ_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_MAX_CONNECTIONS: usize = 1024;

/// Headers the connection layer owns, refused as fixed values by
/// [`RouteBuilder::with_response_header`]. A caller-supplied `Content-Length` would
/// contradict the body hyper is about to write; the other two describe framing this
/// crate does not choose.
const CONNECTION_OWNED_HEADERS: [HeaderName; 3] = [
	hyper::header::CONTENT_LENGTH,
	hyper::header::CONNECTION,
	hyper::header::TRANSFER_ENCODING,
];
/// Grace period `serve_loop` allows in-flight connections to finish after the shutdown
/// signal, before whatever is left is aborted.
///
/// The drain used to be unbounded — `join_next()` until the set emptied — so a single
/// wedged handler, or a client holding a long-lived streaming response, kept the process
/// alive forever and made `bind()` un-returnable. A2 says every wait states its ceiling;
/// shutdown is no exception. Matches `mini-static`'s constant of the same name and value.
const DEFAULT_SHUTDOWN_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
/// How long a transport has to turn an accepted `TcpStream` into a usable connection.
///
/// Ten seconds, carried over from the TLS handshake timeout this replaces. It now bounds
/// every transport rather than one: for the identity transport it can never fire, and for
/// anything that does real work it is the difference between a slow peer and a held
/// connection slot.
const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
/// How long the connection task waits for an upgrade to complete after the `101` has been
/// written, before giving up and releasing the connection's permit.
///
/// A2: this wait needs a ceiling like any other. A handler can return `101` to a client
/// that never actually asked to upgrade, in which case hyper's upgrade future never
/// resolves — and without a bound the task would hold its permit forever, which is a leak
/// reachable from the network by the very mechanism meant to prevent one.
const UPGRADE_HANDOFF_TIMEOUT: Duration = Duration::from_secs(10);
const ACCEPT_BACKOFF_INITIAL: Duration = Duration::from_millis(10);
const ACCEPT_BACKOFF_MAX: Duration = Duration::from_secs(1);

#[cfg(test)]
thread_local! {
	static ERROR_LOG: std::cell::RefCell<Vec<(u16, String)>> = const { std::cell::RefCell::new(Vec::new()) };
}

#[cfg(test)]
fn capture_error(code: u16, message: String) {
	ERROR_LOG.with(|log| log.borrow_mut().push((code, message)));
}

#[cfg(test)]
fn take_error_log() -> Vec<(u16, String)> {
	ERROR_LOG.with(|log| log.borrow_mut().drain(..).collect())
}

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

pub type ErrorHandler =
	Arc<dyn Fn(StatusCode, &str) -> Response<ResponseBody> + Send + Sync>;

/// Where request and failure lines go.
///
/// `App` is shared behind an `Arc` and its connections run on independent tasks, so the
/// sink is shared rather than duplicated; the mutex serializes writes so two tasks
/// finishing at once cannot interleave mid-line and produce an entry belonging to
/// neither.
pub(crate) type LogSink = Arc<Mutex<Box<dyn std::io::Write + Send>>>;

/// An HTTP server application with typed state, routing, and TLS support.
///
/// `App<S>` serves requests by routing them to handlers based on method and path.
/// All handlers share access to a single `S` value (the app state), cloned as an `Arc`
/// per request for zero-allocation sharing.
///
/// # Example
///
/// ```ignore
/// use mini_serve::App;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
///     let app = App::new(());
///     app.bind("127.0.0.1:8080".parse()?).await?;
///     Ok(())
/// }
/// ```
///
/// # Features
///
/// - **Routing**: Register handlers for (Method, Path) pairs with path parameters (`/users/:id`).
/// - **State sharing**: All handlers receive `Arc<S>` to the app state.
/// - **Request extraction**: Parse bodies, extract path/query params, and build responses.
/// - **CORS**: Optional cross-origin request handling with preflight validation.
/// - **TLS**: Serve over HTTPS when the `tls` feature is enabled.
/// - **Graceful shutdown**: Drain in-flight requests, bounded by a grace period, before
///   exiting.
pub struct App<S> {
	state:               Arc<S>,
	log:                 Option<LogSink>,
	extra_headers:       Arc<Vec<(HeaderName, HeaderValue)>>,
	router:              Arc<Router<S>>,
	max_body_size:       usize,
	pub(crate) upgrades_enabled:    bool,
	pub(crate) header_read_timeout: Duration,
	pub(crate) max_connections:     usize,
	pub(crate) connect_timeout:     Duration,
	error_handler:       ErrorHandler,
	cors_config:         Option<CorsConfig>,
}

/// Report a connection task that ended by panicking.
///
/// The join result was previously dropped on the floor, so a panicking handler showed up
/// as a dropped connection with no record anywhere — the client could not tell it from a
/// network fault and the operator could not tell it happened at all.
pub(crate) fn report_if_panicked(sink: &Option<LogSink>, joined: Result<(), tokio::task::JoinError>) {
	let Err(e) = joined else {
		return;
	};
	if e.is_panic() {
		log_line(sink, format_args!("connection task panicked: {e}"));
	}
}

/// Write `line` to `sink`, if there is one.
///
/// A poisoned mutex and a failed write are both ignored: neither is a reason to fail a
/// request that was otherwise served, and a logger that can take down the server is
/// worse than one that occasionally drops a line.
pub(crate) fn log_line(sink: &Option<LogSink>, line: std::fmt::Arguments<'_>) {
	let Some(sink) = sink else {
		return;
	};
	if let Ok(mut out) = sink.lock() {
		let _ = writeln!(out, "{line}");
		let _ = out.flush();
	}
}

/// Parse a non-empty query string into its decoded key/value pairs.
///
/// A repeated key resolves to its last value, which is what `?a=1&a=2` means here.
/// Callers guard on emptiness — see the call site in `route_inner`.
fn parse_query(query: &str) -> QueryParams {
	let mut map = std::collections::HashMap::new();
	for pair in query.split('&').filter(|s| !s.is_empty()) {
		if let Some((key, value)) = pair.split_once('=') {
			let key = decode_query_component(key);
			let value = decode_query_component(value);
			map.insert(key, value);
		} else {
			let pair = decode_query_component(pair);
			map.insert(pair, String::new());
		}
	}
	QueryParams(map)
}

fn decode_query_component(s: &str) -> String {
	let with_spaces = s.replace('+', " ");
	percent_encoding::percent_decode_str(&with_spaces)
		.decode_utf8_lossy()
		.into_owned()
}

fn error_response(status: StatusCode, message: &str) -> Response<ResponseBody> {
	#[cfg(test)]
	capture_error(status.as_u16(), message.to_string());

	let client_message = if status.is_server_error() {
		"internal server error"
	} else {
		message
	};

	let body = serde_json::json!({ "message": client_message });
	let json = serde_json::to_string(&body)
		.unwrap_or_else(|_| r#"{"message":"internal server error"}"#.to_string());
	let mut resp = Response::new(BoxBody::new(
		Full::new(Bytes::from(json)).map_err(|never: std::convert::Infallible| match never {}),
	));
	*resp.status_mut() = status;
	// `insert` rather than the builder's `header()`, which appends and therefore scans
	// the map first. Same reasoning and the same scope limit as `response::json` — see
	// the note there before copying this anywhere else.
	resp.headers_mut().insert(
		hyper::header::CONTENT_TYPE,
		HeaderValue::from_static("application/json"),
	);
	resp
}

fn default_error_handler() -> ErrorHandler {
	Arc::new(error_response)
}

/// Address for ephemeral test/dev binds. Deliberately loopback-only —
/// unlike a production bind, callers never choose this address, so it must
/// not expose the listener beyond the local machine.
fn ephemeral_bind_addr() -> SocketAddr {
	(std::net::Ipv4Addr::LOCALHOST, 0).into()
}

impl<S: Send + Sync + 'static> App<S> {
	/// Create a new app with shared state.
	///
	/// The state is wrapped in an `Arc` and shared with every request handler
	/// as `State::from_arc()`. Route registration is done via `RouteBuilder`.
	pub fn new(state: S) -> Self {
		App {
			state:              Arc::new(state),
			router:             Arc::new(Router::new()),
			max_body_size:       DEFAULT_MAX_BODY_SIZE,
			header_read_timeout: DEFAULT_HEADER_READ_TIMEOUT,
			upgrades_enabled:    false,
			log:                 None,
			extra_headers:       Arc::new(Vec::new()),
			max_connections:     DEFAULT_MAX_CONNECTIONS,
			connect_timeout:     DEFAULT_CONNECT_TIMEOUT,
			error_handler:       default_error_handler(),
			cors_config:         None,
		}
	}

	/// Get an `Arc` to the app state.
	///
	/// Useful for spawning background tasks or accessing state outside the
	/// request-response loop.
	pub fn state_arc(&self) -> Arc<S> {
		Arc::clone(&self.state)
	}

	pub async fn route(&self, req: Request<Incoming>) -> Response<ResponseBody> {
		self.route_with(req, None).await
	}

	/// Route a request, additionally exposing the transport-layer peer
	/// address to handlers and middleware via `req.extensions().get::<PeerAddr>()`.
	///
	/// Used internally by the accept loops, which always know the peer
	/// address; exposed publicly for callers embedding `App` into their own
	/// connection-handling code with a real peer address available.
	pub async fn route_with_peer(&self, req: Request<Incoming>, peer: SocketAddr) -> Response<ResponseBody> {
		self.route_with(req, Some(peer)).await
	}

	async fn route_with(&self, req: Request<Incoming>, peer: Option<SocketAddr>) -> Response<ResponseBody> {
		let method = req.method().clone();
		// Read before `req` is consumed: every exit below needs it, not just the one
		// that used to apply CORS. `get` returns the *first* `Origin` when a request
		// carries several — this is the only read of that header, so the preflight
		// branch and `finalize` cannot end up reflecting different ones.
		let req_origin = req
			.headers()
			.get(hyper::header::ORIGIN)
			.and_then(|v| v.to_str().ok())
			.map(|s| s.to_string());

		let mut resp = self.route_inner(req, peer, req_origin.as_deref()).await;
		self.finalize(&mut resp, req_origin.as_deref());

		// Strip the body uniformly across every branch above (success,
		// handler error, 404, 405) rather than only the success path.
		// hyper's HTTP/1 server already refuses to write a body to the wire
		// for HEAD regardless of what we return here, so this doesn't change
		// observable behavior — it just avoids handing hyper a body (e.g. a
		// freshly-built error JSON payload) that would only be discarded.
		if method == Method::HEAD {
			let (mut parts, body) = resp.into_parts();
			// RFC 9110 §9.3.2: HEAD must report the `Content-Length` its GET would.
			// Handlers that set the header themselves (`json`, for one) already carry
			// it in `parts`; handlers that returned a body and let hyper derive the
			// length lose it here, because the body carrying that length is exactly
			// what is being dropped. Take the length from the body before discarding
			// it, so `curl -I` and any CDN sizing a resource get the real answer
			// instead of nothing.
			if !parts.headers.contains_key(hyper::header::CONTENT_LENGTH) {
				if let Some(len) = HttpBody::size_hint(&body).exact() {
					parts.headers.insert(hyper::header::CONTENT_LENGTH, len.into());
				}
			}
			Response::from_parts(parts, BoxBody::new(Empty::new().map_err(|never: std::convert::Infallible| match never {})))
		} else {
			resp
		}
	}

	/// The single point every response passes through before reaching hyper.
	///
	/// CORS headers used to be applied in exactly one of six exit paths — the branch
	/// where a handler returned `Ok`. A cross-origin request that 404'd, 405'd, or
	/// errored therefore came back *without* them, so the browser reported an opaque
	/// CORS failure and the caller never learned the real status. Applying here fixes
	/// that by construction: a new exit path cannot forget what it never had to
	/// remember.
	///
	/// `apply_to_response` inserts rather than appends, so a preflight response that
	/// already carries these headers is unchanged by passing through again.
	fn finalize(&self, resp: &mut Response<ResponseBody>, req_origin: Option<&str>) {
		if let Some(cfg) = &self.cors_config {
			cfg.apply_to_response(resp, req_origin);
		}

		// Content sniffing turns a mislabelled response into a script-execution vector,
		// and an API serving user-supplied content has no way to know it is safe.
		let headers = resp.headers_mut();
		headers
			.entry(hyper::header::X_CONTENT_TYPE_OPTIONS)
			.or_insert(HeaderValue::from_static("nosniff"));

		for (name, value) in self.extra_headers.iter() {
			headers.entry(name).or_insert(value.clone());
		}
	}

	/// `req_origin` is captured once by [`route_with`] and threaded here rather than
	/// re-read. Two independent lookups of an attacker-controlled header are two chances
	/// to disagree, and the preflight branch below and `finalize` reflecting different
	/// values would be a header-smuggling primitive. One read, one value, both users.
	async fn route_inner(
		&self,
		req: Request<Incoming>,
		peer: Option<SocketAddr>,
		req_origin: Option<&str>,
	) -> Response<ResponseBody> {
		// RFC 9112 §3.2: a server MUST answer 400 to an HTTP/1.1 request with no `Host`.
		// hyper 1.11 serves these, so the check lives here — a request with no authority
		// is ambiguous to anything downstream doing name-based routing, and this crate
		// would otherwise pass that ambiguity along. Absolute-form targets carry the
		// authority in the URI, which satisfies the requirement, and HTTP/1.0 never had
		// it, so both are exempt.
		if req.version() == hyper::Version::HTTP_11
			&& req.uri().authority().is_none()
			&& !req.headers().contains_key(hyper::header::HOST)
		{
			return (self.error_handler)(StatusCode::BAD_REQUEST, "missing host header");
		}
		if req.uri().path().len() > MAX_PATH_LEN {
			return (self.error_handler)(StatusCode::BAD_REQUEST, "path too long");
		}
		if req.uri().query().map(|q| q.len()).unwrap_or(0) > MAX_QUERY_LEN {
			return (self.error_handler)(StatusCode::BAD_REQUEST, "query string too long");
		}

		let method = req.method().clone();
		let path = req.uri().path().to_string();
		let state = State::from_arc(Arc::clone(&self.state));
		// A request with no query string built a HashMap to hold nothing. Guarded on
		// *emptiness*, not absence: `?` alone parses as `Some("")`, so `is_none()`
		// would miss exactly the case this is here to catch. The `MAX_QUERY_LEN`
		// bound above stays where it is and stays unconditional — folding it in here
		// would let an oversized query skip its limit whenever this guard
		// short-circuits.
		let query = req.uri().query().unwrap_or("");
		let query_params = if query.is_empty() {
			QueryParams::default()
		} else {
			parse_query(query)
		};

		// Handle CORS preflight only for existing routes
		if method == Method::OPTIONS && req_origin.is_some() {
			if let Some(cfg) = &self.cors_config {
				if self.router.path_exists(&path) {
					let requested_headers = req
						.headers()
						.get("access-control-request-headers")
						.and_then(|v| v.to_str().ok());
					let allowed = self.allowed_methods_with_head(&path);
					return cfg.preflight_response(req_origin, requested_headers, &allowed);
				}
			}
		}

		let method_to_match = if method == Method::HEAD {
			Method::GET
		} else {
			method.clone()
		};

		match self.router.match_route(&method_to_match, &path) {
			Some((handler, params)) => {
				let mut req = req;
				// Each insert boxes its value and hashes a `TypeId`. A route with no
				// params and a request with no query string pay both for nothing —
				// `/health` was inserting an empty `PathParams` on every request.
				// `query_params()` and `path_params()` read absence as emptiness, so
				// no consumer can tell these were skipped.
				if !query_params.0.is_empty() {
					req.extensions_mut().insert(query_params);
				}
				if !params.0.is_empty() {
					req.extensions_mut().insert(params);
				}
				// `json_body` falls back to `DEFAULT_MAX_BODY_SIZE` when this is absent,
				// so an app on the default gets identical behaviour without paying for a
				// boxed extension on every request.
				if self.max_body_size != DEFAULT_MAX_BODY_SIZE {
					req.extensions_mut().insert(MaxBodySize(self.max_body_size));
				}
				if let Some(peer) = peer {
					req.extensions_mut().insert(PeerAddr(peer));
				}
				match handler(req, state).await {
					Ok(resp) => resp,
					Err(e) => {
						let status = StatusCode::from_u16(e.code)
							.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
						// A 5xx body is sanitized to "internal server error" on purpose —
						// handler messages can carry internals a client must not see. The
						// real message goes here instead of being discarded with it, which
						// is what previously left an operator nothing to debug from. 4xx
						// messages are already sent to the client, so they are not repeated.
						if status.is_server_error() {
							log_line(&self.log, format_args!("{status} {}: {}", path, e.message));
						}
						(self.error_handler)(status, &e.message)
					}
				}
			}
			None => {
				let allowed = self.allowed_methods_with_head(&path);
				if !allowed.is_empty() {
					let mut method_strs: Vec<&str> = allowed.iter().map(|m| m.as_str()).collect();
					method_strs.sort();
					method_strs.dedup();
					let allow_header = method_strs.join(", ");
					let mut resp = (self.error_handler)(StatusCode::METHOD_NOT_ALLOWED, "method not allowed");
					if let Ok(val) = allow_header.parse() {
						resp.headers_mut().insert("allow", val);
					}
					resp
				} else {
					(self.error_handler)(StatusCode::NOT_FOUND, "not found")
				}
			}
		}
	}

	/// Every method the router accepts for `path`, plus `HEAD` whenever
	/// `GET` is one of them (hyper's HTTP/1 server answers `HEAD` by running
	/// the `GET` handler and discarding the body — see `route()` above —
	/// so `HEAD` is always implicitly valid alongside `GET`). Shared by the
	/// CORS preflight branch and the plain 405 branch so the two can never
	/// disagree about what a path actually accepts.
	fn allowed_methods_with_head(&self, path: &str) -> Vec<Method> {
		let mut allowed = self.router.allowed_methods(path);
		if allowed.contains(&Method::GET) {
			allowed.push(Method::HEAD);
		}
		allowed
	}

	/// Bind to an ephemeral port and serve in the background.
	///
	/// Returns the assigned port number. The server runs in a spawned task
	/// and serves until the process exits. For graceful shutdown, use `run()`.
	/// Binds to 127.0.0.1 only—safe for development and testing.
	pub async fn bind_ephemeral(self) -> Result<u16, ServeError> {
		let listener = TcpListener::bind(ephemeral_bind_addr())
			.await
			.map_err(|e| ServeError::new(500, format!("failed to bind to ephemeral port: {e}")))?;
		let port = listener
			.local_addr()
			.map_err(|e| ServeError::new(500, format!("failed to get assigned port: {e}")))?
			.port();
		let app = Arc::new(self);
		tokio::spawn(async move {
			serve_loop(listener, app, std::future::pending(), plain_connect).await;
		});
		Ok(port)
	}


	/// Serve `listener` until `shutdown` resolves, then drain in-flight
	/// connections and return. The production entry point for callers that
	/// want control over the shutdown trigger (tests, custom signals); see
	/// [`App::bind`] for the OS-signal convenience wrapper.
	pub async fn run<F>(self, listener: TcpListener, shutdown: F) -> Result<(), ServeError>
	where
		F: Future<Output = ()> + Send + 'static,
	{
		self.run_with_transport(listener, shutdown, plain_connect).await
	}

	/// Serve `listener` over a caller-supplied transport.
	///
	/// `connect` turns each accepted `TcpStream` into whatever the connection should
	/// actually speak — TLS, or anything else negotiated before HTTP begins. Returning
	/// `None` rejects the connection. [`App::run`] is this method with an identity
	/// transport.
	///
	/// This is the seam TLS plugs into: `mini-tls` is a function of this shape, which is
	/// why it needs no dependency on this crate. The transport is structural, not a trait
	/// with one implementor.
	///
	/// `connect` runs in the connection's own task, holding its semaphore permit, so a
	/// connection being negotiated still counts against
	/// [`RouteBuilder::with_max_connections`].
	///
	/// There is deliberately no `bind_with_transport` or ephemeral variant: an extension
	/// binds its own listener, which keeps this seam to one method. For SIGINT/SIGTERM
	/// handling, pass [`shutdown_signal`].
	pub async fn run_with_transport<F, C, Fut, IO>(
		self,
		listener: TcpListener,
		shutdown: F,
		connect: C,
	) -> Result<(), ServeError>
	where
		F: Future<Output = ()> + Send + 'static,
		C: Fn(TcpStream) -> Fut + Send + Sync + 'static,
		Fut: Future<Output = Option<IO>> + Send + 'static,
		IO: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
	{
		let app = Arc::new(self);
		serve_loop(listener, app, shutdown, connect).await;
		Ok(())
	}

	/// Bind `addr` and serve until SIGINT or SIGTERM, then drain in-flight
	/// connections and return. Unlike [`App::bind_ephemeral`], `addr` is
	/// caller-chosen — e.g. `0.0.0.0:$PORT` for a platform like fly.io that
	/// routes external traffic to the process directly.
	pub async fn bind(self, addr: SocketAddr) -> Result<(), ServeError> {
		let shutdown = shutdown_signal()?;
		let listener = TcpListener::bind(addr)
			.await
			.map_err(|e| ServeError::new(500, format!("failed to bind to {addr}: {e}")))?;
		self.run(listener, shutdown).await
	}


}

impl App<()> {
	pub fn stateless() -> Self {
		App::new(())
	}
}

/// Wires an accepted (and, for TLS, already-handshaken) connection up to the
/// hyper HTTP/1 service and drives it to completion. Shared by every accept
/// loop below so the framing/timeout setup is defined exactly once.
async fn serve_connection<S, IO>(io: IO, app: Arc<App<S>>, header_read_timeout: Duration, peer: SocketAddr)
where
	S: Send + Sync + 'static,
	IO: hyper::rt::Read + hyper::rt::Write + Unpin + Send + 'static,
{
	let app_for_conn = app.clone();
	let log_for_conn = app.log.clone();

	// Where a handler's upgrade callback is parked between being returned and the
	// connection being released. It cannot run inside the service call: the `101` has not
	// been written yet at that point, so hyper's upgrade future cannot resolve and awaiting
	// it there would deadlock the connection it is waiting on.
	let pending_upgrade: Arc<Mutex<Option<(hyper::upgrade::OnUpgrade, OnUpgrade)>>> =
		Arc::new(Mutex::new(None));
	let pending_for_service = pending_upgrade.clone();

	let svc = service_fn(move |mut req: Request<Incoming>| {
		let app = app.clone();
		let pending = pending_for_service.clone();
		// Taken before routing, because routing consumes the request. This is why a
		// handler cannot call `hyper::upgrade::on` itself — documented on `with_upgrades`.
		let upgrade = hyper::upgrade::on(&mut req);
		async move {
			// Pay for the log line only when there is a sink.
			let observed = app.log.as_ref().map(|_| {
				(
					std::time::Instant::now(),
					req.method().clone(),
					req.uri().path().to_string(),
				)
			});

			let resp = app.route_with_peer(req, peer).await;

			if let Some((started, method, path)) = observed {
				log_line(
					&app.log,
					format_args!(
						"{method} {path} {} {:.3}ms",
						resp.status().as_u16(),
						started.elapsed().as_secs_f64() * 1000.0
					),
				);
			}
			let mut resp = resp;
			if let Some(callback) = resp.extensions_mut().remove::<OnUpgrade>() {
				if app.upgrades_enabled {
					*pending.lock().unwrap() = Some((upgrade, callback));
				} else {
					// Silence here would leave the client holding a dead connection and
					// the handler author with nothing to go on.
					log_line(
						&app.log,
						format_args!(
							"handler returned an upgrade for {} but the app was not built \
							 with with_upgrades(); the connection will not be upgraded",
							resp.status().as_u16()
						),
					);
				}
			}
			Ok::<_, hyper::Error>(resp)
		}
	});
	let mut builder = http1::Builder::new();
	builder.timer(TokioTimer::new());
	builder.header_read_timeout(header_read_timeout);

	if app_for_conn.upgrades_enabled {
		let _ = builder.serve_connection(io, svc).with_upgrades().await;
	} else {
		let _ = builder.serve_connection(io, svc).await;
	}

	// The connection is released; if a handler asked to take it over, service the upgraded
	// stream here — still inside this task, which holds the semaphore permit and is the one
	// the shutdown drain aborts. A detached `tokio::spawn` here would escape both.
	let taken = pending_upgrade.lock().unwrap().take();
	if let Some((upgrade, callback)) = taken {
		match tokio::time::timeout(UPGRADE_HANDOFF_TIMEOUT, upgrade).await {
			Ok(Ok(upgraded)) => callback.run(TokioIo::new(upgraded)).await,
			Ok(Err(e)) => log_line(&log_for_conn, format_args!("upgrade failed: {e}")),
			Err(_) => log_line(
				&log_for_conn,
				format_args!(
					"upgrade did not complete within {UPGRADE_HANDOFF_TIMEOUT:?}; \
					 releasing the connection"
				),
			),
		}
	}
}

async fn plain_connect(stream: TcpStream) -> Option<TcpStream> {
	Some(stream)
}


/// Accept a connection and reserve it a connection-limit permit, retrying
/// transient `accept()` errors with [`Backoff`]. 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).
///
/// Deliberately returns one future that covers accept *and* permit
/// acquisition, so a caller can race the whole thing against a shutdown
/// signal in a single `select!`. Racing only the accept and leaving permit
/// acquisition as a bare `.await` afterward was the prior implementation's
/// bug: once a connection was accepted but was waiting on a saturated
/// semaphore, that wait was invisible to the `select!` and shutdown could not
/// preempt it.
async fn accept_and_permit<L: TcpAccept>(
	listener: &L,
	backoff: &mut Backoff,
	semaphore: &Arc<tokio::sync::Semaphore>,
) -> Option<(TcpStream, SocketAddr, tokio::sync::OwnedSemaphorePermit)> {
	// The permit is taken *before* accepting, and the order is load-bearing rather
	// than stylistic. This whole future is dropped whenever another arm of the
	// caller's `select!` wins — a finished connection being reaped, or shutdown — and
	// whatever it is holding at that moment is destroyed with it. A permit is free to
	// lose and re-acquire; an accepted `TcpStream` is a client's connection, and
	// dropping one mid-wait resets it. Accepting second also puts backpressure where
	// it belongs: at capacity the server stops accepting, so waiting clients sit in
	// the kernel's backlog instead of in a userspace queue that a cancellation can
	// silently empty.
	let permit = semaphore.clone().acquire_owned().await.ok()?;

	loop {
		match listener.accept().await {
			Ok((stream, peer)) => {
				backoff.reset();
				return Some((stream, peer, permit));
			}
			Err(_) => tokio::time::sleep(backoff.next_delay()).await,
		}
	}
}

/// Describe a failure to install a signal handler.
///
/// Split out so the mapping is testable: the failure itself needs a
/// signal-handler-hostile environment and cannot be provoked in-process, so the
/// shape of what a caller would receive is what gets asserted.
fn signal_install_error(signal: &str, cause: std::io::Error) -> ServeError {
	ServeError::new(500, format!("failed to install {signal} handler: {cause}"))
}

/// Set up a shutdown future that fires on SIGINT or SIGTERM.
///
/// Both handlers are installed eagerly, before the returned future is awaited,
/// so a server that cannot hear a shutdown signal fails at startup rather than
/// binding a port and then silently ignoring SIGTERM forever.
pub fn shutdown_signal() -> Result<impl Future<Output = ()> + Send, ServeError> {
	let mut sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())
		.map_err(|e| signal_install_error("SIGINT", e))?;
	let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
		.map_err(|e| signal_install_error("SIGTERM", e))?;

	Ok(async move {
		tokio::select! {
			_ = sigint.recv() => {}
			_ = sigterm.recv() => {}
		}
	})
}

/// Accept loop shared by every bind/run entry point (plain and TLS, with and
/// without graceful shutdown). `connect` turns a raw `TcpStream` into the
/// transport handed to `serve_connection` — identity for plaintext
/// ([`plain_connect`]), or whatever a caller supplies to
/// [`App::run_with_transport`] — `mini-tls` supplies a TLS handshake. A caller
/// that never needs graceful shutdown (`bind_ephemeral`)
/// passes `std::future::pending()`, which can never resolve, so this loop
/// degenerates into a plain accept-forever loop for them.
///
/// The accept-and-permit step and the shutdown signal are the two arms of a
/// single `select!`, so shutdown can win the race — and cancel a pending
/// accept or a permit wait cleanly — at any point, not just between
/// iterations.
async fn serve_loop<S, F, C, Fut, IO>(listener: TcpListener, app: Arc<App<S>>, shutdown: F, connect: C)
where
	S: Send + Sync + 'static,
	F: Future<Output = ()> + Send + 'static,
	C: Fn(TcpStream) -> Fut + Send + Sync + 'static,
	Fut: Future<Output = Option<IO>> + Send + 'static,
	IO: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
{
	let header_read_timeout = app.header_read_timeout;
	let connect_timeout = app.connect_timeout;
	let semaphore = Arc::new(tokio::sync::Semaphore::new(app.max_connections));
	let connect = Arc::new(connect);
	let mut backoff = Backoff::new();
	let mut join_set: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
	let mut shutdown_pin = std::pin::pin!(shutdown);
	let mut shutting_down = false;

	loop {
		if !shutting_down {
			tokio::select! {
				accepted = accept_and_permit(&listener, &mut backoff, &semaphore) => {
					match accepted {
						Some((stream, peer, permit)) => {
							let app = app.clone();
							let connect = connect.clone();
							join_set.spawn(async move {
								let _permit = permit;
								// Bounded here, inside the task holding the permit: a
								// transport that negotiates forever would otherwise hold a
								// connection slot forever, which is the same unbounded-wait
								// bug the header timeout exists to prevent, one layer down.
								let negotiated =
									tokio::time::timeout(connect_timeout, connect(stream)).await;
								if let Ok(Some(io)) = negotiated {
									serve_connection(TokioIo::new(io), app, header_read_timeout, peer).await;
								}
							});
						}
						None => shutting_down = true,
					}
				}
				// Reaping finished connections here — rather than only at shutdown — is
				// what makes a handler panic visible while the server is still running.
				// `join_next` on an empty set returns immediately with `None`, which
				// would busy-spin this arm, so it is only polled when work is in flight.
				joined = join_set.join_next(), if !join_set.is_empty() => {
					if let Some(joined) = joined {
						report_if_panicked(&app.log, joined);
					}
				}
				_ = shutdown_pin.as_mut() => {
					shutting_down = true;
				}
			}
			continue;
		}

		// Shutting down: drain what is in flight, but only for so long. A handler that
		// never returns — or a streaming response with no natural end, like a
		// server-sent-event stream — would otherwise hold shutdown open indefinitely.
		// Connections still running when the grace period expires are aborted, which the
		// client observes as a dropped connection: the correct outcome for a server that
		// was asked to stop and said it would.
		let drained = tokio::time::timeout(DEFAULT_SHUTDOWN_DRAIN_TIMEOUT, async {
			while let Some(joined) = join_set.join_next().await {
				report_if_panicked(&app.log, joined);
			}
		})
		.await;

		if drained.is_err() {
			join_set.shutdown().await;
		}
		break;
	}
}

/// Builder for configuring routes and settings before creating an `App`.
///
/// `RouteBuilder` uses a fluent API to register routes, configure CORS, and adjust
/// server settings. Call `.seal()` to produce the final `App<S>`.
///
/// # Example
///
/// ```ignore
/// use mini_serve::{RouteBuilder, handler, body};
/// use hyper::Response;
/// use hyper::body::Bytes;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
///     let app = RouteBuilder::stateless()
///         .get("/health", handler(|_, _| async {
///             Ok(Response::new(body(Bytes::from("OK"))))
///         }))
///         .seal();
///
///     app.bind("127.0.0.1:8080".parse()?).await?;
///     Ok(())
/// }
/// ```
#[must_use = "RouteBuilder does nothing until .seal() is called"]
pub struct RouteBuilder<S> {
	state:               Arc<S>,
	router:              Router<S>,
	max_body_size:       usize,
	header_read_timeout: Duration,
	log:                 Option<LogSink>,
	extra_headers:       Arc<Vec<(HeaderName, HeaderValue)>>,
	max_connections:     usize,
	connect_timeout:     Duration,
	error_handler:       ErrorHandler,
	cors_config:         Option<CorsConfig>,
	middlewares:         Vec<Middleware<S>>,
	upgrades_enabled:    bool,
}

impl<S: Send + Sync + 'static> RouteBuilder<S> {
	/// Create a new builder with shared state.
	pub fn new(state: S) -> Self {
		RouteBuilder {
			state:               Arc::new(state),
			router:              Router::new(),
			max_body_size:       DEFAULT_MAX_BODY_SIZE,
			header_read_timeout: DEFAULT_HEADER_READ_TIMEOUT,
			upgrades_enabled:    false,
			log:                 None,
			extra_headers:       Arc::new(Vec::new()),
			max_connections:     DEFAULT_MAX_CONNECTIONS,
			connect_timeout:     DEFAULT_CONNECT_TIMEOUT,
			error_handler:       default_error_handler(),
			cors_config:         None,
			middlewares:         Vec::new(),
		}
	}

	/// Register a middleware. Applies to every route registered *after* this
	/// call — middlewares wrap in registration order, so the first `.wrap()`
	/// call becomes the outermost layer and runs first.
	pub fn wrap(mut self, middleware: Middleware<S>) -> Self {
		self.middlewares.push(middleware);
		self
	}

	fn apply_middlewares(&self, handler: Handler<S>) -> Handler<S> {
		self.middlewares.iter().rev().fold(handler, |acc, mw| mw(acc))
	}

	/// Set the maximum request body size in bytes (default: 2 MiB).
	pub fn with_max_body_size(mut self, max: usize) -> Self {
		self.max_body_size = max;
		self
	}

	/// Set the header read timeout (default: 30 seconds).
	pub fn with_header_read_timeout(mut self, d: Duration) -> Self {
		self.header_read_timeout = d;
		self
	}

	/// Log one line per request to stderr, plus handler panics and the internal detail
	/// behind 5xx responses.
	///
	/// Off by default: a library writing to its host's stderr uninvited is a surprise.
	/// See [`RouteBuilder::with_request_logging_to`] to choose the destination.
	/// Send `name: value` on every response that does not already carry that header.
	///
	/// For the policy headers an API wants applied uniformly — `Strict-Transport-Security`,
	/// `Content-Security-Policy`, `Referrer-Policy`.
	///
	/// **Apply-if-absent**, deliberately: a handler that sets the header itself wins.
	/// This is the opposite of `mini-static`'s rule, and for the opposite reason — that
	/// crate computes every header itself, so a fixed value fighting a computed one is a
	/// bug, whereas handlers here are arbitrary user code that may legitimately vary a
	/// policy per route.
	///
	/// # Errors
	///
	/// [`ServeError`] if `name` or `value` is not a valid HTTP header, or if `name` is
	/// one the connection layer owns (`Content-Length`, `Connection`,
	/// `Transfer-Encoding`) — those describe framing this crate does not choose, so a
	/// fixed value could only contradict it.
	pub fn with_response_header(mut self, name: &str, value: &str) -> Result<Self, ServeError> {
		let name = HeaderName::from_bytes(name.as_bytes())
			.map_err(|_| ServeError::new(500, format!("invalid header name: {name}")))?;
		let value = HeaderValue::from_str(value)
			.map_err(|_| ServeError::new(500, format!("invalid value for header {name}")))?;

		if CONNECTION_OWNED_HEADERS.contains(&name) {
			return Err(ServeError::new(
				500,
				format!("{name} is owned by the connection layer and cannot be set as a fixed header"),
			));
		}

		Arc::make_mut(&mut self.extra_headers).push((name, value));
		Ok(self)
	}

	pub fn with_request_logging(self) -> Self {
		self.with_request_logging_to(Box::new(std::io::stderr()))
	}

	/// Log one line per request to `writer`, plus handler panics and 5xx internals.
	///
	/// Each served request writes `GET /path 200 0.421ms` — method, path exactly as
	/// received, status, and handling time. Two failures that are otherwise invisible
	/// also come here:
	///
	/// - **Handler panics.** The task's result was previously discarded, so a panicking
	///   handler dropped the client's connection and left no trace anywhere.
	/// - **5xx internal detail.** The client body is sanitized to
	///   `{"message":"internal server error"}` deliberately; without a sink the real
	///   message was discarded with it, leaving an operator nothing to debug from.
	///
	/// The path is logged exactly as received, not decoded: it is attacker-controlled
	/// input, and whoever reads the log deserves the bytes that actually arrived.
	/// Writes are serialized across connections; write failures are ignored rather than
	/// allowed to fail a request.
	pub fn with_request_logging_to(mut self, writer: Box<dyn std::io::Write + Send>) -> Self {
		self.log = Some(Arc::new(Mutex::new(writer)));
		self
	}

	/// How long a transport has to turn an accepted connection into a usable one
	/// (default: 10 seconds).
	///
	/// Bounds the `connect` step of [`App::run_with_transport`], so a transport that
	/// negotiates forever cannot hold a connection slot forever. It has no effect on the
	/// identity transport [`App::run`] uses, which cannot block — the guarantee exists for
	/// transports that do work, which is the only kind worth plugging in.
	///
	/// Replaces the former `with_tls_handshake_timeout`: the bound belongs to the server's
	/// connection lifecycle rather than to any one transport.
	pub fn with_connect_timeout(mut self, d: Duration) -> Self {
		self.connect_timeout = d;
		self
	}

	/// Set the maximum concurrent connections (default: 1024).
	/// Allow handlers to take over a connection with [`OnUpgrade`].
	///
	/// Off by default: an application with no upgrade route should not pay for the
	/// capability, and enabling it changes how every connection on this server is served.
	/// Without it, a `101` response is written and the callback never runs — which is
	/// reported to the log sink if one is configured, since the client would otherwise see
	/// a dead connection and the author would see nothing.
	///
	/// The server takes the request's upgrade future before routing, so a handler that
	/// calls `hyper::upgrade::on` itself receives nothing. Use [`OnUpgrade`] instead; it
	/// exists so the upgraded stream is serviced inside the connection's own task, keeping
	/// it inside the connection ceiling and the shutdown drain.
	pub fn with_upgrades(mut self) -> Self {
		self.upgrades_enabled = true;
		self
	}

	pub fn with_max_connections(mut self, max: usize) -> Self {
		self.max_connections = max;
		self
	}

	pub fn with_error_handler(
		mut self,
		f: impl Fn(StatusCode, &str) -> Response<ResponseBody> + Send + Sync + 'static,
	) -> Self {
		self.error_handler = Arc::new(f);
		self
	}

	pub fn with_cors(mut self, config: CorsConfig) -> Self {
		self.cors_config = Some(config);
		self
	}

	pub fn get(mut self, path: &str, handler: Handler<S>) -> Self {
		let handler = self.apply_middlewares(handler);
		self.router.insert(Method::GET, path, handler);
		self
	}

	pub fn post(mut self, path: &str, handler: Handler<S>) -> Self {
		let handler = self.apply_middlewares(handler);
		self.router.insert(Method::POST, path, handler);
		self
	}

	pub fn put(mut self, path: &str, handler: Handler<S>) -> Self {
		let handler = self.apply_middlewares(handler);
		self.router.insert(Method::PUT, path, handler);
		self
	}

	pub fn delete(mut self, path: &str, handler: Handler<S>) -> Self {
		let handler = self.apply_middlewares(handler);
		self.router.insert(Method::DELETE, path, handler);
		self
	}

	/// Register a `PATCH` handler.
	///
	/// Dispatch and the `Allow` header are generic over `Method`, so this needs nothing
	/// from the router that its siblings do not — it was simply missing.
	pub fn patch(mut self, path: &str, handler: Handler<S>) -> Self {
		let handler = self.apply_middlewares(handler);
		self.router.insert(Method::PATCH, path, handler);
		self
	}

	pub fn seal(self) -> App<S> {
		App {
			state:              self.state,
			router:             Arc::new(self.router),
			max_body_size:       self.max_body_size,
			header_read_timeout: self.header_read_timeout,
			upgrades_enabled:    self.upgrades_enabled,
			log:                 self.log,
			extra_headers:       self.extra_headers,
			max_connections:     self.max_connections,
			connect_timeout:     self.connect_timeout,
			error_handler:       self.error_handler,
			cors_config:         self.cors_config,
		}
	}
}

impl RouteBuilder<()> {
	pub fn stateless() -> Self {
		RouteBuilder::new(())
	}
}

#[cfg(test)]
#[path = "../tests/unit/app.rs"]
mod tests;