net_io/server/

mod.rs

//! The server: listeners, workers, the door, drain, and shutdown.

mod engine;
#[cfg(feature = "reactor")]
mod reactor;
mod source;

use std::any::TypeId;
use std::fmt;
use std::net::{SocketAddr, ToSocketAddrs};
use std::path::PathBuf;
use std::sync::Arc;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use crate::control::ControlLine;
use crate::driver::DriverKind;
use crate::error::{CloseReason, ConfigReason, KillReason, NetError, RefusalCause};
use crate::events::{EventSink, NoopSink};
use crate::gate::Gate;
use crate::handoff::ListenerHandle;
use crate::id::{ConnectionId, ListenerId, WorkerId};
use crate::limits::Limits;
use crate::opts::{ListenOptions, LocalOptions};
use crate::proxy::ProxyMode;
use crate::session::SessionFactory;
use crate::tcp::{ListenerInfo, TcpConnection};
use crate::time::Clock;
use crate::tls::ErasedAcceptor;
use crate::tls::{NoTls, TlsAcceptor};
use crate::trusted::TrustedProxies;

pub(crate) use engine::Shared;

/// Whether the server is accepting.
///
/// ```
/// assert_ne!(net_io::DoorState::Open, net_io::DoorState::Closed);
/// ```
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DoorState {
    /// Accepting new connections.
    Open,
    /// Not accepting; listening sockets stay open.
    Closed,
}

/// What a drain did.
///
/// ```
/// let r = net_io::DrainReport { graceful: 3, killed: 1, took: std::time::Duration::from_millis(40) };
/// assert_eq!(r.graceful + r.killed, 4);
/// ```
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DrainReport {
    /// Connections that finished before the deadline.
    pub graceful: u32,
    /// Connections killed at the deadline.
    pub killed: u32,
    /// How long the drain took.
    pub took: Duration,
}

/// Counters summed over the workers.
///
/// ```no_run
/// let server = net_io::serve("127.0.0.1:0", |_c| Ok(()))?;
/// let stats = server.stats();
/// println!("{} open, {} accepted", stats.open, stats.accepted);
/// # Ok::<(), net_io::NetError>(())
/// ```
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Stats {
    /// Connections open now.
    pub open: u64,
    /// Connections accepted.
    pub accepted: u64,
    /// Connections refused.
    pub refused: u64,
    /// Connections closed.
    pub closed: u64,
    /// Bytes received.
    pub bytes_in: u64,
    /// Bytes sent.
    pub bytes_out: u64,
    /// Writes the gate denied.
    pub egress_denied: u64,
    /// Connections killed.
    pub killed: u64,
    /// Connections closed for the minimum rate.
    pub slow_peers: u64,
    /// Connections closed by a timeout.
    pub timeouts: u64,
    /// Times the process ran out of descriptors.
    pub descriptor_pressure: u64,
    /// Completions for connections that had closed.
    pub stale_completions: u64,
    /// Sessions that panicked.
    pub session_panics: u64,
    refused_by: [u64; RefusalCause::COUNT],
    closed_by: [u64; CloseReason::COUNT],
}

impl Stats {
    /// Refusals with this cause (gate codes summed).
    #[must_use]
    pub fn refused_for(&self, cause: RefusalCause) -> u64 {
        self.refused_by[cause.index()]
    }

    /// Closes with this reason's variant (inner values ignored).
    #[must_use]
    pub fn closed_for(&self, reason: CloseReason) -> u64 {
        self.closed_by[reason.index()]
    }

    /// Every close reason with a non-zero count, one entry per variant (inner
    /// values are not kept: `Killed` carries reason 0, `Io` class `Other(0)`).
    ///
    /// ```no_run
    /// let server = net_io::serve("127.0.0.1:0", |_c| Ok(()))?;
    /// for (reason, n) in server.stats().closes() {
    ///     println!("{}: {n}", reason.as_str());
    /// }
    /// # Ok::<(), net_io::NetError>(())
    /// ```
    pub fn closes(&self) -> impl Iterator<Item = (CloseReason, u64)> + '_ {
        CloseReason::BY_INDEX
            .iter()
            .zip(self.closed_by.iter())
            .filter(|(_, n)| **n > 0)
            .map(|(r, n)| (*r, *n))
    }

    /// Every refusal cause with a non-zero count (gate codes summed under `Gate(0)`).
    pub fn refusals(&self) -> impl Iterator<Item = (RefusalCause, u64)> + '_ {
        RefusalCause::BY_INDEX
            .iter()
            .zip(self.refused_by.iter())
            .filter(|(_, n)| **n > 0)
            .map(|(r, n)| (*r, *n))
    }
}

