sozu_lib/

lib.rs

//! ## What this library does
//!
//! This library provides tools to build and start HTTP, HTTPS and TCP reverse proxies.
//!
//! The proxies handles network polling, HTTP parsing, TLS in a fast single threaded event
//! loop.
//!
//! Each proxy is designed to receive configuration changes at runtime instead of
//! reloading from a file regularly. The event loop runs in its own thread
//! and receives commands through a message queue.
//!
//! ## Difference with the crate `sozu`
//!
//! To create several workers and manage them all at once (which is the most common way to
//! use Sōzu), the crate `sozu` is more indicated than using the lib directly.
//!
//! The crate `sozu` provides a binary called the main process.
//! The main process uses `sozu_lib` to start and manage workers.
//! Each worker can handle HTTP, HTTPS and TCP traffic.
//! The main process receives synchronizes the state of all workers, using UNIX sockets
//! and custom channels to communicate with them.
//! The main process itself is is configurable with a file, and has a CLI.
//!
//! ## How to use this library directly
//!
//! This documentation here explains how to write a binary that will start a single Sōzu
//! worker and give it orders. The method has two steps:
//!
//! 1. Starts a Sōzu worker in a distinct thread
//! 2. sends instructions to the worker on a UNIX socket via a Sōzu channel
//!
//! ### How to start a Sōzu worker
//!
//! Before creating an HTTP proxy, we first need to create an HTTP listener.
//! The listener is an abstraction around a TCP socket provided by the kernel.
//! We need the `sozu_command_lib` to build a listener.
//!
//! ```
//! use sozu_command_lib::{config::ListenerBuilder, proto::command::SocketAddress};
//!
//! let address = SocketAddress::new_v4(127,0,0,1,8080);
//! let http_listener = ListenerBuilder::new_http(address)
//!     .to_http(None)
//!     .expect("Could not create HTTP listener");
//! ```
//!
//! The `http_listener` is of the type `HttpListenerConfig`, that we can be sent to the worker
//! to start the proxy.
//!
//! Then create a pair of channels to communicate with the proxy.
//! The channel is a wrapper around a unix socket.
//!
//! ```ignore
//! use sozu_command_lib::{
//!     channel::Channel,
//!     proto::command::{WorkerRequest, WorkerResponse},
//! };
//!
//! let (mut command_channel, proxy_channel): (
//!     Channel<WorkerRequest, WorkerResponse>,
//!     Channel<WorkerResponse, WorkerRequest>,
//! ) = Channel::generate(1000, 10000).expect("should create a channel");
//!```
//!
//! Here, the `command_channel` end is blocking, it sends `WorkerRequest`s and receives
//! `WorkerResponses`, while the `proxy_channel` end is non-blocking, and the types are reversed.
//! Writing the types here isn't even necessary thanks to the compiler,
//! but it brings the point accross.
//!
//! You can now launch the worker in a separate thread, providing the HTTP listener config,
//! the proxy end of the channel, and your custom number of buffers and their size:
//!
//! ```ignore
//! use std::thread;
//!
//! let worker_thread_join_handle = thread::spawn(move || {
//!     let max_buffers = 500;
//!     let buffer_size = 16384;
//!     sozu_lib::http::testing::start_http_worker(http_listener, proxy_channel, max_buffers, buffer_size);
//! });
//! ```
//!
//! ### Send orders
//!
//! Once the thread is launched, the proxy worker will start its event loop and handle
//! events on the listening interface and port specified when building the HTTP Listener.
//! Since no frontends or backends were specified for the proxy, it will receive
//! the connections, parse the requests, then send a default (but configurable)
//! answer.
//!
//! Before defining a frontend and backends, we need to define a cluster, which describes
//! a routing configuration. A cluster contains:
//!
//! - one frontend
//! - one or several backends
//! - routing rules
//!
//! A cluster is identified by its `cluster_id`, which will be used to define frontends
//! and backends later on.
//!
//! ```
//! use sozu_command_lib::proto::command::{Cluster, LoadBalancingAlgorithms};
//!
//! let cluster = Cluster {
//!     cluster_id: "my-cluster".to_string(),
//!     sticky_session: false,
//!     https_redirect: false,
//!     load_balancing: LoadBalancingAlgorithms::RoundRobin as i32,
//!     answer_503: Some("A custom forbidden message".to_string()),
//!     ..Default::default()
//! };
//! ```
//!
//! The defaults are sensible, so we could define only the `cluster_id`.
//!
//! We can now define a frontend. A frontend is a way to recognize a request and match
//! it to a `cluster_id`, depending on the hostname and the beginning of the URL path.
//! The `address` field must match the one of the HTTP listener we defined before:
//!
//! ```
//! use std::collections::BTreeMap;
//!
//!  use sozu_command_lib::proto::command::{PathRule, RequestHttpFrontend, RulePosition, SocketAddress};
//!
//! let http_front = RequestHttpFrontend {
//!     cluster_id: Some("my-cluster".to_string()),
//!     address: SocketAddress::new_v4(127,0,0,1,8080),
//!     hostname: "example.com".to_string(),
//!     path: PathRule::prefix(String::from("/")),
//!     position: RulePosition::Pre.into(),
//!     tags: BTreeMap::from([
//!         ("owner".to_owned(), "John".to_owned()),
//!         ("id".to_owned(), "my-own-http-front".to_owned()),
//!     ]),
//!     ..Default::default()
//! };
//! ```
//!
//! The `tags` are keys and values that will appear in the access logs,
//! which can come in handy.
//!
//! Now let's define a backend.
//! A backend is an instance of a backend application we want to route traffic to.
//! The `address` field must match the IP and port of the backend server.
//!
//! ```
//! use sozu_command_lib::proto::command::{AddBackend, LoadBalancingParams, SocketAddress};
//!
//! let http_backend = AddBackend {
//!     cluster_id: "my-cluster".to_string(),
//!     backend_id: "test-backend".to_string(),
//!     address: SocketAddress::new_v4(127,0,0,1,8000),
//!     load_balancing_parameters: Some(LoadBalancingParams::default()),
//!     ..Default::default()
//! };
//! ```
//!
//! A cluster can have multiple backend servers, and they can be added or
//! removed while the proxy is running. If a backend is removed from the configuration
//! while the proxy is handling a request to that server, it will finish that
//! request and stop sending new traffic to that server.
//!
//!
//! Now we can use the other end of the channel to send all these requests to the worker,
//! using the WorkerRequest type:
//!
//! ```ignore
//! use sozu_command_lib::{
//!     proto::command::{Request, request::RequestType, WorkerRequest},
//! };
//!
//! command_channel
//!     .write_message(&WorkerRequest {
//!         id: String::from("add-the-cluster"),
//!         content: RequestType::AddCluster(cluster).into(),
//!     })
//!     .expect("Could not send AddHttpFrontend request");
//!
//! command_channel
//!     .write_message(&WorkerRequest {
//!         id: String::from("add-the-frontend"),
//!         content: RequestType::AddHttpFrontend(http_front).into(),
//!     })
//!     .expect("Could not send AddHttpFrontend request");
//!
//! command_channel
//!     .write_message(&WorkerRequest {
//!         id: String::from("add-the-backend"),
//!         content: RequestType::AddBackend(http_backend).into(),
//!     })
//!     .expect("Could not send AddBackend request");
//!
//! println!("HTTP -> {:?}", command_channel.read_message());
//! println!("HTTP -> {:?}", command_channel.read_message());
//! println!("HTTP -> {:?}", command_channel.read_message());
//! ```
//!
//!
//! The event loop of the worker will process these instructions and add them to
//! its state, and the worker will send back an acknowledgement
//! message.
//!
//! Now we can let the worker thread run in the background:
//!
//! ```ignore
//! let _ = worker_thread_join_handle.join();
//! ```
//!
//! Here is the complete example for reference, it matches the `examples/http.rs` example:
//!
//! ```
//! #[macro_use]
//! extern crate sozu_command_lib;
//!
//! use std::{collections::BTreeMap, env, io::stdout, thread};
//!
//! use anyhow::Context;
//! use sozu_command_lib::{
//!     channel::Channel,
//!     config::ListenerBuilder,
//!     logging::setup_default_logging,
//!     proto::command::{
//!         request::RequestType, AddBackend, Cluster, LoadBalancingAlgorithms, LoadBalancingParams,
//!         PathRule, Request, RequestHttpFrontend, RulePosition, SocketAddress,WorkerRequest,
//!     },
//! };
//!
//! fn main() -> anyhow::Result<()> {
//!     setup_default_logging(true, "info", "EXAMPLE").with_context(|| "could not setup logging")?;
//!
//!     info!("starting up");
//!
//!     let http_listener = ListenerBuilder::new_http(SocketAddress::new_v4(127,0,0,1,8080))
//!         .to_http(None)
//!         .expect("Could not create HTTP listener");
//!
//!     let (mut command_channel, proxy_channel) =
//!         Channel::generate(1000, 10000).with_context(|| "should create a channel")?;
//!
//!     let worker_thread_join_handle = thread::spawn(move || {
//!         let max_buffers = 500;
//!         let buffer_size = 16384;
//!         sozu_lib::http::testing::start_http_worker(http_listener, proxy_channel, max_buffers, buffer_size)
//!             .expect("The worker could not be started, or shut down");
//!     });
//!
//!     let cluster = Cluster {
//!         cluster_id: "my-cluster".to_string(),
//!         sticky_session: false,
//!         https_redirect: false,
//!         load_balancing: LoadBalancingAlgorithms::RoundRobin as i32,
//!         answer_503: Some("A custom forbidden message".to_string()),
//!         ..Default::default()
//!     };
//!
//!     let http_front = RequestHttpFrontend {
//!         cluster_id: Some("my-cluster".to_string()),
//!         address: SocketAddress::new_v4(127,0,0,1,8080),
//!         hostname: "example.com".to_string(),
//!         path: PathRule::prefix(String::from("/")),
//!         position: RulePosition::Pre.into(),
//!         tags: BTreeMap::from([
//!             ("owner".to_owned(), "John".to_owned()),
//!             ("id".to_owned(), "my-own-http-front".to_owned()),
//!         ]),
//!         ..Default::default()
//!     };
//!     let http_backend = AddBackend {
//!         cluster_id: "my-cluster".to_string(),
//!         backend_id: "test-backend".to_string(),
//!         address: SocketAddress::new_v4(127,0,0,1,8000),
//!         load_balancing_parameters: Some(LoadBalancingParams::default()),
//!         ..Default::default()
//!     };
//!
//!     command_channel
//!         .write_message(&WorkerRequest {
//!             id: String::from("add-the-cluster"),
//!             content: RequestType::AddCluster(cluster).into(),
//!         })
//!         .expect("Could not send AddHttpFrontend request");
//!
//!     command_channel
//!         .write_message(&WorkerRequest {
//!             id: String::from("add-the-frontend"),
//!             content: RequestType::AddHttpFrontend(http_front).into(),
//!         })
//!         .expect("Could not send AddHttpFrontend request");
//!
//!     command_channel
//!         .write_message(&WorkerRequest {
//!             id: String::from("add-the-backend"),
//!             content: RequestType::AddBackend(http_backend).into(),
//!         })
//!         .expect("Could not send AddBackend request");
//!
//!     println!("HTTP -> {:?}", command_channel.read_message());
//!     println!("HTTP -> {:?}", command_channel.read_message());
//!
//!     // uncomment to let it run in the background
//!     // let _ = worker_thread_join_handle.join();
//!     info!("good bye");
//!     Ok(())
//! }
//! ```

