Merge pull request #824 from telemt/flow

MSS Tuning

Cargo.lock

@@ -2790,7 +2790,7 @@ checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
 
 [[package]]
 name = "telemt"
-version = "3.4.14"
+version = "3.4.15"
 dependencies = [
  "aes",
  "anyhow",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "telemt"
-version = "3.4.14"
+version = "3.4.15"
 edition = "2024"
 
 [features]

docs/Config_params/CONFIG_PARAMS.en.md

@@ -1805,6 +1805,7 @@ This document lists all configuration keys accepted by `config.toml`.
 | [`listen_unix_sock`](#listen_unix_sock) | `String` | — | `✘` |
 | [`listen_unix_sock_perm`](#listen_unix_sock_perm) | `String` | — | `✘` |
 | [`listen_tcp`](#listen_tcp) | `bool` | — (auto) | `✘` |
+| [`client_mss`](#client_mss) | `String` | `""` | `✘` |
 | [`proxy_protocol`](#proxy_protocol) | `bool` | `false` | `✘` |
 | [`proxy_protocol_header_timeout_ms`](#proxy_protocol_header_timeout_ms) | `u64` | `500` | `✘` |
 | [`proxy_protocol_trusted_cidrs`](#proxy_protocol_trusted_cidrs) | `IpNetwork[]` | `[]` | `✘` |
@@ -1887,6 +1888,16 @@ This document lists all configuration keys accepted by `config.toml`.
     listen_unix_sock = "/run/telemt.sock"
     listen_tcp = true
     ```
+## client_mss
+  - **Constraints / validation**: `String`. Empty or omitted means do not change kernel MSS. Presets: `"extreme-low"` = `88`, `"tspu"` = `92`, `"2in8"` = `256`. Custom decimal strings must be within `88..=4096`.
+  - **Description**: Client-facing TCP MSS applied to TCP listener sockets before `listen(2)`, so Linux can announce it in SYN/ACK. This affects only proxy client TCP listeners, not API, metrics, Unix sockets, Telegram upstreams, ME sockets, or mask backend connections. Changes require listener restart/rebind.
+  - **Performance note**: Low MSS increases packet count predictably. Approximate segment multiplier is `ceil(1460 / client_mss)`.
+  - **Example**:
+
+    ```toml
+    [server]
+    client_mss = "tspu"
+    ```
 ## proxy_protocol
   - **Constraints / validation**: `bool`.
   - **Description**: Enables HAProxy PROXY protocol parsing on incoming connections (PROXY v1/v2). When enabled, client source address is taken from the PROXY header.
@@ -2207,6 +2218,7 @@ Note: This section also accepts the legacy alias `[server.admin_api]` (same sche
 | --- | ---- | ------- | ---------- |
 | [`ip`](#ip) | `IpAddr` | — | `✘` |
 | [`port`](#port-serverlisteners) | `u16` | `server.port` | `✘` |
+| [`client_mss`](#client_mss-serverlisteners) | `String` | `[server].client_mss` | `✘` |
 | [`announce`](#announce) | `String` | — | `✘` |
 | [`announce_ip`](#announce_ip) | `IpAddr` | — | `✘` |
 | [`proxy_protocol`](#proxy_protocol) | `bool` | — | `✘` |
@@ -2231,6 +2243,17 @@ Note: This section also accepts the legacy alias `[server.admin_api]` (same sche
     ip = "0.0.0.0"
     port = 443
     ```
+## client_mss (server.listeners)
+  - **Constraints / validation**: `String` (optional). Same values as `[server].client_mss`.
+  - **Description**: Per-listener MSS override. When omitted, inherits `[server].client_mss`; when set to an empty string, disables MSS shaping for this listener even if the global value is set. Changes require listener restart/rebind.
+  - **Example**:
+
+    ```toml
+    [[server.listeners]]
+    ip = "0.0.0.0"
+    port = 443
+    client_mss = "256"
+    ```
 ## announce
   - **Constraints / validation**: `String` (optional). Must not be empty when set.
   - **Description**: Public IP/domain announced in proxy links for this listener. Takes precedence over `announce_ip`.

docs/Config_params/CONFIG_PARAMS.ru.md

@@ -1807,6 +1807,7 @@
 | [`listen_unix_sock`](#listen_unix_sock) | `String` | — | `✘` |
 | [`listen_unix_sock_perm`](#listen_unix_sock_perm) | `String` | — | `✘` |
 | [`listen_tcp`](#listen_tcp) | `bool` | — (auto) | `✘` |
+| [`client_mss`](#client_mss) | `String` | `""` | `✘` |
 | [`proxy_protocol`](#proxy_protocol) | `bool` | `false` | `✘` |
 | [`proxy_protocol_header_timeout_ms`](#proxy_protocol_header_timeout_ms) | `u64` | `500` | `✘` |
 | [`proxy_protocol_trusted_cidrs`](#proxy_protocol_trusted_cidrs) | `IpNetwork[]` | `[]` | `✘` |
@@ -1889,6 +1890,16 @@
     listen_unix_sock = "/run/telemt.sock"
     listen_tcp = true
     ```
+## client_mss
+  - **Ограничения / валидация**: `String`. Пустое значение или отсутствие параметра означает, что Telemt не изменяет MSS, выбранный ядром. Поддерживаемые presets: `"extreme-low"` = `88`, `"tspu"` = `92`, `"2in8"` = `256`. Пользовательское десятичное значение должно быть строкой в диапазоне `88..=4096`.
+  - **Описание**: MSS для входящих TCP-соединений клиентов. Значение применяется к TCP listener-сокетам до `listen(2)`, чтобы Linux мог объявить его в SYN/ACK. Параметр влияет только на proxy client TCP listeners и не применяется к API, metrics, Unix sockets, Telegram upstreams, ME sockets или mask backend connections. Изменение требует restart/rebind listener’ов.
+  - **Performance note**: Низкий MSS предсказуемо увеличивает количество TCP-сегментов. Приблизительный multiplier: `ceil(1460 / client_mss)`.
+  - **Пример**:
+
+    ```toml
+    [server]
+    client_mss = "tspu"
+    ```
 ## proxy_protocol
   - **Ограничения / валидация**: `bool`.
   - **Описание**: Включает поддержку разбора PROXY protocol от HAProxy (v1/v2) на входящих соединениях. При включении исходный IP клиента берётся из PROXY-заголовка.
@@ -2213,6 +2224,7 @@
 | --- | ---- | ------- | ---------- |
 | [`ip`](#ip) | `IpAddr` | — | `✘` |
 | [`port`](#port-serverlisteners) | `u16` | `server.port` | `✘` |
+| [`client_mss`](#client_mss-serverlisteners) | `String` | `[server].client_mss` | `✘` |
 | [`announce`](#announce) | `String` | — | `✘` |
 | [`announce_ip`](#announce_ip) | `IpAddr` | — | `✘` |
 | [`proxy_protocol`](#proxy_protocol) | `bool` | — | `✘` |
@@ -2237,6 +2249,17 @@
     ip = "0.0.0.0"
     port = 443
     ```
+## client_mss (server.listeners)
+  - **Ограничения / валидация**: `String` (необязательный параметр). Допустимые значения совпадают с `[server].client_mss`.
+  - **Описание**: Per-listener override для MSS. Если параметр не задан, listener наследует `[server].client_mss`; если задана пустая строка, MSS shaping отключается только для этого listener’а, даже когда глобальный параметр задан. Изменение требует restart/rebind listener’а.
+  - **Пример**:
+
+    ```toml
+    [[server.listeners]]
+    ip = "0.0.0.0"
+    port = 443
+    client_mss = "256"
+    ```
 ## announce
   - **Ограничения / валидация**: `String` (необязательный параметр). Не должен быть пустым, если задан.
   - **Описание**: Публичный IP-адрес или домен, объявляемый в proxy-ссылках для данного listener’а. Имеет приоритет над `announce_ip`.

src/config/hot_reload.rs

@@ -312,6 +312,7 @@ fn listeners_equal(
     lhs.iter().zip(rhs.iter()).all(|(a, b)| {
         a.ip == b.ip
             && a.port == b.port
+            && a.client_mss == b.client_mss
             && a.announce == b.announce
             && a.announce_ip == b.announce_ip
             && a.proxy_protocol == b.proxy_protocol
@@ -608,6 +609,7 @@ fn warn_non_hot_changes(old: &ProxyConfig, new: &ProxyConfig, non_hot_changed: b
         || 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
+        || old.server.client_mss != new.server.client_mss
         || old.server.listen_unix_sock != new.server.listen_unix_sock
         || old.server.listen_unix_sock_perm != new.server.listen_unix_sock_perm
     {

src/config/load.rs

@@ -299,6 +299,7 @@ const SERVER_CONFIG_KEYS: &[&str] = &[
     "listen_unix_sock",
     "listen_unix_sock_perm",
     "listen_tcp",
+    "client_mss",
     "proxy_protocol",
     "proxy_protocol_header_timeout_ms",
     "proxy_protocol_trusted_cidrs",
@@ -344,6 +345,7 @@ const CONNTRACK_CONTROL_CONFIG_KEYS: &[&str] = &[
 const LISTENER_CONFIG_KEYS: &[&str] = &[
     "ip",
     "port",
+    "client_mss",
     "announce",
     "announce_ip",
     "proxy_protocol",
@@ -1933,6 +1935,20 @@ impl ProxyConfig {
             ));
         }
 
+        config
+            .server
+            .client_mss_value()
+            .map_err(|error| ProxyError::Config(format!("server.client_mss {error}")))?;
+        for (idx, listener) in config.server.listeners.iter().enumerate() {
+            if listener.client_mss.is_some() {
+                listener
+                    .effective_client_mss(&config.server)
+                    .map_err(|error| {
+                        ProxyError::Config(format!("server.listeners[{idx}].client_mss {error}"))
+                    })?;
+            }
+        }
+
         if config.server.accept_permit_timeout_ms > 60_000 {
             return Err(ProxyError::Config(
                 "server.accept_permit_timeout_ms must be within [0, 60000]".to_string(),
@@ -2173,6 +2189,7 @@ impl ProxyConfig {
                 config.server.listeners.push(ListenerConfig {
                     ip: ipv4,
                     port: Some(config.server.port),
+                    client_mss: None,
                     announce: None,
                     announce_ip: None,
                     proxy_protocol: None,
@@ -2185,6 +2202,7 @@ impl ProxyConfig {
                 config.server.listeners.push(ListenerConfig {
                     ip: ipv6,
                     port: Some(config.server.port),
+                    client_mss: None,
                     announce: None,
                     announce_ip: None,
                     proxy_protocol: None,
@@ -2460,6 +2478,7 @@ mod tests {
         assert_eq!(cfg.general.update_every, default_update_every());
         assert_eq!(cfg.server.listen_addr_ipv4, default_listen_addr_ipv4());
         assert_eq!(cfg.server.listen_addr_ipv6, default_listen_addr_ipv6_opt());
+        assert_eq!(cfg.server.client_mss_value(), Ok(None));
         assert_eq!(
             cfg.server.proxy_protocol_trusted_cidrs,
             default_proxy_protocol_trusted_cidrs()
@@ -3787,6 +3806,153 @@ mod tests {
         let _ = std::fs::remove_file(path);
     }
 
+    #[test]
+    fn client_mss_presets_and_listener_override_are_resolved() {
+        let toml = r#"
+            [server]
+            client_mss = "tspu"
+
+            [[server.listeners]]
+            ip = "127.0.0.1"
+            port = 1443
+
+            [[server.listeners]]
+            ip = "127.0.0.2"
+            port = 1444
+            client_mss = "2in8"
+
+            [[server.listeners]]
+            ip = "127.0.0.3"
+            port = 1445
+            client_mss = ""
+
+            [[server.listeners]]
+            ip = "127.0.0.4"
+            port = 1446
+            client_mss = "extreme-low"
+
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+        "#;
+        let dir = std::env::temp_dir();
+        let path = dir.join("telemt_client_mss_valid_test.toml");
+        std::fs::write(&path, toml).unwrap();
+        let cfg = ProxyConfig::load(&path).unwrap();
+
+        assert_eq!(cfg.server.client_mss_value(), Ok(Some(92)));
+        assert_eq!(
+            cfg.server.listeners[0].effective_client_mss(&cfg.server),
+            Ok(Some(92))
+        );
+        assert_eq!(
+            cfg.server.listeners[1].effective_client_mss(&cfg.server),
+            Ok(Some(256))
+        );
+        assert_eq!(
+            cfg.server.listeners[2].effective_client_mss(&cfg.server),
+            Ok(None)
+        );
+        assert_eq!(
+            cfg.server.listeners[3].effective_client_mss(&cfg.server),
+            Ok(Some(88))
+        );
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[test]
+    fn client_mss_custom_value_is_accepted() {
+        let toml = r#"
+            [server]
+            client_mss = "4096"
+
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+        "#;
+        let dir = std::env::temp_dir();
+        let path = dir.join("telemt_client_mss_custom_valid_test.toml");
+        std::fs::write(&path, toml).unwrap();
+        let cfg = ProxyConfig::load(&path).unwrap();
+
+        assert_eq!(cfg.server.client_mss_value(), Ok(Some(4096)));
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[test]
+    fn client_mss_out_of_range_is_rejected() {
+        for value in ["87", "4097"] {
+            let toml = format!(
+                r#"
+                [server]
+                client_mss = "{value}"
+
+                [censorship]
+                tls_domain = "example.com"
+
+                [access.users]
+                user = "00000000000000000000000000000000"
+            "#
+            );
+            let dir = std::env::temp_dir();
+            let path = dir.join(format!("telemt_client_mss_out_of_range_{value}_test.toml"));
+            std::fs::write(&path, toml).unwrap();
+            let err = ProxyConfig::load(&path).unwrap_err().to_string();
+
+            assert!(err.contains("server.client_mss custom value must be within [88, 4096]"));
+            let _ = std::fs::remove_file(path);
+        }
+    }
+
+    #[test]
+    fn client_mss_unquoted_number_is_rejected() {
+        let toml = r#"
+            [server]
+            client_mss = 256
+
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+        "#;
+        let dir = std::env::temp_dir();
+        let path = dir.join("telemt_client_mss_unquoted_number_test.toml");
+        std::fs::write(&path, toml).unwrap();
+        let err = ProxyConfig::load(&path).unwrap_err().to_string();
+
+        assert!(err.contains("client_mss"));
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[test]
+    fn listener_client_mss_invalid_preset_is_rejected() {
+        let toml = r#"
+            [[server.listeners]]
+            ip = "127.0.0.1"
+            port = 1443
+            client_mss = "tiny"
+
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+        "#;
+        let dir = std::env::temp_dir();
+        let path = dir.join("telemt_listener_client_mss_invalid_test.toml");
+        std::fs::write(&path, toml).unwrap();
+        let err = ProxyConfig::load(&path).unwrap_err().to_string();
+
+        assert!(err.contains("server.listeners[0].client_mss"));
+        assert!(err.contains("must be \"\", extreme-low, tspu, 2in8"));
+        let _ = std::fs::remove_file(path);
+    }
+
     #[test]
     fn api_runtime_edge_cache_ttl_out_of_range_is_rejected() {
         let toml = r#"

src/config/types.rs

@@ -1451,6 +1451,11 @@ pub struct ServerConfig {
     #[serde(default)]
     pub listen_tcp: Option<bool>,
 
+    /// Client-facing TCP MSS preset or custom value for all TCP listeners.
+    /// Empty string or omitted value keeps the kernel default.
+    #[serde(default)]
+    pub client_mss: Option<String>,
+
     /// Accept HAProxy PROXY protocol headers on incoming connections.
     /// When enabled, real client IPs are extracted from PROXY v1/v2 headers.
     #[serde(default)]
@@ -1517,6 +1522,7 @@ impl Default for ServerConfig {
             listen_unix_sock: None,
             listen_unix_sock_perm: None,
             listen_tcp: None,
+            client_mss: None,
             proxy_protocol: false,
             proxy_protocol_header_timeout_ms: default_proxy_protocol_header_timeout_ms(),
             proxy_protocol_trusted_cidrs: default_proxy_protocol_trusted_cidrs(),
@@ -2087,6 +2093,10 @@ pub struct ListenerConfig {
     /// Per-listener TCP port. If omitted, falls back to legacy `server.port`.
     #[serde(default)]
     pub port: Option<u16>,
+    /// Per-listener client-facing TCP MSS preset or custom value.
+    /// Empty string disables MSS shaping for this listener.
+    #[serde(default)]
+    pub client_mss: Option<String>,
     /// IP address or hostname to announce in proxy links.
     /// Takes precedence over `announce_ip` if both are set.
     #[serde(default)]
@@ -2104,6 +2114,64 @@ pub struct ListenerConfig {
     pub reuse_allow: bool,
 }
 
+/// Client-facing TCP MSS preset for extreme-low fragmentation profiles.
+pub const CLIENT_MSS_EXTREME_LOW: u16 = 88;
+/// Client-facing TCP MSS preset matching TSPU-oriented deployments.
+pub const CLIENT_MSS_TSPU: u16 = 92;
+/// Client-facing TCP MSS preset for 2-in-8 segment shaping.
+pub const CLIENT_MSS_2IN8: u16 = 256;
+/// Minimum accepted custom client-facing TCP MSS value.
+pub const CLIENT_MSS_MIN: u16 = CLIENT_MSS_EXTREME_LOW;
+/// Maximum accepted custom client-facing TCP MSS value.
+pub const CLIENT_MSS_MAX: u16 = 4096;
+
+impl ServerConfig {
+    /// Resolves the global client-facing TCP MSS setting.
+    pub fn client_mss_value(&self) -> std::result::Result<Option<u16>, String> {
+        parse_client_mss(self.client_mss.as_deref())
+    }
+}
+
+impl ListenerConfig {
+    /// Resolves the listener MSS override, falling back to the global server value.
+    pub fn effective_client_mss(
+        &self,
+        server: &ServerConfig,
+    ) -> std::result::Result<Option<u16>, String> {
+        match self.client_mss.as_deref() {
+            Some(value) => parse_client_mss(Some(value)),
+            None => server.client_mss_value(),
+        }
+    }
+}
+
+fn parse_client_mss(raw: Option<&str>) -> std::result::Result<Option<u16>, String> {
+    let Some(raw) = raw else {
+        return Ok(None);
+    };
+    let value = raw.trim();
+    if value.is_empty() {
+        return Ok(None);
+    }
+
+    match value.to_ascii_lowercase().as_str() {
+        "extreme-low" => return Ok(Some(CLIENT_MSS_EXTREME_LOW)),
+        "tspu" => return Ok(Some(CLIENT_MSS_TSPU)),
+        "2in8" => return Ok(Some(CLIENT_MSS_2IN8)),
+        _ => {}
+    }
+
+    let parsed = value
+        .parse::<u16>()
+        .map_err(|_| "must be \"\", extreme-low, tspu, 2in8, or a decimal value".to_string())?;
+    if !(CLIENT_MSS_MIN..=CLIENT_MSS_MAX).contains(&parsed) {
+        return Err(format!(
+            "custom value must be within [{CLIENT_MSS_MIN}, {CLIENT_MSS_MAX}]"
+        ));
+    }
+    Ok(Some(parsed))
+}
+
 // ============= ShowLink =============
 
 /// Controls which users' proxy links are displayed at startup.

src/maestro/listeners.rs

@@ -47,6 +47,10 @@ fn default_link_port(config: &ProxyConfig) -> u16 {
         .unwrap_or(config.server.port)
 }
 
+fn mss_segment_multiplier(client_mss: u16) -> u16 {
+    1460u16.div_ceil(client_mss)
+}
+
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn bind_listeners(
     config: &Arc<ProxyConfig>,
@@ -90,10 +94,22 @@ pub(crate) async fn bind_listeners(
             warn!(%addr, "Skipping IPv6 listener: IPv6 disabled by [network]");
             continue;
         }
+        let client_mss = match listener_conf.effective_client_mss(&config.server) {
+            Ok(value) => value,
+            Err(error) => {
+                warn!(
+                    %addr,
+                    error = %error,
+                    "Invalid listener client MSS after config validation; using kernel default"
+                );
+                None
+            }
+        };
         let options = ListenOptions {
             reuse_port: listener_conf.reuse_allow,
             ipv6_only: listener_conf.ip.is_ipv6(),
             backlog: config.server.listen_backlog,
+            client_mss,
             ..Default::default()
         };
 
@@ -101,6 +117,14 @@ pub(crate) async fn bind_listeners(
             Ok(socket) => {
                 let listener = TcpListener::from_std(socket.into())?;
                 info!("Listening on {}", addr);
+                if let Some(client_mss) = client_mss {
+                    info!(
+                        %addr,
+                        client_mss,
+                        segment_multiplier = mss_segment_multiplier(client_mss),
+                        "Client-facing TCP MSS configured"
+                    );
+                }
                 let listener_proxy_protocol = listener_conf
                     .proxy_protocol
                     .unwrap_or(config.server.proxy_protocol);

src/transport/socket.rs

@@ -9,7 +9,7 @@ use std::io::Result;
 use std::net::{IpAddr, SocketAddr};
 use std::time::Duration;
 use tokio::net::TcpStream;
-use tracing::debug;
+use tracing::{debug, warn};
 
 const DEFAULT_SOCKET_BUFFER_BYTES: usize = 256 * 1024;
 
@@ -283,6 +283,8 @@ pub struct ListenOptions {
     pub backlog: u32,
     /// IPv6 only (disable dual-stack)
     pub ipv6_only: bool,
+    /// Client-facing TCP MSS to announce on accepted TCP sessions.
+    pub client_mss: Option<u16>,
 }
 
 impl Default for ListenOptions {
@@ -292,6 +294,7 @@ impl Default for ListenOptions {
             reuse_port: true,
             backlog: 1024,
             ipv6_only: false,
+            client_mss: None,
         }
     }
 }
@@ -319,6 +322,19 @@ pub fn create_listener(addr: SocketAddr, options: &ListenOptions) -> Result<Sock
         socket.set_only_v6(true)?;
     }
 
+    if let Some(client_mss) = options.client_mss {
+        if let Err(error) = socket.set_tcp_mss(u32::from(client_mss)) {
+            warn!(
+                addr = %addr,
+                client_mss,
+                error = %error,
+                "Failed to apply listener client MSS; continuing with kernel default"
+            );
+        } else {
+            debug!(addr = %addr, client_mss, "Applied listener client MSS");
+        }
+    }
+
     socket.set_nonblocking(true)?;
     socket.bind(&addr.into())?;
     socket.listen(options.backlog as i32)?;
@@ -637,5 +653,28 @@ mod tests {
         assert!(opts.reuse_addr);
         assert!(opts.reuse_port);
         assert_eq!(opts.backlog, 1024);
+        assert_eq!(opts.client_mss, None);
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn test_create_listener_applies_client_mss() {
+        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
+        let options = ListenOptions {
+            reuse_port: false,
+            client_mss: Some(256),
+            ..Default::default()
+        };
+        let socket = match create_listener(addr, &options) {
+            Ok(socket) => socket,
+            Err(e) if e.kind() == ErrorKind::PermissionDenied => return,
+            Err(e) => panic!("create_listener failed: {e}"),
+        };
+        let mss = match socket.tcp_mss() {
+            Ok(mss) => mss,
+            Err(e) if e.kind() == ErrorKind::PermissionDenied => return,
+            Err(e) => panic!("tcp_mss failed: {e}"),
+        };
+        assert_eq!(mss, 256);
     }
 }