/// One listener of a server.
///
/// ```
/// use net_io::{ListenerSpec, Limits};
/// let spec = ListenerSpec::tcp("127.0.0.1:0")
///     .name("admin")
///     .limits(Limits::builder().max_per_listener(10).build()?);
/// # let _ = spec;
/// # Ok::<(), net_io::NetError>(())
/// ```
#[must_use]
#[derive(Debug)]
pub struct ListenerSpec {
    pub(crate) addrs: Result<Vec<SocketAddr>, String>,
    pub(crate) options: Option<ListenOptions>,
    pub(crate) name: Option<String>,
    pub(crate) limits: Limits,
    pub(crate) proxy: ProxyMode,
    pub(crate) trusted: Option<Arc<TrustedProxies>>,
    pub(crate) tls: bool,
    pub(crate) unix: Option<(PathBuf, LocalOptions)>,
    #[cfg(all(windows, feature = "local"))]
    pub(crate) pipe: Option<(String, crate::pipe::PipeOptions)>,
    #[cfg(feature = "udp")]
    pub(crate) udp: Option<crate::udp::UdpOptions>,
    pub(crate) adopt: Option<ListenerHandle>,
    pub(crate) lockdown_allowed: bool,
}

impl ListenerSpec {
    /// A TCP listener on `addr` (resolved now; a failure is reported by `build`).
    pub fn tcp<A: ToSocketAddrs>(addr: A) -> ListenerSpec {
        ListenerSpec {
            addrs: addr
                .to_socket_addrs()
                .map(|i| i.collect())
                .map_err(|e| e.to_string()),
            options: None,
            name: None,
            limits: Limits::default(),
            proxy: ProxyMode::Off,
            trusted: None,
            tls: false,
            unix: None,
            #[cfg(all(windows, feature = "local"))]
            pipe: None,
            #[cfg(feature = "udp")]
            udp: None,
            adopt: None,
            lockdown_allowed: false,
        }
    }

    /// Keeps admitting under `ControlState::Lockdown`, for the listener an operator
    /// comes in through while everything else is refused.
    pub fn lockdown_allowed(mut self, on: bool) -> ListenerSpec {
        self.lockdown_allowed = on;
        self
    }

    /// A TCP listener with explicit options (their backlog wins over the limits').
    pub fn tcp_with<A: ToSocketAddrs>(addr: A, options: ListenOptions) -> ListenerSpec {
        ListenerSpec {
            options: Some(options),
            ..ListenerSpec::tcp(addr)
        }
    }

    /// A UDP socket served by the [`UdpSession`](crate::UdpSession) that
    /// `ServerBuilder::udp_sessions` makes (feature `udp`). A thread of its own reads
    /// it, with the door and a stop like any listener, and the socket's budgets and
    /// amplification guard apply.
    ///
    /// ```
    /// use net_io::{ListenerSpec, UdpOptions};
    /// let spec = ListenerSpec::udp("0.0.0.0:5353", UdpOptions::new());
    /// # let _ = spec;
    /// ```
    #[cfg(feature = "udp")]
    pub fn udp<A: ToSocketAddrs>(addr: A, options: crate::udp::UdpOptions) -> ListenerSpec {
        ListenerSpec {
            udp: Some(options),
            ..ListenerSpec::tcp(addr)
        }
    }

    /// A named pipe listener on `\\\\.\\pipe\\<name>` (feature `local`, Windows).
    /// Remote clients are always rejected, and connections carry `Transport::Pipe`,
    /// the client's pid, and no IP address: the per-source cap does not apply. With
    /// `proxy`, every peer is trusted: the pipe's security decides who may connect.
    ///
    /// ```no_run
    /// use net_io::{ListenerSpec, PipeOptions};
    /// let spec = ListenerSpec::pipe("myapp-control", PipeOptions::new());
    /// # let _ = spec;
    /// ```
    #[cfg(all(windows, feature = "local"))]
    pub fn pipe(name: &str, options: crate::pipe::PipeOptions) -> ListenerSpec {
        ListenerSpec {
            pipe: Some((name.to_owned(), options)),
            ..ListenerSpec::tcp(std::net::SocketAddr::from(([127, 0, 0, 1], 0)))
        }
    }

    /// A Unix domain socket listener at `path` (feature `local`; on Windows, AF_UNIX,
    /// where NTFS ACLs replace `options`' mode and owner). The socket file gets
    /// `options`' mode and owner before it listens, and is removed at shutdown if it
    /// is still the file this server created. Connections carry `Transport::Unix`, the peer's
    /// credentials, and no IP address: the per-source cap does not apply, and a
    /// `BlockingHandler` receives them as a `TcpConnection` whose `peer_addr` is
    /// unspecified. With `proxy`, every peer is trusted: the socket file's
    /// permissions decide who may connect.
    ///
    /// ```
    /// use net_io::{ListenerSpec, LocalOptions};
    /// let spec = ListenerSpec::unix("/run/myapp/api.sock", LocalOptions::new().group_access(33));
    /// # let _ = spec;
    /// ```
    pub fn unix(path: impl Into<PathBuf>, options: LocalOptions) -> ListenerSpec {
        ListenerSpec {
            addrs: Ok(Vec::new()),
            options: None,
            name: None,
            limits: Limits::default(),
            proxy: ProxyMode::Off,
            trusted: None,
            tls: false,
            unix: Some((path.into(), options)),
            lockdown_allowed: false,
            #[cfg(all(windows, feature = "local"))]
            pipe: None,
            #[cfg(feature = "udp")]
            udp: None,
            adopt: None,
        }
    }