#[macro_use]
extern crate sozu_command_lib as sozu_command;

#[macro_use]
pub mod util;
#[macro_use]
pub mod metrics;

pub mod backends;
pub mod crypto;
pub mod features;
pub mod health_check;
pub mod http;
pub mod load_balancing;
pub mod pool;
pub mod protocol;
pub mod retry;
pub mod router;
pub mod socket;
pub mod timer;
pub mod tls;

/// Linux zero-copy TCP forwarder. Used by `protocol::pipe::Pipe` when
/// the listener is `Protocol::TCP` and the `splice` feature is enabled.
#[cfg(all(target_os = "linux", feature = "splice"))]
pub(crate) mod splice;

pub mod server;
pub mod tcp;
pub mod udp;

pub mod https;

use std::{
    cell::RefCell,
    collections::{BTreeMap, HashMap},
    fmt::{self, Display, Formatter},
    net::SocketAddr,
    rc::Rc,
    str,
    time::{Duration, Instant, SystemTime},
};

use backends::BackendError;
use hex::FromHexError;
use mio::{Interest, Token, net::TcpStream};
use protocol::http::{answers::HttpAnswers, answers::TemplateError, parser::Method};
use router::RouterError;
use socket::ServerBindError;
use sozu_command::{
    AsStr, ObjectKind,
    logging::{CachedTags, LogContext},
    proto::command::{Cluster, ListenerType, RequestHttpFrontend, WorkerRequest, WorkerResponse},
    ready::Ready,
    state::ClusterId,
};
use tls::CertificateResolverError;

use crate::{backends::BackendMap, metrics::names, router::RouteResult};

/// Anything that can be registered in mio (subscribe to kernel events)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Protocol {
    HTTP,
    HTTPS,
    TCP,
    UDP,
    HTTPListen,
    HTTPSListen,
    TCPListen,
    UDPListen,
    Channel,
    Metrics,
    Timer,
}

