simoom 0.1.0

//! [`Server`], its [`Builder`] and [`ServerHandle`]; the state shared by every connection; the
//! service call with panic catching and the stall watchdog.

use crate::limits::Limits;
use crate::reply::{Reply, Target};
use crate::request::Request;
use crate::service::Service;
use simoom_rt::offload::{OffloadConfig, OffloadStats};
use simoom_rt::{Backend, ConnInfo, Connection, Runtime, RuntimeHandle};
use std::io;
use std::net::{SocketAddr, TcpListener, ToSocketAddrs};
use std::panic::{catch_unwind, AssertUnwindSafe};
use crate::watchdog::Watch;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};

/// Counters of one loop thread (each loop writes only its own, so they never contend).
#[derive(Default, Debug)]
#[repr(align(64))]
pub(crate) struct Counters {
    pub(crate) requests: AtomicU64,
    pub(crate) bad_requests: AtomicU64,
    pub(crate) too_large: AtomicU64,
    pub(crate) timeouts: AtomicU64,
    pub(crate) panics: AtomicU64,
    pub(crate) stalls: AtomicU64,
    pub(crate) stall_checks: AtomicU64,
    pub(crate) offloaded_routes: AtomicU64,
    pub(crate) overloaded: AtomicU64,
    pub(crate) shed: AtomicU64,
    pub(crate) upgrades: AtomicU64,
    pub(crate) h2_connections: AtomicU64,
    pub(crate) h3_connections: AtomicU64,
    pub(crate) quic_endpoints: AtomicU64,
    pub(crate) tls_errors: AtomicU64,
}

/// Counter slots, indexed by loop number modulo this (loops beyond it share slots).
const COUNTER_SLOTS: usize = 64;

impl Counters {
    const fn zero() -> Counters {
        Counters {
            requests: AtomicU64::new(0),
            bad_requests: AtomicU64::new(0),
            too_large: AtomicU64::new(0),
            timeouts: AtomicU64::new(0),
            panics: AtomicU64::new(0),
            stalls: AtomicU64::new(0),
            stall_checks: AtomicU64::new(0),
            offloaded_routes: AtomicU64::new(0),
            overloaded: AtomicU64::new(0),
            shed: AtomicU64::new(0),
            upgrades: AtomicU64::new(0),
            h2_connections: AtomicU64::new(0),
            h3_connections: AtomicU64::new(0),
            quic_endpoints: AtomicU64::new(0),
            tls_errors: AtomicU64::new(0),
        }
    }
}

fn new_counters() -> Box<[Counters; COUNTER_SLOTS]> {
    // All zero bytes: filled with one memset, not field by field in a loop.
    Box::new([const { Counters::zero() }; COUNTER_SLOTS])
}

pub(crate) fn bump(c: &AtomicU64) {
    c.fetch_add(1, Ordering::Relaxed);
}

/// Everything the connections share: the service, limits, protocol settings and counters.
pub(crate) struct Shared {
    pub(crate) service: Arc<dyn Service>,
    pub(crate) limits: Limits,
    #[cfg(feature = "http1")]
    pub(crate) h1: simoom_http1::Limits,
    #[cfg_attr(not(feature = "http2"), allow(dead_code))]
    pub(crate) http2: bool,
    #[cfg(feature = "http2")]
    pub(crate) h2: simoom_h2::Config,
    /// `alt-svc: ...\r\n`, set once the UDP port is known.
    pub(crate) alt_svc: OnceLock<(Vec<u8>, Vec<u8>)>,
    counters: Box<[Counters; COUNTER_SLOTS]>,
    pub(crate) watchdog: crate::watchdog::Policy,
    /// [`Limits::shed_lag`] in ns (0: off).
    pub(crate) shed_ns: u64,
    /// Watchdog state per path slot, for services without routes ([`Service::watch`] `None`).
    slots: Box<[Arc<Watch>]>,
    /// Slots that are suspect or offloaded: the path is hashed only while this is not 0.
    slots_active: Arc<AtomicU32>,
    log: Arc<dyn Fn(&str) + Send + Sync>,
    /// [`Limits::defer_timeout`] in ns (0: off).
    pub(crate) defer_ns: u64,
    /// [`Builder::on_complete`].
    pub(crate) on_complete: Option<CompleteHook>,
    /// [`Builder::compression`].
    #[cfg(feature = "compression")]
    pub(crate) compression: Option<Arc<crate::compress::Compression>>,
    /// [`Builder::websocket_deflate`].
    #[cfg(feature = "ws-deflate")]
    pub(crate) ws_deflate: Option<simoom_http1::ws::deflate::DeflateConfig>,
    /// The TLS configuration for new connections; [`ServerHandle::set_tls`] swaps it.
    #[cfg(feature = "tls")]
    pub(crate) tls: Option<std::sync::RwLock<Arc<simoom_tls::ServerConfig>>>,
    /// HTTP/3's TLS configuration; [`ServerHandle::set_quic_tls`] swaps it.
    #[cfg(feature = "http3")]
    quic_tls: Option<Arc<crate::h3::QuicTls>>,
    /// Open connections per client IP, kept only when [`Limits::max_connections_per_ip`] is set.
    per_ip: std::sync::Mutex<crate::util::IpCounts>,
    /// Connection rate per client IP, used only when [`Limits::reconnect_rate`] is set.
    #[cfg(feature = "guard")]
    reconnects: std::sync::Mutex<crate::util::Reconnects>,
}