    /// A listener this process was given: inherited from systemd or a FastCGI
    /// manager, received in a hand-off, or released by another server. Its name
    /// is the handle's unless [`ListenerSpec::name`] sets one. The socket is not
    /// bound again; options set at bind time stay as they are, and the backlog is
    /// reported as unknown. An adopted Unix socket's file is left in place at
    /// shutdown (the next bind removes it as stale).
    ///
    /// ```no_run
    /// use net_io::ListenerSpec;
    /// for inherited in net_io::systemd()? {
    ///     let spec = ListenerSpec::adopt(inherited.handle);
    ///     # let _ = spec;
    /// }
    /// # Ok::<(), net_io::InheritError>(())
    /// ```
    pub fn adopt(handle: ListenerHandle) -> ListenerSpec {
        ListenerSpec {
            addrs: Ok(Vec::new()),
            options: None,
            name: None,
            limits: Limits::default(),
            proxy: ProxyMode::Off,
            trusted: None,
            tls: false,
            unix: None,
            #[cfg(all(windows, feature = "local"))]
            pipe: None,
            #[cfg(feature = "udp")]
            udp: None,
            adopt: Some(handle),
            lockdown_allowed: false,
        }
    }

    /// A name (up to 64 bytes) for events and hand-off.
    pub fn name(mut self, name: &str) -> ListenerSpec {
        self.name = Some(name.to_owned());
        self
    }

    /// Per-listener caps and deadlines; values set here override the server's.
    pub fn limits(mut self, limits: Limits) -> ListenerSpec {
        self.limits = limits;
        self
    }

    /// Requires a PROXY protocol header on every connection (feature `proxy`).
    /// Peers outside the trusted proxies are closed before any byte is read.
    ///
    /// ```
    /// use net_io::{ListenerSpec, ProxyMode, ProxyVersions};
    /// let spec = ListenerSpec::tcp("127.0.0.1:0").proxy(ProxyMode::Required { versions: ProxyVersions::Both });
    /// # let _ = spec;
    /// ```
    pub fn proxy(mut self, mode: ProxyMode) -> ListenerSpec {
        self.proxy = mode;
        self
    }

    /// Serves TLS on this listener through the builder's acceptor
    /// ([`ServerBuilder::tls`], feature `tls`).
    pub fn tls(mut self, on: bool) -> ListenerSpec {
        self.tls = on;
        self
    }

    /// The proxies trusted on this listener, instead of the server's list.
    pub fn trusted_proxies(mut self, proxies: TrustedProxies) -> ListenerSpec {
        self.trusted = Some(Arc::new(proxies));
        self
    }
}

/// What runs per connection; the unset state (no `build` exists for it).
#[derive(Clone, Copy, Debug, Default)]
pub struct NoWork;

/// What `ServerBuilder::udp_sessions` produces when a server has no TCP work; programs
/// never name it.
#[cfg(feature = "udp")]
#[derive(Debug)]
pub struct UdpOnly;

/// The unset gate (no `build` exists for it): name `AllowAll` to run without a gate.
#[derive(Clone, Copy, Debug, Default)]
pub struct NoGate;

/// Where workers run.
///
/// ```
/// let pinned = net_io::Placement::Explicit(vec![0, 2, 4, 6]);
/// # let _ = pinned;
/// ```
#[non_exhaustive]
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub enum Placement {
    /// The OS decides.
    #[default]
    Off,
    /// Worker `n` on CPU `n`, wrapping around.
    PinPerCore,
    /// Worker `n` on `cpus[n % cpus.len()]`.
    Explicit(Vec<usize>),
}

/// What `ServerBuilder::sessions` produces; programs never name it.
#[derive(Debug)]
pub struct Sessions<F> {
    pub(crate) factory: F,
}

#[cfg(feature = "reactor")]
impl<F> Sessions<F> {
    pub(crate) fn factory(&self) -> &F {
        &self.factory
    }
}

/// What `ServerBuilder::handler` produces; programs never name it.
pub struct BlockingHandler<H> {
    pub(crate) handler: H,
}

impl<H> fmt::Debug for BlockingHandler<H> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("BlockingHandler")
    }
}

pub(crate) mod sealed {
    pub trait Sealed {}
}

/// What runs per connection: [`Sessions`] or a [`BlockingHandler`]. Sealed.
pub trait Workload: sealed::Sealed + engine::Serve + Send + Sync + 'static {}

#[cfg(feature = "udp")]
impl sealed::Sealed for UdpOnly {}
#[cfg(feature = "udp")]
impl Workload for UdpOnly {}

impl<F: SessionFactory> sealed::Sealed for Sessions<F> {}
impl<F: SessionFactory> Workload for Sessions<F> {}

impl<H> sealed::Sealed for BlockingHandler<H> where
    H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static
{
}
impl<H> Workload for BlockingHandler<H> where
    H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static
{
}