/// trait that must be implemented by listeners and client sessions
pub trait ProxySession {
    /// indicates the protocol associated with the session
    ///
    /// this is used to distinguish sessions from listenrs, channels, metrics
    /// and timers
    fn protocol(&self) -> Protocol;
    /// if a session received an event or can still execute, the event loop will
    /// call this method. Its result indicates if it can still execute, needs to
    /// connect to a backend server, close the session
    fn ready(&mut self, session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed;
    /// if the event loop got an event for a token associated with the session,
    /// it will call this method on the session
    fn update_readiness(&mut self, token: Token, events: Ready);
    /// close a session, frontend and backend sockets,
    /// remove the entries from the session manager slab
    fn close(&mut self);
    /// if a timeout associated with the session triggers, the event loop will
    /// call this method with the timeout's token
    fn timeout(&mut self, t: Token) -> SessionIsToBeClosed;
    /// last time the session got an event
    fn last_event(&self) -> Instant;
    /// display the session's internal state (for debugging purpose)
    fn print_session(&self);
    /// get the token associated with the frontend
    fn frontend_token(&self) -> Token;
    /// tell the session it has to shut down if possible
    ///
    /// if the session handles HTTP requests, it will not close until the response
    /// is completely sent back to the client
    fn shutting_down(&mut self) -> SessionIsToBeClosed;
    /// Best-effort identifier of the cluster currently routed to by this
    /// session. Returns `None` for `ListenSession` (no per-session
    /// cluster), and for client sessions before routing has resolved.
    /// H2 sessions multiplex many streams over one frontend token and may
    /// touch several clusters; the returned value is whichever cluster
    /// the session most recently keep-alive'd to. Used for log/metric
    /// attribution, not for accounting (the tracker keeps the canonical
    /// per-stream `(cluster, IP)` set).
    fn cluster_id(&self) -> Option<String> {
        None
    }
    /// Source address as observed by Sōzu, with proxy-protocol awareness.
    /// HTTP/HTTPS/TCP client sessions return the parsed PROXY-protocol
    /// source when present, else `peer_addr`. `ListenSession` returns
    /// `None`. Used to attribute per-(cluster, source-IP) tracking and
    /// access logs to the real client behind a layer-4 PROXY frontend.
    fn session_address(&self) -> Option<SocketAddr> {
        None
    }
}

#[macro_export]
macro_rules! branch {
    (if $($value:ident)? == $expected:ident { $($then:tt)* } else { $($else:tt)* }) => {
        macro_rules! expect {
            ($expected) => {$($then)*};
            ($a:ident) => {$($else)*};
            () => {$($else)*}
        }
        expect!($($value)?);
    };
    (if $($value:ident)? == $expected:ident { $($then:tt)* } ) => {
        macro_rules! expect {
            ($expected) => {$($then)*};
        }
        expect!($($value)?);
    };
}

#[macro_export]
macro_rules! fallback {
    ({} $($default:tt)*) => {
        $($default)*
    };
    ({$($value:tt)+} $($default:tt)*) => {
        $($value)+
    };
}

#[macro_export]
macro_rules! StateMachineBuilder {
    (
        ($d:tt)
        $(#[$($state_macros:tt)*])*
        enum $state_name:ident $(impl $trait:ident)?  {
            $($(#[$($variant_macros:tt)*])*
            $variant_name:ident($state:ty$(,$($aux:ty),+)?) $(-> $override:expr)?),+ $(,)?
        }
    ) => {
        /// A summary of the last valid State
        #[derive(Clone, Copy, Debug)]
        pub enum StateMarker {
            $($variant_name,)+
        }

        $(#[$($state_macros)*])*
        #[allow(clippy::large_enum_variant)]
        pub enum $state_name {
            $(
                $(#[$($variant_macros)*])*
                $variant_name($state$(,$($aux),+)?),
            )+
            /// Informs about upgrade failure, contains a summary the last valid State
            FailedUpgrade(StateMarker),
        }

        macro_rules! _fn_impl {
            ($function:ident(&$d($mut:ident)?, self $d(,$arg_name:ident: $arg_type:ty)*) $d(-> $ret:ty)? $d(| $marker:tt => $fail:expr)?) => {
                fn $function(&$d($mut)? self $d(,$arg_name: $arg_type)*) $d(-> $ret)? {
                    match self {
                        $($state_name::$variant_name(_state, ..) => $crate::fallback!({$($override)?} _state.$function($d($arg_name),*)),)+
                        $state_name::FailedUpgrade($crate::fallback!({$d($marker)?} _)) => $crate::fallback!({$d($fail)?} unreachable!())
                    }
                }
            };
        }

        impl $state_name {
            /// Informs about the last valid State before upgrade failure
            fn marker(&self) -> StateMarker {
                match self {
                    $($state_name::$variant_name(..) => StateMarker::$variant_name,)+
                    $state_name::FailedUpgrade(marker) => *marker,
                }
            }
            /// Returns wether or not the State is FailedUpgrade
            fn failed(&self) -> bool {
                match self {
                    $state_name::FailedUpgrade(_) => true,
                    _ => false,
                }
            }
            /// Gives back an owned version of the State,
            /// leaving a FailedUpgrade in its place.
            /// The FailedUpgrade retains the marker of the previous State.
            fn take(&mut self) -> $state_name {
                let mut owned_state = $state_name::FailedUpgrade(self.marker());
                std::mem::swap(&mut owned_state, self);
                owned_state
            }
            _fn_impl!{front_socket(&, self) -> &mio::net::TcpStream}
        }

        $crate::branch!{
            if $($trait)? == SessionState {
                impl SessionState for $state_name {
                    _fn_impl!{ready(&mut, self, session: Rc<RefCell<dyn ProxySession>>, proxy: Rc<RefCell<dyn L7Proxy>>, metrics: &mut SessionMetrics) -> SessionResult}
                    _fn_impl!{update_readiness(&mut, self, token: Token, events: Ready)}
                    _fn_impl!{timeout(&mut, self, token: Token, metrics: &mut SessionMetrics) -> StateResult}
                    _fn_impl!{cancel_timeouts(&mut, self)}
                    _fn_impl!{print_state(&, self, context: &str) | marker => error!("{} Session(FailedUpgrade({:?}))", context, marker)}
                    _fn_impl!{close(&mut, self, proxy: Rc<RefCell<dyn L7Proxy>>, metrics: &mut SessionMetrics) | _ => {}}
                    _fn_impl!{shutting_down(&mut, self) -> SessionIsToBeClosed | _ => true}
                }
            } else {}
        }
    };
    ($($tt:tt)+) => {
        StateMachineBuilder!{($) $($tt)+}
    }
}

pub trait ListenerHandler {
    fn get_addr(&self) -> &SocketAddr;

    fn get_tags(&self, key: &str) -> Option<&CachedTags>;

    fn get_concatenated_tags(&self, key: &str) -> Option<&str> {
        self.get_tags(key).map(|tags| tags.concatenated.as_str())
    }

    fn set_tags(&mut self, key: String, tags: Option<BTreeMap<String, String>>);

    fn protocol(&self) -> Protocol;

    fn public_address(&self) -> SocketAddr;
}

#[derive(thiserror::Error, Debug)]
pub enum FrontendFromRequestError {
    #[error("Could not parse hostname from '{host}': {error}")]
    HostParse { host: String, error: String },
    #[error("invalid remaining chars after hostname. Host: {0}")]
    InvalidCharsAfterHost(String),
    #[error("no cluster: {0}")]
    NoClusterFound(RouterError),
}

pub trait L7ListenerHandler {
    fn get_sticky_name(&self) -> &str;

    /// Name of the correlation header Sozu injects into every request and
    /// response body. Default: `"Sozu-Id"`. Operators can rebrand via the
    /// `sozu_id_header` listener config knob.
    fn get_sozu_id_header(&self) -> &str {
        "Sozu-Id"
    }

    fn get_connect_timeout(&self) -> u32;

    /// retrieve a frontend by parsing a request's hostname, uri and method
    fn frontend_from_request(
        &self,
        host: &str,
        uri: &str,
        method: &Method,
    ) -> Result<RouteResult, FrontendFromRequestError>;

    /// retrieve the listener's configured HTTP answers (templates)
    fn get_answers(&self) -> &Rc<RefCell<HttpAnswers>>;

    /// H2 flood detection thresholds from the listener config.
    /// Returns the default config when the listener does not provide custom values.
    fn get_h2_flood_config(&self) -> protocol::mux::H2FloodConfig {
        protocol::mux::H2FloodConfig::default()
    }

    /// H2 connection tuning from the listener config.
    /// Returns the default config when the listener does not provide custom values.
    fn get_h2_connection_config(&self) -> protocol::mux::H2ConnectionConfig {
        protocol::mux::H2ConnectionConfig::default()
    }

    /// Whether requests must have their `:authority` / `Host` exact-match
    /// the TLS SNI negotiated at handshake (CWE-346 / CWE-444).
    ///
    /// Defaults to `true` — the safe setting that closes the
    /// CWE-346 / CWE-444 cross-SNI smuggling vector. Operators can opt
    /// out per-listener via `HttpsListenerConfig::strict_sni_binding =
    /// false` when cross-SNI routing is explicitly required. Plaintext
    /// HTTP listeners return the default value; they never have an SNI
    /// to compare against, so the routing-layer check short-circuits on
    /// `tls_server_name: None`.
    fn get_strict_sni_binding(&self) -> bool {
        true
    }

    /// Whether to strip any client-supplied `X-Real-IP` header from
    /// forwarded requests (anti-spoofing).
    ///
    /// Defaults to `false` — preserves the historical pass-through
    /// behaviour. Operators opt in via
    /// `HttpListenerConfig::elide_x_real_ip = true` (and the equivalent on
    /// HTTPS listeners). Independent of [`Self::get_send_x_real_ip`]: the
    /// two flags can be combined freely (anti-spoof only, send only, both,
    /// or neither). The elision branch lives in
    /// `HttpContext::on_request_headers`, so it covers H1 and H2 alike.
    fn get_elide_x_real_ip(&self) -> bool {
        false
    }

    /// Whether to append a proxy-generated `X-Real-IP` header carrying the
    /// connection peer IP (post-PROXY-v2 unwrap, i.e. the original client
    /// IP) to every forwarded request.
    ///
    /// Defaults to `false` — preserves the historical no-injection
    /// behaviour. Operators opt in via
    /// `HttpListenerConfig::send_x_real_ip = true` (and the equivalent on
    /// HTTPS listeners). Independent of [`Self::get_elide_x_real_ip`]: the
    /// two flags can be combined freely. The injection branch lives next
    /// to the existing X-Forwarded-For / Forwarded synthesis in
    /// `HttpContext::on_request_headers`.
    fn get_send_x_real_ip(&self) -> bool {
        false
    }

    /// Per-stream idle timeout for H2 connections. An open stream that makes
    /// no forward progress for this duration is cancelled (RST_STREAM / CANCEL).
    /// Mitigates slow-multiplex Slowloris where a client keeps connection-level
    /// activity high (resetting the connection idle timer on every frame) while
    /// pinning streams for the full nominal connection timeout.
    ///
    /// Listeners inherit `max(30s, back_timeout)` when `h2_stream_idle_timeout_seconds`
    /// is absent so operators who raised the socket-level backend budget do not
    /// have to duplicate the value here; the 30 s floor preserves the baseline
    /// slow-multiplex mitigation when `back_timeout` is shorter. Set the knob
    /// explicitly to cap the per-stream deadline below `back_timeout` (useful
    /// when under a slow-multiplex attack).
    fn get_h2_stream_idle_timeout(&self) -> std::time::Duration {
        std::time::Duration::from_secs(30)
    }

    /// Wall-clock budget granted to in-flight H2 streams after soft-stop sent
    /// the initial `GOAWAY(NO_ERROR)`. Once the deadline elapses the mux
    /// transitions to a forced close (final GOAWAY + session teardown).
    ///
    /// Returning `None` disables the forced close entirely — shutdown waits
    /// for every stream to drain naturally. Returning `Some(d)` enforces the
    /// budget. Default: `Some(Duration::from_secs(5))` (matches the historic
    /// hard-coded 5 s deadline). Listeners expose the
    /// `h2_graceful_shutdown_deadline_seconds` knob; value `0` maps to `None`.
    fn get_h2_graceful_shutdown_deadline(&self) -> Option<std::time::Duration> {
        Some(std::time::Duration::from_secs(5))
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BackendConnectionStatus {
    NotConnected,
    Connecting(Instant),
    Connected,
}

impl BackendConnectionStatus {
    pub fn is_connecting(&self) -> bool {
        matches!(self, BackendConnectionStatus::Connecting(_))
    }
}

#[derive(Debug, PartialEq, Eq)]
pub enum BackendConnectAction {
    New,
    Reuse,
    Replace,
}

#[derive(thiserror::Error, Debug)]
pub enum BackendConnectionError {
    #[error("Not found: {0:?}")]
    NotFound(ObjectKind),
    #[error("Too many connections on cluster {0:?}")]
    MaxConnectionRetries(Option<String>),
    #[error("the sessions slab has reached maximum capacity")]
    MaxSessionsMemory,
    #[error("error from the backend: {0}")]
    Backend(BackendError),
    #[error("failed to retrieve the cluster: {0}")]
    RetrieveClusterError(RetrieveClusterError),
    #[error("maximum number of buffers reached")]
    MaxBuffers,
    /// Per-(cluster, source-IP) connection limit reached. The protocol
    /// layer translates this into HTTP 429 Too Many Requests (with an
    /// optional `Retry-After`) for HTTP/HTTPS sessions, or a graceful TCP
    /// close for raw TCP. The `cluster_id` is included so log/metric
    /// pipelines can attribute the rejection.
    #[error("per-(cluster, source-IP) connection limit reached for cluster {cluster_id:?}")]
    TooManyConnectionsPerIp { cluster_id: String },
}

/// used in kawa_h1 module for the Http session state
#[derive(thiserror::Error, Debug)]
pub enum RetrieveClusterError {
    #[error("No method given")]
    NoMethod,
    #[error("No host given")]
    NoHost,
    #[error("No path given")]
    NoPath,
    #[error("unauthorized route")]
    UnauthorizedRoute,
    #[error("{0}")]
    RetrieveFrontend(FrontendFromRequestError),
    #[error("HTTPS redirect required")]
    HttpsRedirect,
    /// The HTTP `:authority` / `Host` host does not match the TLS SNI that was
    /// negotiated for this connection, which would cross the TLS trust boundary.
    /// Maps to HTTP 421 Misdirected Request (RFC 9110 §15.5.20).
    #[error("TLS SNI {sni:?} does not match HTTP authority {authority:?}")]
    SniAuthorityMismatch { sni: String, authority: String },
}

/// Used in sessions
#[derive(Debug, PartialEq, Eq)]
pub enum AcceptError {
    IoError,
    TooManySessions,
    WouldBlock,
    RegisterError,
    WrongSocketAddress,
    BufferCapacityReached,
}

/// returned by the HTTP, HTTPS and TCP listeners
#[derive(thiserror::Error, Debug)]
pub enum ListenerError {
    #[error("failed to handle certificate request, got a resolver error, {0}")]
    Resolver(CertificateResolverError),
    #[error("failed to parse pem, {0}")]
    PemParse(String),
    #[error("failed to parse template {0:?}: {1}")]
    TemplateParse(String, TemplateError),
    #[error("failed to build rustls context, {0}")]
    BuildRustls(String),
    #[error("could not activate listener with address {address:?}: {error}")]
    Activation { address: SocketAddr, error: String },
    #[error("Could not register listener socket: {0}")]
    SocketRegistration(std::io::Error),
    #[error("could not add frontend: {0}")]
    AddFrontend(RouterError),
    #[error("could not remove frontend: {0}")]
    RemoveFrontend(RouterError),
    #[error("invalid value for field '{field}': {reason}")]
    InvalidValue {
        field: &'static str,
        reason: &'static str,
    },
    /// `UpdateHttpsListenerConfig.hsts` was present but its `enabled`
    /// field was unset. Per the partial-update contract, `enabled` is
    /// the explicit-disambiguator between "explicit disable" (false) and
    /// "explicit enable" (true); the patch handler refuses an
    /// `enabled = None` block rather than silently picking one.
    #[error(
        "UpdateHttpsListenerConfig.hsts is present but `enabled` is unset; the partial-update \
         contract requires `enabled` whenever the `hsts` block is present"
    )]
    HstsEnabledRequired,
}

/// Lift control-plane validation errors into listener-level errors so the
/// worker can surface the same message without duplicating the match.
/// Non-`InvalidValue` variants fall back to a generic `InvalidValue` — they
/// are not expected on the worker's `update_config` path (state lookups
/// happen on the master) but we avoid panicking if one slips through.
impl From<sozu_command::state::StateError> for ListenerError {
    fn from(err: sozu_command::state::StateError) -> Self {
        match err {
            sozu_command::state::StateError::InvalidValue { field, reason } => {
                ListenerError::InvalidValue { field, reason }
            }
            _ => ListenerError::InvalidValue {
                field: "state",
                reason: "unexpected state error on worker path",
            },
        }
    }
}

/// Returned by the HTTP, HTTPS and TCP proxies
#[derive(thiserror::Error, Debug)]
pub enum ProxyError {
    #[error("error while soft stopping {proxy_protocol} proxy: {error}")]
    SoftStop {
        proxy_protocol: String,
        error: String,
    },
    #[error("error while hard stopping {proxy_protocol} proxy: {error}")]
    HardStop {
        proxy_protocol: String,
        error: String,
    },
    #[error("found no listener with address {0:?}")]
    NoListenerFound(SocketAddr),
    #[error("a listener is already present for this token")]
    ListenerAlreadyPresent,
    #[error("could not add listener: {0}")]
    AddListener(ListenerError),
    #[error("could not add cluster: {0}")]
    AddCluster(ListenerError),
    #[error("failed to activate listener with address {address:?}: {listener_error}")]
    ListenerActivation {
        address: SocketAddr,
        listener_error: ListenerError,
    },
    #[error("can not add frontend {front:?}: {error}")]
    WrongInputFrontend {
        front: Box<RequestHttpFrontend>,
        error: String,
    },
    #[error("could not add frontend: {0}")]
    AddFrontend(ListenerError),
    #[error("could not remove frontend: {0}")]
    RemoveFrontend(ListenerError),
    #[error("could not add certificate: {0}")]
    AddCertificate(CertificateResolverError),
    #[error("could not remove certificate: {0}")]
    RemoveCertificate(CertificateResolverError),
    #[error("could not replace certificate: {0}")]
    ReplaceCertificate(CertificateResolverError),
    #[error("wrong certificate fingerprint: {0}")]
    WrongCertificateFingerprint(FromHexError),
    #[error("this request is not supported by the proxy")]
    UnsupportedMessage,
    #[error("failed to acquire the lock, {0}")]
    Lock(String),
    #[error("could not bind to socket {0:?}: {1}")]
    BindToSocket(SocketAddr, ServerBindError),
    #[error("error registering socket of listener: {0}")]
    RegisterListener(std::io::Error),
    #[error("the listener is not activated")]
    UnactivatedListener,
    /// HSTS (RFC 6797) was attached to a frontend on a plain-HTTP
    /// listener. RFC 6797 §7.2 forbids `Strict-Transport-Security` on
    /// plaintext-HTTP responses; the worker rejects the request rather
    /// than ship a non-conformant policy. The TOML loader rejects the
    /// same shape at config-load time
    /// (`command/src/config.rs::ConfigError::HstsOnPlainHttp`); this
    /// arm catches the same misconfiguration when the request reaches
    /// the worker over the IPC channel without going through the TOML
    /// path (e.g. via `sozu frontend http add`).
    #[error(
        "HSTS is only valid on HTTPS frontends; rejecting AddHttpFrontend with hsts.enabled = \
         true on address {0:?} (RFC 6797 §7.2)"
    )]
    HstsOnPlainHttp(SocketAddr),
    /// A TCP `AddTcpFrontend` would corrupt this listener's SNI/ALPN
    /// routing invariants (sozu-proxy/sozu#1279): mixing a no-SNI
    /// catch-all with SNI-scoped routes, an `alpn`-scoped frontend with no
    /// `sni` to preread against, or an ALPN protocol / catch-all that
    /// overlaps an existing route on the same `(address, sni)`. TOML
    /// config-load already rejects the same shapes
    /// (`command/src/config.rs::ConfigError`), but `AddTcpFrontend` can
    /// also arrive directly over the command socket or via `LoadState`
    /// replay of a hand-edited/stale state file, bypassing config.rs
    /// entirely — see `TcpListener::validate_new_tcp_front`
    /// (`lib/src/tcp.rs`).
    #[error("rejected AddTcpFrontend on listener {address:?}: {reason}")]
    InvalidTcpFrontend { address: SocketAddr, reason: String },
}

use self::server::ListenToken;
pub trait ProxyConfiguration {
    fn notify(&mut self, message: WorkerRequest) -> WorkerResponse;
    fn accept(&mut self, token: ListenToken) -> Result<TcpStream, AcceptError>;
    fn create_session(
        &mut self,
        socket: TcpStream,
        token: ListenToken,
        wait_time: Duration,
        proxy: Rc<RefCell<Self>>,
        // should we insert the tags here?
    ) -> Result<(), AcceptError>;
}

pub trait L7Proxy {
    fn kind(&self) -> ListenerType;

    fn register_socket(
        &self,
        socket: &mut TcpStream,
        token: Token,
        interest: Interest,
    ) -> Result<(), std::io::Error>;

    fn deregister_socket(&self, tcp_stream: &mut TcpStream) -> Result<(), std::io::Error>;

    fn add_session(&self, session: Rc<RefCell<dyn ProxySession>>) -> Token;

    /// Remove the session from the session manager slab.
    /// Returns true if the session was actually there before deletion
    fn remove_session(&self, token: Token) -> bool;

    fn backends(&self) -> Rc<RefCell<BackendMap>>;

    fn clusters(&self) -> &HashMap<ClusterId, Cluster>;

    /// Access the worker's [`SessionManager`] for per-(cluster, source-IP)
    /// connection-limit accounting. The mux uses this to track / untrack
    /// stream-granular `(cluster_id, ip)` entries and consult the
    /// `cluster_ip_at_limit` gate before each backend connect.
    fn sessions(&self) -> Rc<RefCell<crate::server::SessionManager>>;
}

#[derive(Debug, PartialEq, Eq)]
pub enum RequiredEvents {
    FrontReadBackNone,
    FrontWriteBackNone,
    FrontReadWriteBackNone,
    FrontNoneBackNone,
    FrontReadBackRead,
    FrontWriteBackRead,
    FrontReadWriteBackRead,
    FrontNoneBackRead,
    FrontReadBackWrite,
    FrontWriteBackWrite,
    FrontReadWriteBackWrite,
    FrontNoneBackWrite,
    FrontReadBackReadWrite,
    FrontWriteBackReadWrite,
    FrontReadWriteBackReadWrite,
    FrontNoneBackReadWrite,
}

impl RequiredEvents {
    pub fn front_readable(&self) -> bool {
        matches!(
            *self,
            RequiredEvents::FrontReadBackNone
                | RequiredEvents::FrontReadWriteBackNone
                | RequiredEvents::FrontReadBackRead
                | RequiredEvents::FrontReadWriteBackRead
                | RequiredEvents::FrontReadBackWrite
                | RequiredEvents::FrontReadWriteBackWrite
                | RequiredEvents::FrontReadBackReadWrite
                | RequiredEvents::FrontReadWriteBackReadWrite
        )
    }

    pub fn front_writable(&self) -> bool {
        matches!(
            *self,
            RequiredEvents::FrontWriteBackNone
                | RequiredEvents::FrontReadWriteBackNone
                | RequiredEvents::FrontWriteBackRead
                | RequiredEvents::FrontReadWriteBackRead
                | RequiredEvents::FrontWriteBackWrite
                | RequiredEvents::FrontReadWriteBackWrite
                | RequiredEvents::FrontWriteBackReadWrite
                | RequiredEvents::FrontReadWriteBackReadWrite
        )
    }

    pub fn back_readable(&self) -> bool {
        matches!(
            *self,
            RequiredEvents::FrontReadBackRead
                | RequiredEvents::FrontWriteBackRead
                | RequiredEvents::FrontReadWriteBackRead
                | RequiredEvents::FrontNoneBackRead
                | RequiredEvents::FrontReadBackReadWrite
                | RequiredEvents::FrontWriteBackReadWrite
                | RequiredEvents::FrontReadWriteBackReadWrite
                | RequiredEvents::FrontNoneBackReadWrite
        )
    }

    pub fn back_writable(&self) -> bool {
        matches!(
            *self,
            RequiredEvents::FrontReadBackWrite
                | RequiredEvents::FrontWriteBackWrite
                | RequiredEvents::FrontReadWriteBackWrite
                | RequiredEvents::FrontNoneBackWrite
                | RequiredEvents::FrontReadBackReadWrite
                | RequiredEvents::FrontWriteBackReadWrite
                | RequiredEvents::FrontReadWriteBackReadWrite
                | RequiredEvents::FrontNoneBackReadWrite
        )
    }
}

/// Signals transitions between states of a given Protocol
#[derive(Debug, PartialEq, Eq)]
pub enum StateResult {
    /// Signals to the Protocol to close its backend
    CloseBackend,
    /// Signals to the parent Session to close itself
    CloseSession,
    /// Signals to the Protocol to connect to backend
    ConnectBackend,
    /// Signals to the Protocol to continue
    Continue,
    /// Signals to the parent Session to upgrade to the next Protocol
    Upgrade,
}

/// Signals transitions between states of a given Session
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionResult {
    /// Signals to the Session to close itself
    Close,
    /// Signals to the Session to continue
    Continue,
    /// Signals to the Session to upgrade its Protocol
    Upgrade,
}

#[derive(Debug, PartialEq, Eq)]
pub enum SocketType {
    Listener,
    FrontClient,
}

type SessionIsToBeClosed = bool;

#[derive(Clone)]
pub struct Readiness {
    /// the current readiness
    pub event: Ready,
    /// the readiness we wish to attain
    pub interest: Ready,
}

impl Display for Readiness {
    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
        let i = &mut [b'-'; 4];
        let r = &mut [b'-'; 4];
        let mixed = &mut [b'-'; 4];

        display_ready(i, self.interest);
        display_ready(r, self.event);
        display_ready(mixed, self.interest & self.event);

        write!(
            f,
            "I({:?})&R({:?})=M({:?})",
            String::from_utf8_lossy(i),
            String::from_utf8_lossy(r),
            String::from_utf8_lossy(mixed)
        )
    }
}

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

impl Readiness {
    /// Mask of every bit `Ready` defines (READABLE | WRITABLE | ERROR | HUP).
    /// Any bit outside this set in `event` or `interest` is a corrupted
    /// readiness word — checked by [`Self::check_invariants`]. Not
    /// `#[cfg(debug_assertions)]`-gated: it is read from inside `debug_assert!`s
    /// whose arguments must still compile in release (HARD RULE 2 / E0425).
    const KNOWN_BITS: Ready =
        Ready(Ready::READABLE.0 | Ready::WRITABLE.0 | Ready::ERROR.0 | Ready::HUP.0);

    pub const fn new() -> Readiness {
        Readiness {
            event: Ready::EMPTY,
            interest: Ready::EMPTY,
        }
    }

    /// Cross-field invariant sweep: neither `event` nor `interest` may carry a
    /// bit `Ready` does not define. A stray bit would silently widen
    /// `filter_interest`'s mask and wake (or starve) a session on a phantom
    /// readiness. Cheap enough to call as a postcondition from every mutator.
    #[cfg(debug_assertions)]
    fn check_invariants(&self) {
        debug_assert_eq!(
            self.event & Self::KNOWN_BITS,
            self.event,
            "Readiness.event carries a bit outside READABLE|WRITABLE|ERROR|HUP"
        );
        debug_assert_eq!(
            self.interest & Self::KNOWN_BITS,
            self.interest,
            "Readiness.interest carries a bit outside READABLE|WRITABLE|ERROR|HUP"
        );
    }

    pub fn reset(&mut self) {
        self.event = Ready::EMPTY;
        self.interest = Ready::EMPTY;
        // Post-condition: a reset clears *both* words — a half-reset leaves a
        // session armed on stale interest after teardown.
        debug_assert!(
            self.event.is_empty() && self.interest.is_empty(),
            "reset must clear both event and interest"
        );
        #[cfg(debug_assertions)]
        self.check_invariants();
    }

    /// filters the readiness we actually want
    pub fn filter_interest(&self) -> Ready {
        // Pre-condition: both source words must be well-formed before we mask —
        // a stray bit upstream would leak through the intersection and wake (or
        // starve) a session on a phantom readiness.
        #[cfg(debug_assertions)]
        self.check_invariants();
        let filtered = self.event & self.interest;
        // Post-condition: the result is a subset of the recognized bits and can
        // only contain bits the session both saw AND asked for.
        debug_assert_eq!(
            filtered & Self::KNOWN_BITS,
            filtered,
            "filter_interest must not yield an unknown bit"
        );
        debug_assert!(
            self.interest.contains(filtered) && self.event.contains(filtered),
            "filtered readiness must be present in both interest and event"
        );
        filtered
    }

    /// Signal that the socket has buffered data to write (e.g., TLS internal
    /// buffers) that won't generate a new epoll WRITABLE event.
    pub fn signal_pending_write(&mut self) {
        // Snapshot the unrelated bits (everything but WRITABLE) so we can prove
        // we flipped exactly the WRITABLE bit. `Ready` exposes no `!`, so mask on
        // the public `.0` word.
        let other_event_before = Ready(self.event.0 & !Ready::WRITABLE.0);
        self.event.insert(Ready::WRITABLE);
        debug_assert!(
            self.event.is_writable(),
            "signal_pending_write must set the WRITABLE event bit"
        );
        debug_assert_eq!(
            Ready(self.event.0 & !Ready::WRITABLE.0),
            other_event_before,
            "signal_pending_write must touch only the WRITABLE bit"
        );
        #[cfg(debug_assertions)]
        self.check_invariants();
    }

    /// Signal that the socket has buffered data to read (e.g., TLS plaintext
    /// buffer after a 1xx clear) that won't generate a new epoll READABLE event.
    pub fn signal_pending_read(&mut self) {
        let other_event_before = Ready(self.event.0 & !Ready::READABLE.0);
        self.event.insert(Ready::READABLE);
        debug_assert!(
            self.event.is_readable(),
            "signal_pending_read must set the READABLE event bit"
        );
        debug_assert_eq!(
            Ready(self.event.0 & !Ready::READABLE.0),
            other_event_before,
            "signal_pending_read must touch only the READABLE bit"
        );
        #[cfg(debug_assertions)]
        self.check_invariants();
    }

    /// Pair `Ready::WRITABLE` insert with `signal_pending_write` — the canonical
    /// invariant-15 form for any path that writes bytes to sozu-owned buffers
    /// under edge-triggered epoll. See `lib/src/protocol/mux/LIFECYCLE.md`.
    #[inline]
    pub fn arm_writable(&mut self) {
        // Snapshot the non-WRITABLE bits of both words: arm_writable must set the
        // WRITABLE bit in *both* interest and event and leave everything else as-is.
        let other_interest_before = Ready(self.interest.0 & !Ready::WRITABLE.0);
        let other_event_before = Ready(self.event.0 & !Ready::WRITABLE.0);
        self.interest.insert(Ready::WRITABLE);
        self.signal_pending_write();
        debug_assert!(
            self.interest.is_writable() && self.event.is_writable(),
            "arm_writable must set WRITABLE in both interest and event"
        );
        debug_assert_eq!(
            Ready(self.interest.0 & !Ready::WRITABLE.0),
            other_interest_before,
            "arm_writable must touch only the WRITABLE interest bit"
        );
        debug_assert_eq!(
            Ready(self.event.0 & !Ready::WRITABLE.0),
            other_event_before,
            "arm_writable must touch only the WRITABLE event bit"
        );
        #[cfg(debug_assertions)]
        self.check_invariants();
    }
}

#[cfg(test)]
mod readiness_tests {
    use super::{Readiness, Ready};

    #[test]
    fn arm_writable_sets_interest_and_event() {
        let mut r = Readiness::new();
        r.arm_writable();
        assert!(r.interest.is_writable());
        assert!(r.event.is_writable());
    }

    #[test]
    fn arm_writable_is_idempotent() {
        let mut r = Readiness::new();
        r.arm_writable();
        r.arm_writable();
        assert_eq!(r.interest, Ready::WRITABLE);
        assert_eq!(r.event, Ready::WRITABLE);
    }
}

pub fn display_ready(s: &mut [u8], readiness: Ready) {
    if readiness.is_readable() {
        s[0] = b'R';
    }
    if readiness.is_writable() {
        s[1] = b'W';
    }
    if readiness.is_error() {
        s[2] = b'E';
    }
    if readiness.is_hup() {
        s[3] = b'H';
    }
}

pub fn ready_to_string(readiness: Ready) -> String {
    let s = &mut [b'-'; 4];
    display_ready(s, readiness);
    String::from_utf8(s.to_vec()).unwrap()
}

impl fmt::Debug for Readiness {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        let i = &mut [b'-'; 4];
        let r = &mut [b'-'; 4];
        let mixed = &mut [b'-'; 4];

        display_ready(i, self.interest);
        display_ready(r, self.event);
        display_ready(mixed, self.interest & self.event);

        write!(
            f,
            "Readiness {{ interest: {}, readiness: {}, mixed: {} }}",
            str::from_utf8(i).unwrap(),
            str::from_utf8(r).unwrap(),
            str::from_utf8(mixed).unwrap()
        )
    }
}

#[derive(Clone, Debug)]
pub struct SessionMetrics {
    /// date at which we started handling that request
    pub start: Option<Instant>,
    /// wall-clock timestamp captured alongside `start`, for access-log
    /// consumers that need an absolute start time (e.g. OTel span
    /// reconstruction) without subtracting a monotonic duration from a
    /// wall-clock end time — which mixes two unsynchronised clock sources.
    pub start_wall: Option<SystemTime>,
    /// time actually spent handling the request
    pub service_time: Duration,
    /// time spent waiting for its turn
    pub wait_time: Duration,
    /// bytes received by the frontend
    pub bin: usize,
    /// bytes sent by the frontend
    pub bout: usize,