type CompleteHook = Arc<dyn Fn(&Completion) + Send + Sync>;

/// One finished request, for access logs and metrics ([`Builder::on_complete`]).
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct Completion {
    /// The status sent (the error status for a request refused by the parser).
    pub status: u16,
    /// Request body bytes received.
    pub bytes_in: u64,
    /// Response body bytes handed to the connection (not the head or the framing).
    pub bytes_out: u64,
    /// From the loop turn that read the request head to the one that finished the response.
    pub duration: Duration,
    /// The protocol.
    pub version: crate::types::Version,
    /// The client's address, when known.
    pub peer: Option<SocketAddr>,
    /// The listener ([`Request::listener`]).
    pub listener: usize,
    /// The response did not complete: the client went away, the stream was reset, or a
    /// streamed body was aborted.
    pub aborted: bool,
}

impl Shared {
    /// Reports a finished request to [`Builder::on_complete`]; callers check
    /// `on_complete.is_some()` first, so a server without a hook pays one branch.
    #[inline(never)]
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn complete(&self, status: u16, bytes_in: u64, bytes_out: u64, start_ns: u64, now_ns: u64, version: crate::types::Version, peer: Option<SocketAddr>, listener: u32, aborted: bool) {
        let Some(f) = &self.on_complete else { return };
        let c = Completion { status, bytes_in, bytes_out, duration: Duration::from_nanos(now_ns.saturating_sub(start_ns)), version, peer, listener: listener as usize, aborted };
        let _ = catch_unwind(AssertUnwindSafe(|| f(&c)));
    }
}

/// A connection's place in the per-IP count; released when the connection is dropped.
pub(crate) struct IpSlot {
    sh: Arc<Shared>,
    ip: std::net::IpAddr,
}

impl Drop for IpSlot {
    fn drop(&mut self) {
        self.sh.per_ip.lock().unwrap_or_else(|e| e.into_inner()).release(self.ip);
    }
}

impl Shared {
    /// [`Limits::reconnect_rate`] (out of line: off by default).
    #[cfg(feature = "guard")]
    #[cold]
    #[inline(never)]
    fn reconnect_allowed(&self, peer: Option<std::net::SocketAddr>) -> bool {
        let Some(p) = peer else { return true };
        self.reconnects.lock().unwrap_or_else(|e| e.into_inner()).allow(p.ip(), self.limits.reconnect_rate, simoom_sys::time::now_ns())
    }

    /// Takes a per-IP slot for a new connection; `Err` when the IP is at its cap or over its
    /// reconnect rate.
    pub(crate) fn take_ip_slot(self: &Arc<Self>, peer: Option<std::net::SocketAddr>) -> Result<Option<IpSlot>, ()> {
        #[cfg(feature = "guard")]
        if self.limits.reconnect_rate != 0 && !self.reconnect_allowed(peer) {
            return Err(());
        }
        let cap = self.limits.max_connections_per_ip;
        let Some(peer) = peer.filter(|_| cap != 0) else { return Ok(None) };
        let ip = peer.ip();
        if !self.per_ip.lock().unwrap_or_else(|e| e.into_inner()).take(ip, cap) {
            return Err(());
        }
        Ok(Some(IpSlot { sh: self.clone(), ip }))
    }
}

impl Shared {
    pub(crate) fn counters(&self, loop_index: usize) -> &Counters {
        &self.counters[loop_index % COUNTER_SLOTS]
    }

    /// The `alt-svc` line for HTTP/1.1 responses.
    pub(crate) fn alt_svc_line(&self) -> Option<&[u8]> {
        self.alt_svc.get().map(|a| &a.0[..])
    }

    /// The `alt-svc` value for HTTP/2 responses.
    #[cfg_attr(not(feature = "http2"), allow(dead_code))]
    pub(crate) fn alt_svc_value(&self) -> Option<&[u8]> {
        self.alt_svc.get().map(|a| &a.1[..])
    }