/// Builds a [`Server`]. A builder without a gate or without `sessions`/`handler`
/// has no `build` method: forgetting either is a compile error.
///
/// # Examples
///
/// ```no_run
/// use net_io::{AllowAll, DriverKind, Limits, ListenerSpec, ServerBuilder};
/// use std::time::{Duration, Instant};
///
/// let server = ServerBuilder::new()
///     .listener(ListenerSpec::tcp("0.0.0.0:8080").name("public"))
///     .listener(ListenerSpec::tcp("127.0.0.1:9090").name("admin"))
///     .driver(DriverKind::Blocking)
///     .workers(64)
///     .limits(Limits::builder().idle_timeout(Duration::from_secs(5)).build()?)
///     .handler(|conn| { conn.send_all(b"hello\n")?; Ok(()) })
///     .gate(AllowAll)
///     .build()?;              // listeners bound, workers started, door closed
/// // ... drop privileges here ...
/// server.open_door()?;
/// let report = server.shutdown(Instant::now() + Duration::from_secs(30));
/// println!("{report:?}");
/// # Ok::<(), net_io::NetError>(())
/// ```
#[must_use]
pub struct ServerBuilder<W = NoWork, G = NoGate, S = NoopSink, T = NoTls> {
    listeners: Vec<ListenerSpec>,
    driver: DriverKind,
    workers: Option<u16>,
    limits: Limits,
    work: W,
    gate: G,
    sink: S,
    tls: T,
    control: Option<Arc<dyn ControlLine>>,
    wipe: bool,
    clock: Option<Arc<dyn Clock>>,
    trusted: Option<Arc<TrustedProxies>>,
    placement: Placement,
    #[cfg(feature = "udp")]
    udp: Option<Arc<dyn crate::udp::UdpSessionFactory>>,
}

impl ServerBuilder {
    /// An empty builder.
    pub fn new() -> ServerBuilder {
        ServerBuilder {
            listeners: Vec::new(),
            driver: DriverKind::Auto,
            workers: None,
            limits: Limits::default(),
            work: NoWork,
            gate: NoGate,
            sink: NoopSink,
            tls: NoTls,
            control: None,
            wipe: true,
            clock: None,
            trusted: None,
            placement: Placement::Off,
            #[cfg(feature = "udp")]
            udp: None,
        }
    }
}

impl Default for ServerBuilder {
    fn default() -> ServerBuilder {
        ServerBuilder::new()
    }
}

impl<W, G, S, T> ServerBuilder<W, G, S, T> {
    fn map<W2, G2, S2, T2>(
        self,
        f: impl FnOnce(W, G, S, T) -> (W2, G2, S2, T2),
    ) -> ServerBuilder<W2, G2, S2, T2> {
        let ServerBuilder {
            listeners,
            driver,
            workers,
            limits,
            work,
            gate,
            sink,
            tls,
            control,
            wipe,
            clock,
            trusted,
            placement,
            #[cfg(feature = "udp")]
            udp,
        } = self;
        let (work, gate, sink, tls) = f(work, gate, sink, tls);
        ServerBuilder {
            listeners,
            driver,
            workers,
            limits,
            work,
            gate,
            sink,
            tls,
            control,
            wipe,
            clock,
            trusted,
            placement,
            #[cfg(feature = "udp")]
            udp,
        }
    }

    /// The proxies whose PROXY headers are believed, for every listener that
    /// requires one and has no list of its own.
    pub fn trusted_proxies(mut self, proxies: TrustedProxies) -> Self {
        self.trusted = Some(Arc::new(proxies));
        self
    }

    /// Adds a listener (at most 64).
    pub fn listener(mut self, spec: ListenerSpec) -> Self {
        self.listeners.push(spec);
        self
    }

    /// The driver (default `Auto`).
    pub fn driver(mut self, kind: DriverKind) -> Self {
        self.driver = kind;
        self
    }

    /// Pins workers to CPUs (`Unsupported` on macOS). A worker that stays on one CPU
    /// keeps its cache; the cost is that the OS can no longer move it off a busy CPU.
    pub fn placement(mut self, placement: Placement) -> Self {
        self.placement = placement;
        self
    }

    /// Worker threads (default 128 on the blocking driver; 1 to 4,096).
    pub fn workers(mut self, n: u16) -> Self {
        self.workers = Some(n);
        self
    }

    /// The server's limits; unset values take the driver's defaults.
    pub fn limits(mut self, limits: Limits) -> Self {
        self.limits = limits;
        self
    }

    /// Runs a sans-IO session per connection.
    pub fn sessions<F: SessionFactory>(self, factory: F) -> ServerBuilder<Sessions<F>, G, S, T> {
        self.map(|_, g, s, t| (Sessions { factory }, g, s, t))
    }

    /// Runs a blocking handler per connection (on the blocking driver).
    pub fn handler<H>(self, handler: H) -> ServerBuilder<BlockingHandler<H>, G, S, T>
    where
        H: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static,
    {
        self.map(|_, g, s, t| (BlockingHandler { handler }, g, s, t))
    }

    /// The TLS engine for listeners with `ListenerSpec::tls(true)` (feature `tls`).
    pub fn tls<T2: TlsAcceptor>(self, acceptor: T2) -> ServerBuilder<W, G, S, T2> {
        self.map(|w, g, s, _| (w, g, s, acceptor))
    }

    /// The connection gate (required; `AllowAll` for none).
    pub fn gate<G2: Gate>(self, gate: G2) -> ServerBuilder<W, G2, S, T> {
        self.map(|w, _, s, t| (w, gate, s, t))
    }

    /// The event sink (default `NoopSink`).
    pub fn events<S2: EventSink>(self, sink: S2) -> ServerBuilder<W, G, S2, T> {
        self.map(|w, g, _, t| (w, g, sink, t))
    }