    /// date at which we started working on the request
    pub service_start: Option<Instant>,
    pub wait_start: Instant,

    pub backend_id: Option<String>,
    pub backend_start: Option<Instant>,
    pub backend_connected: Option<Instant>,
    pub backend_stop: Option<Instant>,
    pub backend_bin: usize,
    pub backend_bout: usize,
}

impl SessionMetrics {
    pub fn new(wait_time: Option<Duration>) -> SessionMetrics {
        SessionMetrics {
            start: Some(Instant::now()),
            start_wall: Some(SystemTime::now()),
            service_time: Duration::from_secs(0),
            wait_time: wait_time.unwrap_or_else(|| Duration::from_secs(0)),
            bin: 0,
            bout: 0,
            service_start: None,
            wait_start: Instant::now(),
            backend_id: None,
            backend_start: None,
            backend_connected: None,
            backend_stop: None,
            backend_bin: 0,
            backend_bout: 0,
        }
    }

    pub fn reset(&mut self) {
        self.start = None;
        self.start_wall = None;
        self.service_time = Duration::from_secs(0);
        self.wait_time = Duration::from_secs(0);
        self.bin = 0;
        self.bout = 0;
        self.service_start = None;
        self.backend_start = None;
        self.backend_connected = None;
        self.backend_stop = None;
        self.backend_bin = 0;
        self.backend_bout = 0;
    }