    pub(crate) fn log(&self, msg: &str) {
        (self.log)(msg)
    }

    /// The watchdog slot of a request to a service without routes: its method and path.
    fn slot(&self, req: &Request<'_>) -> &Arc<Watch> {
        use std::hash::Hasher;
        let mut h = crate::util::Fnv::default();
        h.write(req.method());
        h.write(req.path());
        &self.slots[(h.finish() as usize) % self.slots.len()]
    }

    /// The watchdog's part after a call that overran the budget or was measured.
    #[cold]
    #[inline(never)]
    fn after_call(&self, req: &Request<'_>, took_ns: u64) {
        let p = &self.watchdog;
        let (w, k, name) = match self.service.watch(req) {
            Some(r) => (r.watch, r.stalls.unwrap_or(p.stalls), std::borrow::Cow::Borrowed(r.name)),
            None => (self.slot(req), p.stalls, String::from_utf8_lossy(req.path())),
        };
        let c = self.counters(req.loop_index);
        let Some(s) = req.sample else {
            if took_ns > p.budget_ns && w.wall_overrun() {
                bump(&c.stall_checks);
            }
            return;
        };
        let ev = w.measured(took_ns, s, p, k, simoom_sys::time::now_ns());
        if ev == crate::watchdog::Event::None {
            return;
        }
        bump(&c.stalls);
        if !w.logged.swap(true, Ordering::Relaxed) {
            // Microseconds as integers: `Duration`'s Debug costs 2 KB of code.
            self.log(&format!(
                "simoom: an inline handler on {name} took {} us ({} us of CPU; budget {} us); slow handlers belong on the offload pool",
                took_ns / 1000,
                s.cpu_ns / 1000,
                p.budget_ns / 1000
            ));
        }
        if ev == crate::watchdog::Event::Offloaded {
            bump(&c.offloaded_routes);
            self.log(&format!("simoom: {name} stalled {k} times within {} ms and now runs on the offload pool", p.window_ns / 1_000_000));
        }
    }
}

/// Calls the service for one request: on the offload pool when the watchdog moved the
/// request's path there, else inline, timed, with panics caught. Returns `true` if the service
/// panicked. (A [`crate::Router`] does the same per route inside its own call.)
///
/// `clock` carries the time the previous call of the same callback ended, so pipelined
/// requests cost one clock read each instead of two.
pub(crate) fn call(sh: &Shared, req: &mut Request<'_>, t: &mut dyn Target, clock: &mut Option<Instant>) -> bool {
    let budget = sh.watchdog.budget_ns;
    let watched = budget != 0 && cfg!(feature = "watchdog");
    let mut measure = false;
    let mut t0 = None;
    if watched {
        if sh.slots_active.load(Ordering::Relaxed) != 0 {
            let w = sh.slot(req);
            if w.offloaded() {
                offload_slot(sh, req, t, w);
                return false;
            }
            measure = w.suspect();
        }
        t0 = Some(clock.take().unwrap_or_else(Instant::now));
    }
    // One call site for the service (each one inlines the reply set-up).
    let mark = if measure { Some(crate::watchdog::mark()) } else { None };
    let mut deferred = false;
    let r = catch_unwind(AssertUnwindSafe(|| {
        deferred = matches!(sh.service.call(req, new_reply(sh, req, t)), crate::service::Outcome::Deferred);
    }));
    if let Some(t0) = t0 {
        let t1 = Instant::now();
        *clock = Some(t1);
        // A handler that deferred (K1) handed its work to another thread: whatever the hand-off
        // cost (a contended queue lock, a wake-up), it is never a stall, and moving its route to
        // the offload pool would only add a second hand-off per request.
        if deferred {
            req.sample = None;
        } else {
            if let Some(m) = mark {
                req.sample = crate::watchdog::since(m);
            }
            let took = (t1 - t0).as_nanos() as u64;
            if took > budget || req.sample.is_some() {
                sh.after_call(req, took);
            }
        }
    }
    finish_call(sh, req, r)
}

/// A request whose path slot the watchdog moved to the offload pool (rare: out of line).
#[cold]
#[inline(never)]
fn offload_slot(sh: &Shared, req: &mut Request<'_>, t: &mut dyn Target, w: &Arc<Watch>) {
    let _ = new_reply(sh, req, t).offload_watched(req, sh.service.clone(), w.clone());
}

