Merge pull request #596 from xaosproxy/fix/listen_backlog

feat(server): configurable TCP listen_backlog

src/config/defaults.rs

@@ -209,6 +209,10 @@ pub(crate) fn default_server_max_connections() -> u32 {
     10_000
 }
 
+pub(crate) fn default_listen_backlog() -> u32 {
+    1024
+}
+
 pub(crate) fn default_accept_permit_timeout_ms() -> u64 {
     DEFAULT_ACCEPT_PERMIT_TIMEOUT_MS
 }

src/config/hot_reload.rs

@@ -570,6 +570,7 @@ fn warn_non_hot_changes(old: &ProxyConfig, new: &ProxyConfig, non_hot_changed: b
     }
     if old.server.proxy_protocol != new.server.proxy_protocol
         || !listeners_equal(&old.server.listeners, &new.server.listeners)
+        || old.server.listen_backlog != new.server.listen_backlog
         || old.server.listen_addr_ipv4 != new.server.listen_addr_ipv4
         || old.server.listen_addr_ipv6 != new.server.listen_addr_ipv6
         || old.server.listen_tcp != new.server.listen_tcp

src/config/types.rs

@@ -1272,6 +1272,11 @@ pub struct ServerConfig {
     #[serde(default)]
     pub listeners: Vec<ListenerConfig>,
 
+    /// TCP `listen(2)` backlog for client-facing sockets (also used for the metrics HTTP listener).
+    /// The effective queue is capped by the kernel (for example `somaxconn` on Linux).
+    #[serde(default = "default_listen_backlog")]
+    pub listen_backlog: u32,
+
     /// Maximum number of concurrent client connections.
     /// 0 means unlimited.
     #[serde(default = "default_server_max_connections")]
@@ -1300,6 +1305,7 @@ impl Default for ServerConfig {
             metrics_whitelist: default_metrics_whitelist(),
             api: ApiConfig::default(),
             listeners: Vec::new(),
+            listen_backlog: default_listen_backlog(),
             max_connections: default_server_max_connections(),
             accept_permit_timeout_ms: default_accept_permit_timeout_ms(),
         }

src/maestro/listeners.rs

@@ -72,6 +72,7 @@ pub(crate) async fn bind_listeners(
         let options = ListenOptions {
             reuse_port: listener_conf.reuse_allow,
             ipv6_only: listener_conf.ip.is_ipv6(),
+            backlog: config.server.listen_backlog,
             ..Default::default()
         };
 

src/maestro/runtime_tasks.rs

@@ -323,10 +323,12 @@ pub(crate) async fn spawn_metrics_if_configured(
         let config_rx_metrics = config_rx.clone();
         let ip_tracker_metrics = ip_tracker.clone();
         let whitelist = config.server.metrics_whitelist.clone();
+        let listen_backlog = config.server.listen_backlog;
         tokio::spawn(async move {
             metrics::serve(
                 port,
                 listen,
+                listen_backlog,
                 stats,
                 beobachten,
                 ip_tracker_metrics,

src/metrics.rs

@@ -22,6 +22,7 @@ use crate::transport::{ListenOptions, create_listener};
 pub async fn serve(
     port: u16,
     listen: Option<String>,
+    listen_backlog: u32,
     stats: Arc<Stats>,
     beobachten: Arc<BeobachtenStore>,
     ip_tracker: Arc<UserIpTracker>,
@@ -40,7 +41,7 @@ pub async fn serve(
             }
         };
         let is_ipv6 = addr.is_ipv6();
-        match bind_metrics_listener(addr, is_ipv6) {
+        match bind_metrics_listener(addr, is_ipv6, listen_backlog) {
             Ok(listener) => {
                 info!("Metrics endpoint: http://{}/metrics and /beobachten", addr);
                 serve_listener(
@@ -60,7 +61,7 @@ pub async fn serve(
     let mut listener_v6 = None;
 
     let addr_v4 = SocketAddr::from(([0, 0, 0, 0], port));
-    match bind_metrics_listener(addr_v4, false) {
+    match bind_metrics_listener(addr_v4, false, listen_backlog) {
         Ok(listener) => {
             info!(
                 "Metrics endpoint: http://{}/metrics and /beobachten",
@@ -74,7 +75,7 @@ pub async fn serve(
     }
 
     let addr_v6 = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 0], port));
-    match bind_metrics_listener(addr_v6, true) {
+    match bind_metrics_listener(addr_v6, true, listen_backlog) {
         Ok(listener) => {
             info!(
                 "Metrics endpoint: http://[::]:{}/metrics and /beobachten",
@@ -122,10 +123,15 @@ pub async fn serve(
     }
 }
 
-fn bind_metrics_listener(addr: SocketAddr, ipv6_only: bool) -> std::io::Result<TcpListener> {
+fn bind_metrics_listener(
+    addr: SocketAddr,
+    ipv6_only: bool,
+    listen_backlog: u32,
+) -> std::io::Result<TcpListener> {
     let options = ListenOptions {
         reuse_port: false,
         ipv6_only,
+        backlog: listen_backlog,
         ..Default::default()
     };
     let socket = create_listener(addr, &options)?;