    pub fn service_start(&mut self) {
        let now = if self.start.is_none() {
            self.mark_request_start()
        } else {
            Instant::now()
        };
        self.service_start = Some(now);
        self.wait_time += now - self.wait_start;
    }

    pub fn service_stop(&mut self) {
        if let Some(start) = self.service_start.take() {
            let duration = Instant::now() - start;
            self.service_time += duration;
        }
    }

    pub fn wait_start(&mut self) {
        self.wait_start = Instant::now();
    }

    pub fn service_time(&self) -> Duration {
        match self.service_start {
            Some(start) => {
                let last_duration = Instant::now() - start;
                self.service_time + last_duration
            }
            None => self.service_time,
        }
    }

    /// Arm both the monotonic and wall-clock start timestamps together.
    /// This must be the single place that sets `start` + `start_wall` outside
    /// of `new()`, so the two fields can never desynchronize.
    /// Returns the monotonic instant so callers that need it (e.g.
    /// `service_start`) can reuse it without a second syscall.
    pub fn mark_request_start(&mut self) -> Instant {
        let now = Instant::now();
        self.start = Some(now);
        self.start_wall = Some(SystemTime::now());
        now
    }

    /// time elapsed since the beginning of the session
    pub fn request_time(&self) -> Duration {
        match self.start {
            Some(start) => Instant::now() - start,
            None => Duration::from_secs(0),
        }
    }