    /// The control line (default: none).
    pub fn control<C: ControlLine>(mut self, line: Arc<C>) -> Self {
        self.control = Some(line);
        self
    }

    /// Zero the used part of every buffer when it returns to a pool (default on).
    pub fn wipe_buffers(mut self, on: bool) -> Self {
        self.wipe = on;
        self
    }

    /// The clock deadlines read (tests inject a `ManualClock`).
    pub fn clock<K: Clock>(mut self, clock: Arc<K>) -> Self {
        self.clock = Some(clock);
        self
    }
}

#[cfg(feature = "udp")]
impl<G, S, T> ServerBuilder<NoWork, G, S, T> {
    /// Serves UDP listeners only; a TCP listener then fails `build`. One thread per
    /// listener, and no worker pool.
    ///
    /// ```no_run
    /// use std::net::SocketAddr;
    /// use net_io::{AllowAll, ListenerInfo, ListenerSpec, ServerBuilder, UdpContext, UdpOptions, UdpSession};
    /// struct Echo;
    /// impl UdpSession for Echo {
    ///     fn on_datagram(&mut self, cx: &mut UdpContext<'_>, from: SocketAddr, bytes: &[u8]) {
    ///         let _ = cx.send_to(bytes, from);
    ///     }
    /// }
    /// let server = ServerBuilder::new()
    ///     .listener(ListenerSpec::udp("0.0.0.0:7000", UdpOptions::new()))
    ///     .udp_sessions(|_: &ListenerInfo| Echo)
    ///     .gate(AllowAll)
    ///     .build()?;
    /// server.open_door()?;
    /// # Ok::<(), net_io::NetError>(())
    /// ```
    pub fn udp_sessions<F>(self, factory: F) -> ServerBuilder<UdpOnly, G, S, T>
    where
        F: crate::udp::UdpSessionFactory,
    {
        let mut b = self.map(|_, g, s, t| (UdpOnly, g, s, t));
        b.udp = Some(Arc::new(factory));
        b.workers.get_or_insert(1);
        b
    }
}

#[cfg(feature = "udp")]
impl<W: Workload, G, S, T> ServerBuilder<W, G, S, T> {
    /// Serves the UDP listeners beside this server's TCP work.
    pub fn udp_sessions<F>(mut self, factory: F) -> Self
    where
        F: crate::udp::UdpSessionFactory,
    {
        self.udp = Some(Arc::new(factory));
        self
    }
}

impl<W, G, S, T> ServerBuilder<W, G, S, T>
where
    W: Workload,
    G: Gate,
    S: EventSink,
    T: TlsAcceptor,
{
    /// Validates everything, binds the listeners, starts the workers, and returns
    /// the server with the door **closed**.
    ///
    /// # Errors
    ///
    /// `InvalidConfig` (`field` names the method), bind errors, `Unsupported` for a
    /// driver this version does not have.
    pub fn build(self) -> Result<Server, NetError> {
        // A handler holds its worker for the whole connection (blocking pool only).
        let requested = match self.driver {
            DriverKind::Auto if <W as engine::Serve>::BLOCKING_ONLY => DriverKind::Blocking,
            other => other,
        };
        let driver = match requested.resolve() {
            DriverKind::Reactor if <W as engine::Serve>::BLOCKING_ONLY => {
                return Err(invalid("driver", ConfigReason::Inconsistent));
            }
            DriverKind::Reactor if !DriverKind::reactor_built() => {
                return Err(NetError::Unsupported {
                    feature: "the reactor driver",
                    platform: std::env::consts::OS,
                });
            }
            DriverKind::Uring => {
                return Err(NetError::Unsupported {
                    feature: "the io_uring driver (after 1.0.0)",
                    platform: std::env::consts::OS,
                });
            }
            kind => kind,
        };
        if self.listeners.is_empty() {
            return Err(invalid("listener", ConfigReason::Missing));
        }
        if self.listeners.len() > 64 {
            return Err(invalid("listener", ConfigReason::TooLarge { max: 64 }));
        }
        let workers = self.workers.unwrap_or_else(|| {
            if driver == DriverKind::Reactor {
                std::thread::available_parallelism()
                    .map_or(4, |n| u16::try_from(n.get()).unwrap_or(4096).min(4096))
            } else {
                128
            }
        });
        if workers == 0 {
            return Err(invalid("workers", ConfigReason::Zero));
        }
        if workers > 4096 {
            return Err(invalid("workers", ConfigReason::TooLarge { max: 4096 }));
        }
        let mut listeners = self.listeners;
        let has_tls = TypeId::of::<T>() != TypeId::of::<NoTls>();
        let mut names = Vec::new();
        #[cfg(feature = "udp")]
        for spec in &listeners {
            if (spec.udp.is_some() && self.udp.is_none())
                || (spec.udp.is_none() && <W as engine::Serve>::UDP_ONLY)
            {
                return Err(invalid("udp_sessions", ConfigReason::Inconsistent));
            }
            if spec.udp.is_some() && (spec.tls || spec.proxy != ProxyMode::Off) {
                return Err(invalid("udp", ConfigReason::Inconsistent));
            }
        }
        for spec in &mut listeners {
            check_spec(spec, self.trusted.as_ref(), has_tls, &names)?;
            if let Some(n) = &spec.name {
                names.push(n.clone());
            }
        }
        // On the blocking driver an unset global cap is one connection per worker;
        // on the reactor the default, but never more than the descriptors allow.
        let limits = if self.limits.max_connections().is_none() {
            let cap = if driver == DriverKind::Reactor {
                reactor_default_cap(workers)
            } else {
                u32::from(workers)
            };
            self.limits.to_builder().max_connections(cap).build()?
        } else {
            self.limits.clone()
        };
        limits.validate(workers, crate::sys::descriptor_limit())?;
        for spec in &listeners {
            limits.overlay(&spec.limits).validate(workers, None)?;
        }
        if self.placement != Placement::Off && !crate::sys::CAN_PIN {
            return Err(NetError::Unsupported {
                feature: "worker placement",
                platform: std::env::consts::OS,
            });
        }
        if self.placement == Placement::Explicit(Vec::new()) {
            return Err(invalid("placement", ConfigReason::Zero));
        }
        let tls: Option<Arc<dyn ErasedAcceptor>> = if has_tls {
            Some(Arc::new(self.tls))
        } else {
            None
        };
        engine::start(engine::Config {
            tls,
            trusted: self.trusted,
            listeners,
            driver,
            workers,
            limits,
            work: self.work,
            gate: self.gate,
            sink: self.sink,
            control: self.control,
            wipe: self.wipe,
            clock: self.clock,
            placement: self.placement,
            #[cfg(feature = "udp")]
            udp: self.udp,
        })
    }
}