/// The reply for a request, with its content coding negotiated (feature `compression`).
#[inline(always)]
fn new_reply<'a>(sh: &Shared, req: &Request<'_>, t: &'a mut dyn Target) -> Reply<'a> {
    #[cfg(not(feature = "compression"))]
    let _ = (sh, req);
    #[cfg(feature = "compression")]
    if let Some(c) = &sh.compression {
        return Reply::new(t).with_coding(crate::compress::negotiate(c, req.header("accept-encoding")));
    }
    Reply::new(t)
}

/// Path slots of the watchdog for services without routes.
const WATCH_SLOTS: usize = 64;

fn finish_call(sh: &Shared, req: &Request<'_>, r: std::thread::Result<()>) -> bool {
    if r.is_err() {
        bump(&sh.counters(req.loop_index).panics);
        return true;
    }
    false
}

/// Server counters.
#[derive(Clone, Debug, Default)]
pub struct Stats {
    /// Connections accepted.
    pub accepted: u64,
    /// Connections refused (over a cap).
    pub rejected: u64,
    /// Connections open now.
    pub active: usize,
    /// Requests started (every protocol).
    pub requests: u64,
    /// Malformed requests (400-class parse errors).
    pub bad_requests: u64,
    /// Requests over a size limit (413, 414, 431).
    pub too_large: u64,
    /// Connections closed by a timeout (head deadline, body rate, idle with a partial request,
    /// write deadline).
    pub timeouts: u64,
    /// Handlers that panicked.
    pub panics: u64,
    /// Inline handler calls the watchdog confirmed as stalls (over the CPU budget, or blocked).
    pub stalls: u64,
    /// Calls over the budget on the wall clock that made their route suspect (and measured);
    /// most are a preempted loop thread, not a slow handler.
    pub stall_checks: u64,
    /// Times a route moved to the offload pool.
    pub offloaded_routes: u64,
    /// Requests answered 503 because the offload pool was full or overloaded.
    pub overloaded: u64,
    /// Requests answered 503 by admission control ([`Limits::shed_lag`]).
    pub shed: u64,
    /// WebSocket and h2c upgrades.
    pub upgrades: u64,
    /// HTTP/2 connections.
    pub h2_connections: u64,
    /// HTTP/3 connections.
    pub h3_connections: u64,
    /// QUIC endpoints built: one per loop, on that loop's first HTTP/3 datagram (a loop that
    /// never sees HTTP/3 traffic builds none).
    pub quic_endpoints: u64,
    /// Failed TLS handshakes.
    pub tls_errors: u64,
    /// The offload pool.
    pub offload: OffloadStats,
    /// The runtime: per-loop system calls, buffers, connections.
    pub runtime: simoom_rt::Stats,
}

/// Entry point: [`Server::builder`].
#[derive(Debug)]
pub struct Server;

impl Server {
    /// A [`Builder`].
    pub fn builder() -> Builder {
        Builder::default()
    }

    /// A builder serving existing listeners (socket activation, a shared port).
    pub fn from_listeners(listeners: Vec<TcpListener>) -> Builder {
        let mut b = Builder::default();
        for l in listeners {
            b = b.listen_on(l);
        }
        b
    }
}

enum Listen {
    Addr(io::Result<Vec<SocketAddr>>),
    Std(TcpListener),
}

type RtConfigure = Box<dyn FnOnce(&mut simoom_rt::Config)>;
#[cfg(feature = "http3")]
type QuicConfigure = Box<dyn FnOnce(&mut simoom_quic::Config)>;

/// Configures and starts a server. See [`Server::builder`].
pub struct Builder {
    tcp: Vec<(Listen, bool)>,
    #[cfg(feature = "http3")]
    quic: Vec<io::Result<Vec<SocketAddr>>>,
    threads: usize,
    backend: Backend,
    limits: Limits,
    http2: bool,
    alt_svc: Option<String>,
    service: Option<Arc<dyn Service>>,
    runtime: Vec<RtConfigure>,
    log: Arc<dyn Fn(&str) + Send + Sync>,
    #[cfg(feature = "tls")]
    tls: Option<simoom_tls::ServerConfig>,
    #[cfg(feature = "http3")]
    quic_tls: Option<simoom_tls::ServerConfig>,
    #[cfg(feature = "http3")]
    quic_config: Option<QuicConfigure>,
    #[cfg(feature = "http3")]
    quic_gro: bool,
    on_complete: Option<CompleteHook>,
    #[cfg(feature = "compression")]
    compression: Option<crate::compress::Compression>,
    #[cfg(feature = "ws-deflate")]
    ws_deflate: Option<simoom_http1::ws::deflate::DeflateConfig>,
}