    /// Wall-clock start time as nanoseconds since the Unix epoch, or `None`
    /// if the monotonic start has not been set yet (post-`reset()`, pre-`service_start()`).
    pub fn start_wall_ns(&self) -> Option<i128> {
        self.start_wall.and_then(|t| {
            t.duration_since(SystemTime::UNIX_EPOCH)
                .ok()
                .map(|d| d.as_nanos() as i128)
        })
    }

    pub fn backend_start(&mut self) {
        self.backend_start = Some(Instant::now());
    }

    pub fn backend_connected(&mut self) {
        self.backend_connected = Some(Instant::now());
    }

    pub fn backend_stop(&mut self) {
        self.backend_stop = Some(Instant::now());
    }

    pub fn backend_response_time(&self) -> Option<Duration> {
        match (self.backend_connected, self.backend_stop) {
            (Some(start), Some(end)) => Some(end - start),
            (Some(start), None) => Some(Instant::now() - start),
            _ => None,
        }
    }

    pub fn backend_connection_time(&self) -> Option<Duration> {
        match (self.backend_start, self.backend_connected) {
            (Some(start), Some(end)) => Some(end - start),
            _ => None,
        }
    }

    pub fn register_end_of_session(&self, context: &LogContext) {
        let request_time = self.request_time();
        let service_time = self.service_time();

        if let Some(cluster_id) = context.cluster_id {
            time!(
                names::event_loop::REQUEST_TIME,
                cluster_id,
                request_time.as_millis()
            );
            time!(
                names::event_loop::SERVICE_TIME,
                cluster_id,
                service_time.as_millis()
            );
        }
        time!(names::event_loop::REQUEST_TIME, request_time.as_millis());
        time!(names::event_loop::SERVICE_TIME, service_time.as_millis());

        if let Some(backend_id) = self.backend_id.as_ref()
            && let Some(backend_response_time) = self.backend_response_time()
        {
            record_backend_metrics!(
                context.cluster_id.as_str_or("-"),
                backend_id,
                backend_response_time.as_millis(),
                self.backend_connection_time(),
                self.backend_bin,
                self.backend_bout
            );
        }

        incr!(
            names::access_logs::COUNT,
            context.cluster_id,
            context.backend_id
        );
    }
}

/// exponentially weighted moving average with high sensibility to latency bursts
///
/// cf Finagle for the original implementation: <https://github.com/twitter/finagle/blob/9cc08d15216497bb03a1cafda96b7266cfbbcff1/finagle-core/src/main/scala/com/twitter/finagle/loadbalancer/PeakEwma.scala>
#[derive(Debug, PartialEq, Clone)]
pub struct PeakEWMA {
    /// decay in nanoseconds
    ///
    /// higher values will make the EWMA decay slowly to 0
    pub decay: f64,
    /// estimated RTT in nanoseconds
    ///
    /// must be set to a high enough default value so that new backends do not
    /// get all the traffic right away
    pub rtt: f64,
    /// last modification
    pub last_event: Instant,
}

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

impl PeakEWMA {
    // hardcoded default values for now
    pub fn new() -> Self {
        PeakEWMA {
            // 1s
            decay: 1_000_000_000f64,
            // 50ms
            rtt: 50_000_000f64,
            last_event: Instant::now(),
        }
    }