/// The reactor's default global cap: 10,000, less where the process's descriptor
/// limit would not hold it.
fn reactor_default_cap(workers: u16) -> u32 {
    let wanted = 10_000u32;
    match crate::sys::descriptor_limit() {
        Some(have) => {
            let spare = u64::from(workers).saturating_mul(2) + 64;
            let room = have.saturating_sub(spare).clamp(16, u64::from(wanted));
            u32::try_from(room).unwrap_or(wanted)
        }
        None => wanted,
    }
}

fn invalid(field: &'static str, reason: ConfigReason) -> NetError {
    NetError::InvalidConfig { field, reason }
}

/// Where `Server::adopt` registers a connection: the listener (by name) whose
/// limits and caps apply, and an optional starting lane (the gate may change it).
///
/// ```
/// let spec = net_io::AdoptSpec::new("main").lane(net_io::Lane(1));
/// # let _ = spec;
/// ```
#[must_use]
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct AdoptSpec {
    listener: String,
    lane: Option<crate::info::Lane>,
}

impl AdoptSpec {
    /// Registers on the listener named `listener`.
    pub fn new(listener: &str) -> AdoptSpec {
        AdoptSpec {
            listener: listener.to_owned(),
            lane: None,
        }
    }

    /// The lane the connection starts in, before the gate's admission.
    pub fn lane(mut self, lane: crate::info::Lane) -> AdoptSpec {
        self.lane = Some(lane);
        self
    }
}

/// The checks `build` and `add_listener` run on one listener: the features it
/// needs, a TLS engine, trusted proxies (resolved into the spec), and its name.
fn check_spec(
    spec: &mut ListenerSpec,
    server_trusted: Option<&Arc<TrustedProxies>>,
    has_tls: bool,
    taken: &[String],
) -> Result<(), NetError> {
    if spec.name.is_none() {
        spec.name = spec
            .adopt
            .as_ref()
            .and_then(|h| h.name())
            .map(str::to_owned);
    }
    let unix = spec.unix.is_some()
        || spec
            .adopt
            .as_ref()
            .is_some_and(|h| h.kind() == crate::info::Transport::Unix);
    if unix && !cfg!(feature = "local") {
        return Err(NetError::Unsupported {
            feature: "Unix domain sockets (the `local` feature)",
            platform: std::env::consts::OS,
        });
    }
    if spec.tls {
        if !cfg!(feature = "tls") {
            return Err(NetError::Unsupported {
                feature: "TLS (the `tls` feature)",
                platform: std::env::consts::OS,
            });
        }
        if !has_tls {
            return Err(invalid("tls", ConfigReason::Missing));
        }
    }
    if spec.proxy != ProxyMode::Off {
        if !cfg!(feature = "proxy") {
            return Err(NetError::Unsupported {
                feature: "PROXY protocol (the `proxy` feature)",
                platform: std::env::consts::OS,
            });
        }
        // On a Unix socket the file's permissions decide who may connect.
        if !unix {
            let trusted = spec.trusted.clone().or_else(|| server_trusted.cloned());
            match trusted {
                Some(t) if !t.is_empty() => spec.trusted = Some(t),
                // Required without anyone to trust would refuse every connection.
                _ => return Err(invalid("trusted_proxies", ConfigReason::Missing)),
            }
        }
    }
    if let Some(n) = &spec.name {
        if n.len() > 64 {
            return Err(invalid("name", ConfigReason::TooLarge { max: 64 }));
        }
        if taken.contains(n) {
            return Err(invalid("name", ConfigReason::Inconsistent));
        }
    }
    Ok(())
}