impl Default for Builder {
    fn default() -> Self {
        Builder {
            tcp: Vec::new(),
            #[cfg(feature = "http3")]
            quic: Vec::new(),
            threads: 0,
            backend: Backend::Auto,
            limits: Limits::default(),
            http2: true,
            alt_svc: None,
            service: None,
            runtime: Vec::new(),
            log: Arc::new(|m: &str| eprintln!("{m}")),
            #[cfg(feature = "tls")]
            tls: None,
            #[cfg(feature = "http3")]
            quic_tls: None,
            #[cfg(feature = "http3")]
            quic_config: None,
            #[cfg(feature = "http3")]
            quic_gro: true,
            on_complete: None,
            #[cfg(feature = "compression")]
            compression: None,
            #[cfg(feature = "ws-deflate")]
            ws_deflate: None,
        }
    }
}

impl std::fmt::Debug for Builder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Builder").field("listeners", &self.tcp.len()).field("threads", &self.threads).field("limits", &self.limits).finish()
    }
}

/// The ALPN list of TCP TLS listeners.
#[cfg(feature = "tls")]
fn tcp_alpn(http2: bool) -> Vec<Vec<u8>> {
    if http2 {
        vec![b"h2".to_vec(), b"http/1.1".to_vec()]
    } else {
        vec![b"http/1.1".to_vec()]
    }
}

fn resolve(addr: impl ToSocketAddrs) -> io::Result<Vec<SocketAddr>> {
    // The runtime puts IPv4 first (`localhost` should mean 127.0.0.1).
    let v: Vec<SocketAddr> = addr.to_socket_addrs()?.collect();
    if v.is_empty() {
        return Err(io::Error::new(io::ErrorKind::InvalidInput, "no address to listen on"));
    }
    Ok(v)
}

impl Builder {
    /// Listens on `addr` (plain TCP: HTTP/1.1, and h2c when HTTP/2 is on). Address errors are
    /// reported by [`Builder::start`].
    pub fn listen(mut self, addr: impl ToSocketAddrs) -> Self {
        self.tcp.push((Listen::Addr(resolve(addr)), false));
        self
    }

    /// Listens on `addr` with TLS ([`Builder::tls`]); ALPN chooses h2 or http/1.1.
    #[cfg(feature = "tls")]
    pub fn listen_tls(mut self, addr: impl ToSocketAddrs) -> Self {
        self.tcp.push((Listen::Addr(resolve(addr)), true));
        self
    }

    /// Serves an existing listener (plain TCP).
    pub fn listen_on(mut self, listener: TcpListener) -> Self {
        self.tcp.push((Listen::Std(listener), false));
        self
    }

    /// Serves an existing listener with TLS.
    #[cfg(feature = "tls")]
    pub fn listen_on_tls(mut self, listener: TcpListener) -> Self {
        self.tcp.push((Listen::Std(listener), true));
        self
    }

    /// Serves HTTP/3 over QUIC on the UDP address `addr`, with the TLS configuration given to
    /// [`Builder::quic_tls`] (or, failing that, error at start). HTTP/1.1 and HTTP/2 responses
    /// then announce it with `alt-svc`.
    #[cfg(feature = "http3")]
    pub fn listen_quic(mut self, addr: impl ToSocketAddrs) -> Self {
        self.quic.push(resolve(addr));
        self
    }

    /// The TLS configuration for [`Builder::listen_tls`] listeners. Its ALPN list is replaced
    /// with `h2` (when HTTP/2 is on) and `http/1.1`.
    #[cfg(feature = "tls")]
    pub fn tls(mut self, config: simoom_tls::ServerConfig) -> Self {
        self.tls = Some(config);
        self
    }

    /// The TLS configuration for HTTP/3 listeners (its ALPN becomes `h3`).
    #[cfg(feature = "http3")]
    pub fn quic_tls(mut self, config: simoom_tls::ServerConfig) -> Self {
        self.quic_tls = Some(config);
        self
    }

    /// Adjusts the QUIC configuration (transport parameters, Retry) before start.
    #[cfg(feature = "http3")]
    pub fn quic_config(mut self, f: impl FnOnce(&mut simoom_quic::Config) + 'static) -> Self {
        self.quic_config = Some(Box::new(f));
        self
    }

    /// UDP generic receive offload on HTTP/3 listeners (Linux only; on by default): the kernel
    /// hands over a batch of datagrams from one peer as one 64 KiB buffer, one receive call
    /// for up to 64 packets.
    #[cfg(feature = "http3")]
    pub fn quic_gro(mut self, on: bool) -> Self {
        self.quic_gro = on;
        self
    }

    /// Loop threads: 0 (the default) means one per CPU.
    pub fn threads(mut self, n: usize) -> Self {
        self.threads = n;
        self
    }