    pub fn observe(&mut self, rtt: f64) {
        let now = Instant::now();
        let dur = now - self.last_event;

        // if latency is rising, we will immediately raise the cost
        if rtt > self.rtt {
            self.rtt = rtt;
        } else {
            // new_rtt = old_rtt * e^(-elapsed/decay) + observed_rtt * (1 - e^(-elapsed/decay))
            let weight = (-(dur.as_nanos() as f64) / self.decay).exp();
            self.rtt = self.rtt * weight + rtt * (1.0 - weight);
        }

        self.last_event = now;
    }

    pub fn get(&mut self, active_requests: usize) -> f64 {
        // decay the current value
        // (we might not have seen a request in a long time)
        self.observe(0.0);

        (active_requests + 1) as f64 * self.rtt
    }
}

pub mod testing {
    pub use std::{cell::RefCell, os::fd::IntoRawFd, rc::Rc};

    pub use anyhow::Context;
    pub use mio::{Poll, Registry, Token, net::UnixStream};
    pub use slab::Slab;
    pub use sozu_command::{
        proto::command::{
            HttpListenerConfig, HttpsListenerConfig, ServerConfig, TcpListenerConfig,
        },
        scm_socket::{Listeners, ScmSocket},
    };

    pub use crate::{
        Protocol, ProxySession,
        backends::BackendMap,
        http::HttpProxy,
        https::HttpsProxy,
        pool::Pool,
        server::{ListenSession, ProxyChannel, Server, SessionManager},
        tcp::TcpProxy,
    };