/// A running server. Dropping it stops it (with a drain), so keep it and call
/// [`Server::wait_for_stop`] to run until a stop is requested.
#[must_use = "dropping a Server stops it; call wait_for_stop() to run until a stop is requested"]
pub struct Server {
    pub(crate) shared: Arc<Shared>,
    pub(crate) threads: Vec<JoinHandle<()>>,
    pub(crate) listeners: Vec<ListenerInfo>,
    pub(crate) stopped: bool,
}

impl Server {
    /// Starts accepting on every listener; a second call is a no-op.
    ///
    /// # Errors
    ///
    /// `ShuttingDown` while draining or after shutdown; `ForkedChild`.
    pub fn open_door(&self) -> Result<(), NetError> {
        self.shared.open_door()
    }

    /// Stops accepting within one loop iteration; listening sockets stay open. Idempotent.
    pub fn close_door(&self) {
        self.shared.close_door();
    }

    /// Whether the server is accepting.
    #[must_use]
    pub fn door(&self) -> DoorState {
        self.shared.door()
    }

    /// Closes the door, tells sessions to finish, closes idle connections at once,
    /// waits for the rest until `deadline`, then kills them with `DrainDeadline`.
    pub fn drain(&self, deadline: Instant) -> DrainReport {
        self.shared.drain(deadline)
    }

    /// Drains, then closes the listening sockets and joins every thread.
    pub fn shutdown(mut self, deadline: Instant) -> DrainReport {
        self.stop(deadline)
    }

    /// Blocks until a stop is requested (`ServerHandle::request_stop`, or the
    /// control line moves to `Draining` or `Stopping`), then shuts down.
    pub fn wait_for_stop(mut self) -> DrainReport {
        let deadline = self.shared.wait_for_stop_request();
        self.stop(deadline)
    }

    fn stop(&mut self, deadline: Instant) -> DrainReport {
        if self.stopped {
            return DrainReport {
                graceful: 0,
                killed: 0,
                took: Duration::ZERO,
            };
        }
        self.stopped = true;
        let report = self.shared.drain(deadline);
        self.shared.stop_threads();
        for t in self.threads.drain(..) {
            let _ = t.join();
        }
        self.shared.join_listeners();
        report
    }

    /// Registers a dialed connection exactly like an accepted one on the
    /// listener named in `spec`: a new id, the caps, `Gate::admit` with
    /// `AdmitStage::Adopt`, that listener's limits, the server's handler or
    /// sessions, its kill handle, and its events. For back-channels and outbound
    /// connections a server should run and revoke like its own. A gate the
    /// connection had from `set_gate` is closed (`SessionClosed`) first.
    ///
    /// # Errors
    ///
    /// `Refused { cause }` (caps, the gate, `Busy`; the connection was reset),
    /// `InvalidConfig { field: "listener" }` (no listener with that name),
    /// `InvalidConfig { field: "connection" }` (a TLS connection, or bytes read
    /// ahead), `ShuttingDown`, `ForkedChild`, `Closed`.
    ///
    /// ```no_run
    /// use net_io::{AdoptSpec, AllowAll, ListenerSpec, ServerBuilder};
    /// let server = ServerBuilder::new()
    ///     .listener(ListenerSpec::tcp("127.0.0.1:7000").name("main"))
    ///     .handler(|conn| { conn.send_all(b"hello, upstream\n")?; Ok(()) })
    ///     .gate(AllowAll)
    ///     .build()?;
    /// let upstream = net_io::connect("10.0.0.9:9000")?;
    /// let id = server.adopt(upstream, AdoptSpec::new("main"))?;
    /// # let _ = id;
    /// # Ok::<(), net_io::NetError>(())
    /// ```
    pub fn adopt(&self, conn: TcpConnection, spec: AdoptSpec) -> Result<ConnectionId, NetError> {
        let mut conn = conn;
        let link = conn.take_link().ok_or(NetError::Closed)?;
        let (Some(peer), Some(local)) =
            (link.info.peer.socket_addr(), link.info.local.socket_addr())
        else {
            return Err(invalid("connection", ConfigReason::Inconsistent));
        };
        let stream = link
            .into_stream(CloseReason::SessionClosed)
            .and_then(crate::sys::Stream::into_tcp)
            .ok_or(invalid("connection", ConfigReason::Inconsistent))?;
        self.shared
            .adopt(stream, peer, local, &spec.listener, spec.lane)
    }

    /// Owned duplicates of every listening socket, for a hand-off to the next
    /// process (`send_listeners`, Unix). This server keeps
    /// accepting meanwhile, so the new process can start before this one stops
    /// (decision D-042); a later `shutdown` closes only this server's copies, and
    /// a Unix socket file stays in place for the new process.
    ///
    /// # Errors
    ///
    /// `Os` when a socket cannot be duplicated, `ForkedChild`.
    pub fn release_listeners(&self) -> Result<Vec<ListenerHandle>, NetError> {
        self.shared.release_listeners()
    }