    /// The I/O back end: `Auto` picks io_uring only when enabled, else epoll, IOCP or kqueue.
    pub fn backend(mut self, b: Backend) -> Self {
        self.backend = b;
        self
    }

    /// Every limit (plan §7.1 defaults).
    pub fn limits(mut self, limits: Limits) -> Self {
        self.limits = limits;
        self
    }

    /// HTTP/2 on or off (on by default with the `http2` feature): h2c prior knowledge,
    /// `Upgrade: h2c`, and ALPN `h2`.
    pub fn http2(mut self, on: bool) -> Self {
        self.http2 = on;
        self
    }

    /// Overrides the `alt-svc` value announced on HTTP/1.1 and HTTP/2 responses (by default
    /// `h3=":<port>"; ma=86400` when HTTP/3 listens).
    pub fn alt_svc(mut self, value: impl Into<String>) -> Self {
        self.alt_svc = Some(value.into());
        self
    }

    /// The service that answers every request.
    pub fn service(mut self, service: impl Service) -> Self {
        self.service = Some(Arc::new(service));
        self
    }

    /// Adjusts the runtime configuration (buffer sizes, pinning, back-end details).
    pub fn runtime(mut self, f: impl FnOnce(&mut simoom_rt::Config) + 'static) -> Self {
        self.runtime.push(Box::new(f));
        self
    }

    /// Where log lines go (stall warnings). Default: standard error.
    pub fn log(mut self, f: impl Fn(&str) + Send + Sync + 'static) -> Self {
        self.log = Arc::new(f);
        self
    }

    /// Calls `f` once per finished request, on the loop thread that served it, with its status,
    /// byte counts, duration, protocol and peer: for access logs and per-request metrics.
    /// Requests the parser refuses (400, 413, 431...) are reported too. `f` must be quick and
    /// must not block. Without a hook (the default) the server only checks that there is none.
    pub fn on_complete(mut self, f: impl Fn(&Completion) + Send + Sync + 'static) -> Self {
        self.on_complete = Some(Arc::new(f));
        self
    }

    /// Offers WebSocket permessage-deflate (RFC 7692) to clients that ask for it (feature
    /// `ws-deflate`), on HTTP/1.1 and HTTP/2 WebSockets. Inflated messages are bounded by
    /// [`crate::Limits::max_ws_message`] as well as `cfg.max_message`. Each compressing
    /// WebSocket holds about 330 KiB of codec state.
    #[cfg(feature = "ws-deflate")]
    pub fn websocket_deflate(mut self, cfg: crate::WsDeflate) -> Self {
        self.ws_deflate = Some(cfg);
        self
    }

    /// Compresses responses with gzip or deflate when the client accepts it (feature
    /// `compression`); see [`crate::Compression`] for what is compressed and how.
    #[cfg(feature = "compression")]
    pub fn compression(mut self, c: crate::Compression) -> Self {
        self.compression = Some(c);
        self
    }