    use std::sync::atomic::{AtomicU16, Ordering};

    /// Port counter for sozu listener addresses in lib tests.
    /// Starts at 10000 to avoid collision with:
    /// - Privileged ports (<1024)
    /// - e2e suite (starts at 2000)
    /// - Ephemeral port range (typically 32768+)
    static PORT_PROVIDER: AtomicU16 = AtomicU16::new(10000);

    /// Get a unique port for a sozu listener address.
    /// Each call returns a different port, safe for parallel test execution.
    pub fn provide_port() -> u16 {
        PORT_PROVIDER.fetch_add(1, Ordering::SeqCst)
    }

    /// Everything needed to create a Server
    pub struct ServerParts {
        pub event_loop: Poll,
        pub registry: Registry,
        pub sessions: Rc<RefCell<SessionManager>>,
        pub pool: Rc<RefCell<Pool>>,
        pub backends: Rc<RefCell<BackendMap>>,
        pub client_scm_socket: ScmSocket,
        pub server_scm_socket: ScmSocket,
        pub server_config: ServerConfig,
    }

    /// Setup a standalone server, for testing purposes
    pub fn prebuild_server(
        max_buffers: usize,
        buffer_size: usize,
        send_scm: bool,
    ) -> anyhow::Result<ServerParts> {
        let event_loop = Poll::new().with_context(|| "Failed at creating event loop")?;
        let backends = Rc::new(RefCell::new(BackendMap::new()));
        let server_config = ServerConfig {
            max_connections: max_buffers as u64,
            ..Default::default()
        };

        let pool = Rc::new(RefCell::new(Pool::with_capacity(
            1,
            max_buffers,
            buffer_size,
        )));

        let mut sessions: Slab<Rc<RefCell<dyn ProxySession>>> = Slab::with_capacity(max_buffers);
        {
            let entry = sessions.vacant_entry();
            info!("taking token {:?} for channel", entry.key());
            entry.insert(Rc::new(RefCell::new(ListenSession {
                protocol: Protocol::Channel,
            })));
        }
        {
            let entry = sessions.vacant_entry();
            info!("taking token {:?} for timer", entry.key());
            entry.insert(Rc::new(RefCell::new(ListenSession {
                protocol: Protocol::Timer,
            })));
        }
        {
            let entry = sessions.vacant_entry();
            info!("taking token {:?} for metrics", entry.key());
            entry.insert(Rc::new(RefCell::new(ListenSession {
                protocol: Protocol::Metrics,
            })));
        }
        // Test fixture: feature disabled (max_connections_per_ip = 0).
        let sessions = SessionManager::new(sessions, max_buffers, 0, 0);

        let registry = event_loop
            .registry()
            .try_clone()
            .with_context(|| "Failed at creating a registry")?;

        let (scm_server, scm_client) =
            UnixStream::pair().with_context(|| "Failed at creating scm unix stream")?;
        let client_scm_socket = ScmSocket::new(scm_client.into_raw_fd())
            .with_context(|| "Failed at creating the scm client socket")?;
        let server_scm_socket = ScmSocket::new(scm_server.into_raw_fd())
            .with_context(|| "Failed at creating the scm server socket")?;
        if send_scm {
            client_scm_socket
                .send_listeners(&Listeners::default())
                .with_context(|| "Failed at sending empty listeners")?;
        }

        Ok(ServerParts {
            event_loop,
            registry,
            sessions,
            pool,
            backends,
            client_scm_socket,
            server_scm_socket,
            server_config,
        })
    }
}