    /// Binds one more listener on the running server and starts its accept
    /// thread. Its door is closed until the next [`Server::open_door`] (which
    /// opens every listener's door); the same checks as `build` apply. The id is
    /// new: ids of removed listeners are never reused.
    ///
    /// # Errors
    ///
    /// As `build` for the listener; `InvalidConfig { field: "listener" }` past 64
    /// listeners, `ShuttingDown`, `ForkedChild`.
    ///
    /// ```no_run
    /// use net_io::{AllowAll, ListenerSpec, ServerBuilder};
    /// let mut server = ServerBuilder::new()
    ///     .listener(ListenerSpec::tcp("127.0.0.1:8080"))
    ///     .handler(|conn| { conn.send_all(b"hi\n")?; Ok(()) })
    ///     .gate(AllowAll)
    ///     .build()?;
    /// server.open_door()?;
    /// let added = server.add_listener(ListenerSpec::tcp("127.0.0.1:8081").name("extra"))?;
    /// server.open_door()?; // opens the new listener's door
    /// # let _ = added;
    /// # Ok::<(), net_io::NetError>(())
    /// ```
    pub fn add_listener(&mut self, spec: ListenerSpec) -> Result<ListenerId, NetError> {
        let mut spec = spec;
        if self.listeners.len() >= 64 {
            return Err(invalid("listener", ConfigReason::TooLarge { max: 64 }));
        }
        let names: Vec<String> = self
            .listeners
            .iter()
            .filter_map(|l| l.name.clone())
            .collect();
        check_spec(
            &mut spec,
            self.shared.trusted.as_ref(),
            self.shared.tls.is_some(),
            &names,
        )?;
        let workers = u16::try_from(self.shared.worker_count()).unwrap_or(u16::MAX);
        self.shared
            .base_limits
            .overlay(&spec.limits)
            .validate(workers, None)?;
        let info = self.shared.add_listener(spec)?;
        let id = info.id;
        self.listeners.push(info);
        Ok(id)
    }

    /// Closes one listener's socket at once (its connections keep running),
    /// then drains them: idle ones close now, the rest may finish until
    /// `deadline` and are then killed (`DrainDeadline`). Other listeners and
    /// their connections are not touched.
    ///
    /// # Errors
    ///
    /// `InvalidConfig { field: "listener", reason: Missing }` for an unknown id,
    /// `ForkedChild`.
    pub fn remove_listener(
        &mut self,
        id: ListenerId,
        deadline: Instant,
    ) -> Result<DrainReport, NetError> {
        let report = self.shared.remove_listener(id, deadline)?;
        self.listeners.retain(|l| l.id != id);
        Ok(report)
    }

    /// A cloneable handle for other threads.
    #[must_use]
    pub fn handle(&self) -> ServerHandle {
        ServerHandle {
            shared: self.shared.clone(),
        }
    }

    /// Counters summed over the workers.
    #[must_use]
    pub fn stats(&self) -> Stats {
        self.shared.stats()
    }

    /// Facts about each listener.
    #[must_use]
    pub fn listeners(&self) -> &[ListenerInfo] {
        &self.listeners
    }
}

impl Drop for Server {
    fn drop(&mut self) {
        let deadline = Instant::now() + self.shared.drain_timeout;
        self.stop(deadline);
    }
}

impl fmt::Debug for Server {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Server")
            .field("listeners", &self.listeners)
            .field("door", &self.door())
            .finish()
    }
}

/// A cloneable handle to a running server, safe to use from any thread.
///
/// ```no_run
/// let server = net_io::serve("127.0.0.1:7000", |conn| { conn.send_all(b"hi\n")?; Ok(()) })?;
/// let handle = server.handle();
/// std::thread::spawn(move || {
///     // ... wait for a signal from the application's signal library ...
///     handle.request_stop();
/// });
/// let report = server.wait_for_stop();
/// println!("drained: {} graceful, {} killed in {:?}", report.graceful, report.killed, report.took);
/// # Ok::<(), net_io::NetError>(())
/// ```
#[derive(Clone)]
pub struct ServerHandle {
    pub(crate) shared: Arc<Shared>,
}

impl ServerHandle {
    /// Aborts the connection `id`; `false` if it already closed or belongs to another server.
    pub fn kill(&self, id: ConnectionId, reason: KillReason) -> bool {
        self.shared.kill(id, reason)
    }

    /// Stops accepting.
    pub fn close_door(&self) {
        self.shared.close_door();
    }

    /// Makes `Server::wait_for_stop` return after the drain. Does not block.
    pub fn request_stop(&self) {
        self.shared.request_stop(None);
    }

    /// Interrupts every worker's and acceptor's wait so a control-line change is seen at once.
    pub fn wake_all(&self) {
        self.shared.wake_all();
    }

    /// The completion queue of `worker`.
    #[must_use]
    pub fn completer(&self, worker: WorkerId) -> Option<crate::completer::Completer> {
        self.shared.completer(worker)
    }

    /// Counters summed over the workers.
    #[must_use]
    pub fn stats(&self) -> Stats {
        self.shared.stats()
    }

    /// Whether the server is accepting.
    #[must_use]
    pub fn door(&self) -> DoorState {
        self.shared.door()
    }
}

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

impl Stats {
    pub(crate) fn from_parts(
        mut base: Stats,
        refused_by: [u64; RefusalCause::COUNT],
        closed_by: [u64; CloseReason::COUNT],
    ) -> Stats {
        base.refused_by = refused_by;
        base.closed_by = closed_by;
        base
    }
}