    /// Binds every listener, starts the loops, and returns once they run.
    pub fn start(self) -> io::Result<ServerHandle> {
        let service = self.service.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "no service"))?;
        let limits = self.limits;
        let http2 = self.http2 && cfg!(feature = "http2");
        #[cfg(feature = "tls")]
        let tls = match self.tls {
            Some(mut c) => {
                c.alpn = tcp_alpn(http2);
                Some(std::sync::RwLock::new(Arc::new(c)))
            }
            None => None,
        };
        #[cfg(feature = "tls")]
        if tls.is_none() && self.tcp.iter().any(|(_, t)| *t) {
            return Err(io::Error::new(io::ErrorKind::InvalidInput, "a TLS listener needs Builder::tls"));
        }
        #[cfg(feature = "http2")]
        let h2 = simoom_h2::Config {
            max_concurrent_streams: limits.max_concurrent_streams,
            max_header_list_size: limits.max_header_list,
            header_table_size: limits.header_table_size,
            budgets: limits.h2_frame_budgets.h2(),
            // WebSocket over HTTP/2 (RFC 8441), feature `websocket-h2`
            enable_connect_protocol: cfg!(feature = "websocket-h2"),
            keepalive_interval_ms: limits.h2_keepalive.map(|d| (d.as_millis() as u64).max(1)),
            keepalive_timeout_ms: (limits.h2_keepalive_timeout.as_millis() as u64).max(1),
            ..simoom_h2::Config::default()
        };
        let slots_active = Arc::new(AtomicU32::new(0));
        let shared = Arc::new(Shared {
            service,
            #[cfg(feature = "http1")]
            h1: limits.http1(),
            http2,
            #[cfg(feature = "http2")]
            h2,
            alt_svc: OnceLock::new(),
            counters: new_counters(),
            watchdog: limits.watchdog(),
            shed_ns: limits.shed_lag.map_or(0, |d| (d.as_nanos() as u64).max(1)),
            slots: (0..if cfg!(feature = "watchdog") { WATCH_SLOTS } else { 0 }).map(|_| Arc::new(Watch::new(Some(slots_active.clone())))).collect(),
            slots_active,
            log: self.log,
            defer_ns: limits.defer_timeout.map_or(0, |d| (d.as_nanos() as u64).max(1)),
            on_complete: self.on_complete,
            #[cfg(feature = "compression")]
            compression: self.compression.map(Arc::new),
            #[cfg(feature = "ws-deflate")]
            ws_deflate: self.ws_deflate.map(|mut c| {
                c.max_message = c.max_message.min(limits.max_ws_message);
                c
            }),
            #[cfg(feature = "http3")]
            quic_tls: match self.quic_tls {
                Some(c) if !self.quic.is_empty() => Some(Arc::new(crate::h3::QuicTls::new(c))),
                _ => None,
            },
            per_ip: std::sync::Mutex::new(crate::util::IpCounts::default()),
            #[cfg(feature = "guard")]
            reconnects: std::sync::Mutex::new(crate::util::Reconnects::default()),
            #[cfg(feature = "tls")]
            tls,
            limits,
        });
        if let Some(v) = &self.alt_svc {
            let _ = shared.alt_svc.set((format!("alt-svc: {v}\r\n").into_bytes(), v.clone().into_bytes()));
        }

        let lim = &shared.limits;
        let mut rt = Runtime::builder().threads(self.threads).backend(self.backend).configure(|c| {
            c.max_connections = lim.max_connections;
            c.max_connections_per_loop = lim.max_connections_per_loop;
            c.max_input = (lim.max_head * 2 + 16 * 1024).max(64 * 1024);
            c.thread_name = "simoom".into();
            // One loop until three connections are open at once: an idle server, or one that
            // serves a connection at a time (FaaS), costs one loop thread in start-up time and
            // memory. Servers with HTTP/3 start every loop (see simoom_rt::Config::lazy_loops).
            c.lazy_loops = true;
            c.offload = OffloadConfig {
                max_threads: lim.offload_threads,
                max_queue: lim.offload_queue,
                target: lim.offload_target,
                interval: lim.offload_interval,
                ..OffloadConfig::default()
            };
        });
        for f in self.runtime {
            rt = rt.configure(f);
        }
        for (src, secure) in self.tcp {
            let sh = shared.clone();
            let acceptor = move |info: &ConnInfo| {
                // Over the per-IP cap: refused (the runtime counts it in `rejected`).
                let slot = sh.take_ip_slot(info.peer).ok()?;
                Some(Box::new(crate::conn::Conn::new(sh.clone(), info, secure, slot)) as Box<dyn Connection>)
            };
            rt = match src {
                Listen::Addr(a) => rt.listen(&a?[..], acceptor),
                Listen::Std(l) => rt.listen_on(l, acceptor),
            };
        }
        #[cfg(feature = "http3")]
        let quic_cfg = if self.quic.is_empty() {
            None
        } else {
            let provider = shared.quic_tls.clone().ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "an HTTP/3 listener needs Builder::quic_tls"))?;
            let mut q = simoom_quic::Config::new(provider);
            q.transport.max_streams_bidi = u64::from(shared.limits.max_concurrent_streams);
            q.transport.max_idle_timeout = shared.limits.keep_alive_timeout.max(Duration::from_secs(1));
            q.transport.disable_active_migration = true;
            if let Some(f) = self.quic_config {
                f(&mut q);
            }
            Some(Arc::new(q))
        };
        #[cfg(feature = "http3")]
        if let Some(q) = &quic_cfg {
            for a in self.quic {
                let a = a?;
                let sh = shared.clone();
                let q = q.clone();
                // GRO on Linux: a batch of datagrams from one peer arrives as one 64 KiB buffer
                // (one receive and one wake-up for up to 64 packets of a bulk upload).
                let opts = simoom_rt::UdpOptions { gro: self.quic_gro && cfg!(target_os = "linux"), ..simoom_rt::UdpOptions::default() };
                rt = rt.udp(&a[..], opts, move |info: &simoom_rt::UdpInfo| crate::h3::endpoint(sh.clone(), q.clone(), info));
            }
        }
        let rt = rt.start()?;
        #[cfg(feature = "http3")]
        if let Some(a) = rt.udp_addrs().first() {
            let v = format!("h3=\":{}\"; ma=86400", a.port());
            let _ = shared.alt_svc.set((format!("alt-svc: {v}\r\n").into_bytes(), v.into_bytes()));
        }
        Ok(ServerHandle { rt, shared })
    }
}

/// A running server. Dropping it stops at once; use [`ServerHandle::stop`] and
/// [`ServerHandle::join`] for a graceful stop.
pub struct ServerHandle {
    rt: RuntimeHandle,
    shared: Arc<Shared>,
}

impl std::fmt::Debug for ServerHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ServerHandle").field("addrs", &self.rt.local_addrs()).field("backend", &self.rt.backend()).finish()
    }
}

impl ServerHandle {
    /// The first bound TCP address.
    pub fn local_addr(&self) -> SocketAddr {
        self.rt.local_addrs().first().or(self.rt.udp_addrs().first()).copied().unwrap_or_else(|| SocketAddr::from(([0, 0, 0, 0], 0)))
    }

    /// Every bound TCP address, in the order of the `listen` calls.
    pub fn local_addrs(&self) -> &[SocketAddr] {
        self.rt.local_addrs()
    }

    /// Every bound UDP (HTTP/3) address.
    pub fn quic_addrs(&self) -> &[SocketAddr] {
        self.rt.udp_addrs()
    }

    /// The I/O back end in use.
    pub fn backend(&self) -> Backend {
        self.rt.backend()
    }

    /// Why the requested back end was replaced, if it was (io_uring falling back to epoll).
    pub fn fallback_reason(&self) -> Option<&str> {
        self.rt.fallback_reason()
    }

    /// Current counters.
    pub fn stats(&self) -> Stats {
        let rt = self.rt.stats();
        let t = rt.total();
        let mut s = Stats { accepted: t.accepted, rejected: t.rejected, active: rt.active, offload: self.rt.offload().stats(), ..Stats::default() };
        let r = Ordering::Relaxed;
        for c in self.shared.counters.iter() {
            s.requests += c.requests.load(r);
            s.bad_requests += c.bad_requests.load(r);
            s.too_large += c.too_large.load(r);
            s.timeouts += c.timeouts.load(r);
            s.panics += c.panics.load(r);
            s.stalls += c.stalls.load(r);
            s.stall_checks += c.stall_checks.load(r);
            s.offloaded_routes += c.offloaded_routes.load(r);
            s.overloaded += c.overloaded.load(r);
            s.shed += c.shed.load(r);
            s.upgrades += c.upgrades.load(r);
            s.h2_connections += c.h2_connections.load(r);
            s.h3_connections += c.h3_connections.load(r);
            s.quic_endpoints += c.quic_endpoints.load(r);
            s.tls_errors += c.tls_errors.load(r);
        }
        s.runtime = rt;
        s
    }

    /// Starts a graceful stop: listeners close; idle HTTP/1.1 connections close at once and busy
    /// ones after their response (with `Connection: close`); HTTP/2 and HTTP/3 connections get
    /// GOAWAY and close when their streams finish. Whatever is still open after `grace` is
    /// aborted.
    pub fn stop(&self, grace: Duration) {
        self.rt.stop(grace);
    }

    /// Replaces the TLS configuration of the TLS listeners without a restart (certificate
    /// rotation): connections accepted from now on use `config`, open ones keep theirs. Its
    /// ALPN list is set as by [`Builder::tls`]. Session tickets issued under the old
    /// configuration resume only if `config` shares its `ticketer`. Fails when the server has
    /// no TLS listener.
    #[cfg(feature = "tls")]
    pub fn set_tls(&self, mut config: simoom_tls::ServerConfig) -> io::Result<()> {
        let Some(slot) = &self.shared.tls else { return Err(io::Error::new(io::ErrorKind::InvalidInput, "the server has no TLS listener")) };
        config.alpn = tcp_alpn(self.shared.http2);
        *slot.write().unwrap_or_else(|e| e.into_inner()) = Arc::new(config);
        Ok(())
    }

    /// Replaces the TLS configuration of the HTTP/3 listeners without a restart: QUIC
    /// connections that start from now on use `config` (ALPN `h3`), open ones keep theirs.
    /// Fails when the server has no HTTP/3 listener.
    #[cfg(feature = "http3")]
    pub fn set_quic_tls(&self, config: simoom_tls::ServerConfig) -> io::Result<()> {
        let Some(q) = &self.shared.quic_tls else { return Err(io::Error::new(io::ErrorKind::InvalidInput, "the server has no HTTP/3 listener")) };
        q.set(config);
        Ok(())
    }

    /// Waits for the loops to exit (after [`ServerHandle::stop`]).
    pub fn join(self) {
        self.rt.join();
    }
}