Merge pull request #871 from xaer981/fix/client-mss-bulk

fix(server): client_mss_bulk — fragment only ServerHello, keep bulk MSS from start

docs/Config_params/CONFIG_PARAMS.en.md

@@ -1971,7 +1971,7 @@ This document lists all configuration keys accepted by `config.toml`.
     ```
 ## 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.
+  - **Description**: Controls the segment size used for client connections. By default, this value is applied to the TCP listener and remains active for the whole connection. When `client_mss_bulk` is also set on Linux, `client_mss` is used only as the maximum chunk size for the initial authenticated FakeTLS response (`ServerHello`). This setting does not affect API, metrics, Unix sockets, Telegram upstreams, ME sockets, or mask backend connections. Changes require a listener restart/rebind.
   - **Operator note**: The two-tier `synlimit` profile does not require Telemt to disable MSS automatically. Operators that follow external host-tuning recipes should decide explicitly whether to leave MSS shaping enabled for handshake fragmentation or disable it for higher media throughput.
   - **Performance note**: Low MSS increases packet count predictably. Approximate segment multiplier is `ceil(1460 / client_mss)`.
   - **Example**:
@@ -1982,7 +1982,7 @@ This document lists all configuration keys accepted by `config.toml`.
     ```
 ## client_mss_bulk
   - **Constraints / validation**: `String`. Same grammar as [`client_mss`](#client_mss) (empty/omitted, presets `"extreme-low"`/`"tspu"`/`"2in8"`, or a decimal in `88..=4096`).
-  - **Description**: Optional bulk-phase MSS. When set, the low `client_mss` is applied only while the TLS handshake (including the DPI-inspected ServerHello) is sent; once the connection transitions to relaying, the client socket MSS is raised to `client_mss_bulk` for the bulk data phase. This keeps the anti-DPI handshake fragmentation but restores normal-size packets for payload, cutting outgoing packets-per-second by roughly the `client_mss` segment multiplier (e.g. ~10x with `"tspu"`). Useful on hosts whose abuse detection counts packets-per-second rather than bandwidth. When empty/omitted, the handshake MSS is kept for the whole connection (previous behavior). Linux only; a no-op elsewhere.
+  - **Description**: Enables separated MSS handling for `ServerHello` and for all other packets. The listener uses `client_mss_bulk` from the start of the connection, including when the client sends `ClientHello`. Telemt sends the initial authenticated FakeTLS response (`ServerHello`) in chunks no larger than `client_mss`. Normal MTProto traffic then continues with `client_mss_bulk`, without changing MSS on an established connection. This keeps the smaller `ServerHello` chunks while avoiding small segments for all later traffic. When this option is empty or omitted, `client_mss` applies to the whole connection. **Linux-only**.
   - **Example**:
 
     ```toml

docs/Config_params/CONFIG_PARAMS.ru.md

@@ -1893,7 +1893,7 @@
     ```
 ## 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’ов.
+  - **Описание**: Управляет размером сегментов в клиентских соединениях. По умолчанию значение применяется к TCP listener и действует в течение всего соединения. Если также задан `client_mss_bulk`, параметр `client_mss` используется только как максимальный размер фрагмента начального аутентифицированного FakeTLS-ответа (`ServerHello`). Настройка не влияет на API, metrics, Unix sockets, Telegram upstreams, ME sockets и mask backend connections. Изменение требует перезапуска или повторного создания listener.
   - **Operator note**: Two-tier `synlimit` profile больше не требует автоматического отключения MSS внутри Telemt. Оператор должен сам решить, оставлять MSS shaping для handshake fragmentation или отключать его ради более высокой скорости media.
   - **Performance note**: Низкий MSS предсказуемо увеличивает количество TCP-сегментов. Приблизительный multiplier: `ceil(1460 / client_mss)`.
   - **Пример**:
@@ -1904,7 +1904,7 @@
     ```
 ## client_mss_bulk
   - **Ограничения / валидация**: `String`. Грамматика та же, что у [`client_mss`](#client_mss) (пусто/не задано, пресеты `"extreme-low"`/`"tspu"`/`"2in8"` либо десятичное число в диапазоне `88..=4096`).
-  - **Описание**: Необязательный MSS для bulk-фазы. Если задан, низкий `client_mss` применяется только на время TLS-handshake (включая инспектируемый DPI ServerHello); как только соединение переходит в фазу relay, MSS клиентского сокета поднимается до `client_mss_bulk` для передачи полезной нагрузки. Так сохраняется anti-DPI фрагментация handshake, но для данных возвращаются пакеты нормального размера — это снижает исходящий packets-per-second примерно во столько раз, каков segment multiplier у `client_mss` (например, ~10x для `"tspu"`). Полезно на хостингах, где abuse-детекция считает packets-per-second, а не полосу. Если пусто/не задано — MSS handshake сохраняется на всё соединение (прежнее поведение). Только Linux; на прочих платформах — no-op.
+  - **Описание**: Включает раздельную работу MSS для `ServerHello` и остального трафика. Listener использует `client_mss_bulk` с самого начала соединения, в том числе при получении `ClientHello`. Начальный аутентифицированный FakeTLS-ответ (`ServerHello`) Telemt отправляет фрагментами не больше `client_mss`. После этого обычный MTProto-трафик продолжает работать с `client_mss_bulk`, без изменения MSS уже установленного соединения. Так `ServerHello` остаётся разбит на небольшие части, а весь последующий трафик не ограничивается маленькими сегментами. Если параметр пуст или не задан, `client_mss` действует в течение всего соединения. **Работает только в Linux.**
   - **Пример**:
 
     ```toml

src/config/types.rs

@@ -1619,12 +1619,11 @@ pub struct ServerConfig {
     #[serde(default)]
     pub client_mss: Option<String>,
 
-    /// Client-facing TCP MSS to switch to AFTER the TLS handshake (ServerHello)
-    /// is sent. Lets `client_mss` fragment ONLY the handshake (the DPI-inspected
-    /// part) while the bulk transfer uses normal-size packets — avoids the ~10x
-    /// packets-per-second blowup that triggers anti-DDoS abuse blocks on
-    /// pps-policing hosts. Empty/omitted = keep the handshake MSS for the whole
-    /// connection (previous behavior). Same preset/int grammar as `client_mss`.
+    /// Client-facing TCP MSS for bulk traffic when initial-response fragmentation
+    /// is enabled on Linux. The listener uses this MSS from connection setup;
+    /// `client_mss` becomes the fragment size for the authenticated FakeTLS
+    /// response. Empty/omitted keeps `client_mss` connection-wide. Uses the same
+    /// preset/integer grammar as `client_mss`.
     #[serde(default)]
     pub client_mss_bulk: Option<String>,
 
@@ -2484,7 +2483,7 @@ impl ServerConfig {
         parse_client_mss(self.client_mss.as_deref())
     }
 
-    /// Resolves the post-handshake (bulk transfer) client MSS, if configured.
+    /// Resolves the bulk-transfer client MSS, if configured.
     pub fn client_mss_bulk_value(&self) -> std::result::Result<Option<u16>, String> {
         parse_client_mss(self.client_mss_bulk.as_deref())
     }

src/maestro/listeners.rs

@@ -30,10 +30,17 @@ use super::helpers::{
 };
 
 pub(crate) struct BoundListeners {
-    pub(crate) listeners: Vec<(TcpListener, bool)>,
+    pub(crate) listeners: Vec<BoundTcpListener>,
     pub(crate) has_unix_listener: bool,
 }
 
+/// A TCP listener and the connection settings fixed when it was bound.
+pub(crate) struct BoundTcpListener {
+    listener: TcpListener,
+    proxy_protocol: bool,
+    tls_response_fragment_size: Option<u16>,
+}
+
 fn listener_port_or_legacy(listener: &crate::config::ListenerConfig, config: &ProxyConfig) -> u16 {
     listener.port.unwrap_or(config.server.port)
 }
@@ -51,6 +58,18 @@ fn mss_segment_multiplier(client_mss: u16) -> u16 {
     1460u16.div_ceil(client_mss)
 }
 
+#[cfg(any(target_os = "linux", test))]
+fn tcp_mss_runtime_profile(
+    handshake_mss: Option<u16>,
+    bulk_mss: Option<u16>,
+) -> (Option<u16>, Option<u16>) {
+    match (handshake_mss, bulk_mss) {
+        (Some(fragment_size), Some(listener_mss)) => (Some(listener_mss), Some(fragment_size)),
+        (listener_mss, None) => (listener_mss, None),
+        (None, Some(_)) => (None, None),
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn bind_listeners(
     config: &Arc<ProxyConfig>,
@@ -82,6 +101,16 @@ pub(crate) async fn bind_listeners(
         )
         .await;
     let mut listeners = Vec::new();
+    let bulk_client_mss = match config.server.client_mss_bulk_value() {
+        Ok(value) => value,
+        Err(error) => {
+            warn!(
+                error = %error,
+                "Invalid bulk client MSS after config validation; disabling bulk MSS"
+            );
+            None
+        }
+    };
 
     for listener_conf in &config.server.listeners {
         let listener_port = listener_port_or_legacy(listener_conf, config);
@@ -94,7 +123,7 @@ 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) {
+        let configured_client_mss = match listener_conf.effective_client_mss(&config.server) {
             Ok(value) => value,
             Err(error) => {
                 warn!(
@@ -105,6 +134,11 @@ pub(crate) async fn bind_listeners(
                 None
             }
         };
+        #[cfg(target_os = "linux")]
+        let (client_mss, tls_response_fragment_size) =
+            tcp_mss_runtime_profile(configured_client_mss, bulk_client_mss);
+        #[cfg(not(target_os = "linux"))]
+        let (client_mss, tls_response_fragment_size) = (configured_client_mss, None);
         let options = ListenOptions {
             reuse_port: listener_conf.reuse_allow,
             ipv6_only: listener_conf.ip.is_ipv6(),
@@ -125,6 +159,15 @@ pub(crate) async fn bind_listeners(
                         "Client-facing TCP MSS configured"
                     );
                 }
+                if let Some(fragment_size) = tls_response_fragment_size {
+                    info!(
+                        %addr,
+                        fragment_size,
+                        bulk_mss = client_mss,
+                        segment_multiplier = mss_segment_multiplier(fragment_size),
+                        "Initial FakeTLS response fragmentation configured"
+                    );
+                }
                 let listener_proxy_protocol = listener_conf
                     .proxy_protocol
                     .unwrap_or(config.server.proxy_protocol);
@@ -152,7 +195,11 @@ pub(crate) async fn bind_listeners(
                     print_proxy_links(&public_host, link_port, config);
                 }
 
-                listeners.push((listener, listener_proxy_protocol));
+                listeners.push(BoundTcpListener {
+                    listener,
+                    proxy_protocol: listener_proxy_protocol,
+                    tls_response_fragment_size,
+                });
             }
             Err(e) => {
                 if e.kind() == std::io::ErrorKind::AddrInUse {
@@ -387,7 +434,7 @@ pub(crate) async fn bind_listeners(
 
 #[allow(clippy::too_many_arguments)]
 pub(crate) fn spawn_tcp_accept_loops(
-    listeners: Vec<(TcpListener, bool)>,
+    listeners: Vec<BoundTcpListener>,
     config_rx: watch::Receiver<Arc<ProxyConfig>>,
     admission_rx: watch::Receiver<bool>,
     stats: Arc<Stats>,
@@ -404,7 +451,10 @@ pub(crate) fn spawn_tcp_accept_loops(
     shared: Arc<ProxySharedState>,
     max_connections: Arc<Semaphore>,
 ) {
-    for (listener, listener_proxy_protocol) in listeners {
+    for bound_listener in listeners {
+        let listener = bound_listener.listener;
+        let listener_proxy_protocol = bound_listener.proxy_protocol;
+        let tls_response_fragment_size = bound_listener.tls_response_fragment_size;
         let mut config_rx: watch::Receiver<Arc<ProxyConfig>> = config_rx.clone();
         let admission_rx_tcp = admission_rx.clone();
         let stats = stats.clone();
@@ -513,6 +563,7 @@ pub(crate) fn spawn_tcp_accept_loops(
                                 #[cfg(unix)]
                                 raw_fd,
                                 rst_mode,
+                                tls_response_fragment_size,
                             )
                             .run()
                             .await
@@ -603,3 +654,26 @@ pub(crate) fn spawn_tcp_accept_loops(
         });
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::tcp_mss_runtime_profile;
+
+    #[test]
+    fn client_mss_without_bulk_remains_connection_wide() {
+        assert_eq!(tcp_mss_runtime_profile(Some(92), None), (Some(92), None));
+    }
+
+    #[test]
+    fn client_mss_with_bulk_uses_bulk_listener_and_fragments_initial_response() {
+        assert_eq!(
+            tcp_mss_runtime_profile(Some(92), Some(1400)),
+            (Some(1400), Some(92))
+        );
+    }
+
+    #[test]
+    fn bulk_mss_without_handshake_mss_does_not_enable_shaping() {
+        assert_eq!(tcp_mss_runtime_profile(None, Some(1400)), (None, None));
+    }
+}

src/proxy/client.rs

@@ -109,7 +109,8 @@ use crate::transport::{UpstreamManager, configure_client_socket, parse_proxy_pro
 
 use crate::proxy::direct_relay::handle_via_direct_with_shared;
 use crate::proxy::handshake::{
-    HandshakeSuccess, handle_mtproto_handshake_with_shared, handle_tls_handshake_with_shared,
+    HandshakeSuccess, TlsResponseWriteOptions, handle_mtproto_handshake_with_shared,
+    handle_tls_handshake_with_shared, handle_tls_handshake_with_shared_and_options,
 };
 #[cfg(test)]
 use crate::proxy::handshake::{handle_mtproto_handshake, handle_tls_handshake};
@@ -989,6 +990,7 @@ pub struct RunningClientHandler {
     #[cfg(unix)]
     raw_fd: std::os::unix::io::RawFd,
     rst_on_close: crate::config::RstOnCloseMode,
+    tls_response_fragment_size: Option<u16>,
 }
 
 impl ClientHandler {
@@ -1036,6 +1038,7 @@ impl ClientHandler {
             #[cfg(unix)]
             raw_fd,
             crate::config::RstOnCloseMode::Off,
+            None,
         )
     }
 
@@ -1060,6 +1063,7 @@ impl ClientHandler {
         real_peer_report: Arc<std::sync::Mutex<Option<SocketAddr>>>,
         #[cfg(unix)] raw_fd: std::os::unix::io::RawFd,
         rst_on_close: crate::config::RstOnCloseMode,
+        tls_response_fragment_size: Option<u16>,
     ) -> RunningClientHandler {
         let normalized_peer = normalize_ip(peer);
         RunningClientHandler {
@@ -1084,6 +1088,7 @@ impl ClientHandler {
             #[cfg(unix)]
             raw_fd,
             rst_on_close,
+            tls_response_fragment_size,
         }
     }
 }
@@ -1105,12 +1110,6 @@ impl RunningClientHandler {
         #[cfg(unix)]
         let raw_fd = self.raw_fd;
         let rst_on_close = self.rst_on_close;
-        // MSS for the bulk data phase: once the handshake (incl. ServerHello) is
-        // sent, restore a normal MSS so only the handshake stays fragmented by the
-        // low listener `client_mss`. Cuts pps ~10x (anti-DDoS abuse on pps-policing
-        // hosts like FastVPS). None = keep handshake MSS for the whole connection.
-        #[cfg(unix)]
-        let bulk_mss: Option<u16> = self.config.server.client_mss_bulk_value().ok().flatten();
 
         let outcome = match self.do_handshake().await? {
             Some(outcome) => outcome,
@@ -1124,14 +1123,6 @@ impl RunningClientHandler {
                 if matches!(rst_on_close, crate::config::RstOnCloseMode::Errors) {
                     let _ = crate::transport::socket::clear_linger_fd(raw_fd);
                 }
-                // Handshake (ServerHello) done — raise MSS for bulk transfer.
-                #[cfg(unix)]
-                if let Some(mss) = bulk_mss {
-                    if let Err(e) = crate::transport::socket::set_tcp_mss_fd(raw_fd, u32::from(mss))
-                    {
-                        debug!(error = %e, "Failed to raise bulk MSS; keeping handshake MSS");
-                    }
-                }
                 fut.await
             }
             HandshakeOutcome::NeedsMasking(fut) => fut.await,
@@ -1412,50 +1403,58 @@ impl RunningClientHandler {
 
         let (read_half, write_half) = self.stream.into_split();
 
-        let (mut tls_reader, tls_writer, tls_user) = match handle_tls_handshake_with_shared(
-            &handshake,
-            read_half,
-            write_half,
-            peer,
-            &config,
-            &replay_checker,
-            &self.rng,
-            self.tls_cache.clone(),
-            self.shared.as_ref(),
-        )
-        .await
-        {
-            HandshakeResult::Success(result) => result,
-            HandshakeResult::BadClient { reader, writer } => {
-                stats.increment_connects_bad_with_class("tls_handshake_bad_client");
-                record_tls_fingerprint_bad_or_probe(
-                    stats.as_ref(),
-                    &config,
-                    peer.ip(),
-                    tls_fingerprint.as_ref(),
-                );
-                return Ok(masking_outcome(
-                    reader,
-                    writer,
-                    handshake.clone(),
-                    peer,
-                    local_addr,
-                    config.clone(),
-                    self.beobachten.clone(),
-                    self.shared.clone(),
-                ));
-            }
-            HandshakeResult::Error(e) => {
-                record_tls_fingerprint_bad_or_probe(
-                    stats.as_ref(),
-                    &config,
-                    peer.ip(),
-                    tls_fingerprint.as_ref(),
-                );
-                increment_bad_on_unknown_tls_sni(stats.as_ref(), &e);
-                return Err(e);
-            }
-        };
+        #[cfg(target_os = "linux")]
+        let response_write_options =
+            TlsResponseWriteOptions::tcp(self.raw_fd, self.tls_response_fragment_size);
+        #[cfg(not(target_os = "linux"))]
+        let response_write_options = TlsResponseWriteOptions::default();
+
+        let (mut tls_reader, tls_writer, tls_user) =
+            match handle_tls_handshake_with_shared_and_options(
+                &handshake,
+                read_half,
+                write_half,
+                peer,
+                &config,
+                &replay_checker,
+                &self.rng,
+                self.tls_cache.clone(),
+                self.shared.as_ref(),
+                response_write_options,
+            )
+            .await
+            {
+                HandshakeResult::Success(result) => result,
+                HandshakeResult::BadClient { reader, writer } => {
+                    stats.increment_connects_bad_with_class("tls_handshake_bad_client");
+                    record_tls_fingerprint_bad_or_probe(
+                        stats.as_ref(),
+                        &config,
+                        peer.ip(),
+                        tls_fingerprint.as_ref(),
+                    );
+                    return Ok(masking_outcome(
+                        reader,
+                        writer,
+                        handshake.clone(),
+                        peer,
+                        local_addr,
+                        config.clone(),
+                        self.beobachten.clone(),
+                        self.shared.clone(),
+                    ));
+                }
+                HandshakeResult::Error(e) => {
+                    record_tls_fingerprint_bad_or_probe(
+                        stats.as_ref(),
+                        &config,
+                        peer.ip(),
+                        tls_fingerprint.as_ref(),
+                    );
+                    increment_bad_on_unknown_tls_sni(stats.as_ref(), &e);
+                    return Err(e);
+                }
+            };
         record_tls_fingerprint_auth_success(
             stats.as_ref(),
             &config,

src/proxy/handshake.rs

@@ -59,6 +59,26 @@ const OVERLOAD_CANDIDATE_BUDGET_UNHINTED: usize = 8;
 const EXPENSIVE_INVALID_SCAN_SATURATION_THRESHOLD: usize = 64;
 const RECENT_USER_RING_SCAN_LIMIT: usize = 32;
 
+/// Controls how the authenticated FakeTLS response is written to a client.
+#[derive(Clone, Copy, Debug, Default)]
+pub(crate) struct TlsResponseWriteOptions {
+    #[cfg(target_os = "linux")]
+    socket_fd: Option<std::os::unix::io::RawFd>,
+    #[cfg(target_os = "linux")]
+    fragment_size: Option<u16>,
+}
+
+impl TlsResponseWriteOptions {
+    /// Creates Linux TCP response fragmentation options for an accepted socket.
+    #[cfg(target_os = "linux")]
+    pub(crate) fn tcp(fd: std::os::unix::io::RawFd, fragment_size: Option<u16>) -> Self {
+        Self {
+            socket_fd: fragment_size.map(|_| fd),
+            fragment_size,
+        }
+    }
+}
+
 #[cfg(test)]
 const AUTH_PROBE_BACKOFF_BASE_MS: u64 = 1;
 #[cfg(not(test))]
@@ -925,6 +945,7 @@ where
         rng,
         tls_cache,
         shared.as_ref(),
+        TlsResponseWriteOptions::default(),
     )
     .await
 }
@@ -954,6 +975,39 @@ where
         rng,
         tls_cache,
         shared,
+        TlsResponseWriteOptions::default(),
+    )
+    .await
+}
+
+/// Handles FakeTLS with optional initial-response fragmentation on a TCP socket.
+pub(crate) async fn handle_tls_handshake_with_shared_and_options<R, W>(
+    handshake: &[u8],
+    reader: R,
+    writer: W,
+    peer: SocketAddr,
+    config: &ProxyConfig,
+    replay_checker: &ReplayChecker,
+    rng: &SecureRandom,
+    tls_cache: Option<Arc<TlsFrontCache>>,
+    shared: &ProxySharedState,
+    response_write_options: TlsResponseWriteOptions,
+) -> HandshakeResult<(FakeTlsReader<R>, FakeTlsWriter<W>, String), R, W>
+where
+    R: AsyncRead + Unpin,
+    W: AsyncWrite + Unpin,
+{
+    handle_tls_handshake_impl(
+        handshake,
+        reader,
+        writer,
+        peer,
+        config,
+        replay_checker,
+        rng,
+        tls_cache,
+        shared,
+        response_write_options,
     )
     .await
 }
@@ -968,6 +1022,7 @@ async fn handle_tls_handshake_impl<R, W>(
     rng: &SecureRandom,
     tls_cache: Option<Arc<TlsFrontCache>>,
     shared: &ProxySharedState,
+    response_write_options: TlsResponseWriteOptions,
 ) -> HandshakeResult<(FakeTlsReader<R>, FakeTlsWriter<W>, String), R, W>
 where
     R: AsyncRead + Unpin,
@@ -1468,16 +1523,11 @@ where
 
     debug!(peer = %peer, response_len = response.len(), "Sending TLS ServerHello");
 
-    if let Err(e) = writer.write_all(&response).await {
+    if let Err(e) = write_tls_response(&mut writer, &response, response_write_options).await {
         warn!(peer = %peer, error = %e, "Failed to write TLS ServerHello");
         return HandshakeResult::Error(ProxyError::Io(e));
     }
 
-    if let Err(e) = writer.flush().await {
-        warn!(peer = %peer, error = %e, "Failed to flush TLS ServerHello");
-        return HandshakeResult::Error(ProxyError::Io(e));
-    }
-
     debug!(
         peer = %peer,
         user = %validated_user,
@@ -1498,6 +1548,26 @@ where
     ))
 }
 
+async fn write_tls_response<W: AsyncWrite + Unpin>(
+    writer: &mut W,
+    response: &[u8],
+    options: TlsResponseWriteOptions,
+) -> std::io::Result<()> {
+    #[cfg(target_os = "linux")]
+    if let (Some(fd), Some(fragment_size)) = (options.socket_fd, options.fragment_size) {
+        return crate::transport::socket::send_tcp_fragmented_fd(
+            fd,
+            response,
+            usize::from(fragment_size),
+        )
+        .await;
+    }
+
+    let _ = options;
+    writer.write_all(response).await?;
+    writer.flush().await
+}
+
 /// Handle MTProto obfuscation handshake
 #[cfg(test)]
 pub async fn handle_mtproto_handshake<R, W>(

src/proxy/tests/client_security_tests.rs

@@ -1940,6 +1940,98 @@ async fn fragmented_tls_mtproto_with_interleaved_ccs_is_accepted() {
     assert_eq!(success.dc_idx, 2);
 }
 
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn chunked_tls_response_preserves_complete_records_and_configured_mss() {
+    use crate::proxy::handshake::{
+        TlsResponseWriteOptions, handle_tls_handshake_with_shared_and_options,
+    };
+    use crate::proxy::shared_state::ProxySharedState;
+    use crate::transport::socket::{ListenOptions, create_listener};
+    use std::os::fd::{AsRawFd, BorrowedFd};
+
+    let options = ListenOptions {
+        reuse_port: false,
+        client_mss: Some(1400),
+        ..Default::default()
+    };
+    let socket = create_listener("127.0.0.1:0".parse().unwrap(), &options).unwrap();
+    let listener = TcpListener::from_std(socket.into()).unwrap();
+    let listener_addr = listener.local_addr().unwrap();
+    let mut client = TcpStream::connect(listener_addr).await.unwrap();
+    let (server, peer) = listener.accept().await.unwrap();
+
+    let secret_hex = "56565656565656565656565656565656";
+    let secret = [0x56u8; 16];
+    let client_hello = make_valid_tls_client_hello(&secret, 0);
+    let mut config = ProxyConfig::default();
+    config.general.beobachten = false;
+    config.access.ignore_time_skew = true;
+    config
+        .access
+        .users
+        .insert("user".to_string(), secret_hex.to_string());
+
+    let replay_checker = ReplayChecker::new(128, Duration::from_secs(60));
+    let rng = SecureRandom::new();
+    let shared = ProxySharedState::new();
+    let raw_fd = server.as_raw_fd();
+    let mss_before = socket2::SockRef::from(&server).tcp_mss().unwrap();
+    let (read_half, write_half) = server.into_split();
+
+    let (tls_reader, tls_writer, tls_user) = match handle_tls_handshake_with_shared_and_options(
+        &client_hello,
+        read_half,
+        write_half,
+        peer,
+        &config,
+        &replay_checker,
+        &rng,
+        None,
+        &shared,
+        TlsResponseWriteOptions::tcp(raw_fd, Some(92)),
+    )
+    .await
+    {
+        HandshakeResult::Success(result) => result,
+        _ => panic!("expected successful TLS handshake"),
+    };
+
+    // SAFETY: both split halves still own the accepted socket while it is borrowed.
+    let borrowed_fd = unsafe { BorrowedFd::borrow_raw(raw_fd) };
+    let mss_after = socket2::SockRef::from(&borrowed_fd).tcp_mss().unwrap();
+    assert_eq!(
+        mss_after, mss_before,
+        "chunked response writes must not change the configured socket MSS"
+    );
+    assert_eq!(tls_user, "user");
+
+    drop(tls_reader);
+    let mut writer = tls_writer.into_inner();
+    writer.shutdown().await.unwrap();
+
+    let mut response = Vec::new();
+    client.read_to_end(&mut response).await.unwrap();
+    assert!(!response.is_empty());
+
+    let mut offset = 0;
+    while offset < response.len() {
+        assert!(response.len() - offset >= 5, "truncated TLS record header");
+        let payload_len = u16::from_be_bytes([response[offset + 3], response[offset + 4]]) as usize;
+        offset += 5;
+        assert!(
+            response.len() - offset >= payload_len,
+            "truncated TLS record payload"
+        );
+        offset += payload_len;
+    }
+    assert_eq!(
+        offset,
+        response.len(),
+        "chunked response writes must preserve complete TLS records"
+    );
+}
+
 #[tokio::test]
 async fn valid_tls_path_does_not_fall_back_to_mask_backend() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

src/transport/socket.rs

@@ -125,36 +125,96 @@ pub fn clear_linger_fd(fd: std::os::unix::io::RawFd) -> Result<()> {
     Ok(())
 }
 
-/// Raise the TCP MSS on an already-accepted connection's fd. Used to fragment
-/// ONLY the TLS handshake (via a low listener MSS) and then restore a normal MSS
-/// for the bulk (post-handshake) data phase — cuts packets-per-second ~10x without losing the
-/// DPI evasion that the fragmented ServerHello provides. No-op safe: errors are
-/// returned to the caller, which logs and continues with the handshake MSS.
 #[cfg(target_os = "linux")]
-pub fn set_tcp_mss_fd(fd: std::os::unix::io::RawFd, mss: u32) -> Result<()> {
-    use std::io::Error;
-    let mss = i32::try_from(mss)
-        .map_err(|_| Error::new(std::io::ErrorKind::InvalidInput, "bulk MSS out of range"))?;
-    // Direct setsockopt(TCP_MAXSEG) — same pattern as the TCP_USER_TIMEOUT call
-    // above; avoids socket2 method-name drift across versions.
+fn force_tcp_push(fd: std::os::unix::io::RawFd) -> Result<()> {
+    let enabled: libc::c_int = 1;
     let rc = unsafe {
         libc::setsockopt(
             fd,
             libc::IPPROTO_TCP,
-            libc::TCP_MAXSEG,
-            &mss as *const libc::c_int as *const libc::c_void,
+            libc::TCP_NODELAY,
+            &enabled as *const libc::c_int as *const libc::c_void,
             std::mem::size_of::<libc::c_int>() as libc::socklen_t,
         )
     };
     if rc != 0 {
-        return Err(Error::last_os_error());
+        return Err(std::io::Error::last_os_error());
     }
     Ok(())
 }
 
-/// Non-Linux stub: MSS shaping only on Linux (TCP_MAXSEG).
-#[cfg(all(unix, not(target_os = "linux")))]
-pub fn set_tcp_mss_fd(_fd: std::os::unix::io::RawFd, _mss: u32) -> Result<()> {
+/// Sends an initial TCP response in collapse-resistant segments on Linux.
+///
+/// `fd` must refer to a connected, nonblocking TCP socket and remain valid for
+/// the duration of this call. The caller retains ownership of the original fd.
+/// Each successful `send(2)` is marked with `MSG_EOR`; Linux preserves that mark
+/// when deciding whether adjacent SKBs may be collapsed, including retransmits.
+/// Re-applying `TCP_NODELAY` forces each marked SKB out despite auto-corking.
+#[cfg(target_os = "linux")]
+pub(crate) async fn send_tcp_fragmented_fd(
+    fd: std::os::unix::io::RawFd,
+    data: &[u8],
+    fragment_size: usize,
+) -> Result<()> {
+    use std::io::{Error, ErrorKind};
+    use std::os::fd::{AsRawFd, BorrowedFd};
+    use tokio::io::Interest;
+    use tokio::io::unix::AsyncFd;
+
+    if fragment_size == 0 {
+        return Err(Error::new(
+            ErrorKind::InvalidInput,
+            "TCP fragment size must be greater than zero",
+        ));
+    }
+    if data.is_empty() {
+        return Ok(());
+    }
+    if fd < 0 {
+        return Err(Error::from_raw_os_error(libc::EBADF));
+    }
+
+    // SAFETY: the caller guarantees that fd remains open for this call.
+    let borrowed_fd = unsafe { BorrowedFd::borrow_raw(fd) };
+    let duplicated_fd = borrowed_fd.try_clone_to_owned()?;
+    let async_fd = AsyncFd::with_interest(duplicated_fd, Interest::WRITABLE)?;
+
+    for fragment in data.chunks(fragment_size) {
+        let mut offset = 0;
+        while offset < fragment.len() {
+            let mut writable = async_fd.writable().await?;
+            let sent = match writable.try_io(|inner| {
+                let remaining = &fragment[offset..];
+                let sent = unsafe {
+                    libc::send(
+                        inner.get_ref().as_raw_fd(),
+                        remaining.as_ptr().cast::<libc::c_void>(),
+                        remaining.len(),
+                        libc::MSG_DONTWAIT | libc::MSG_EOR | libc::MSG_NOSIGNAL,
+                    )
+                };
+                if sent < 0 {
+                    Err(Error::last_os_error())
+                } else if sent == 0 {
+                    Err(Error::new(
+                        ErrorKind::WriteZero,
+                        "fragmented TCP send returned zero",
+                    ))
+                } else {
+                    Ok(sent as usize)
+                }
+            }) {
+                Ok(Ok(sent)) => sent,
+                Ok(Err(error)) if error.kind() == ErrorKind::Interrupted => continue,
+                Ok(Err(error)) => return Err(error),
+                Err(_) => continue,
+            };
+
+            offset += sent;
+            force_tcp_push(async_fd.get_ref().as_raw_fd())?;
+        }
+    }
+
     Ok(())
 }
 
@@ -710,4 +770,72 @@ mod tests {
         };
         assert_eq!(mss, 256);
     }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn test_chunked_send_preserves_stream_and_configured_mss() {
+        use std::os::fd::AsRawFd;
+
+        let options = ListenOptions {
+            reuse_port: false,
+            client_mss: Some(1400),
+            ..Default::default()
+        };
+        let socket = match create_listener("127.0.0.1:0".parse().unwrap(), &options) {
+            Ok(socket) => socket,
+            Err(e) if e.kind() == ErrorKind::PermissionDenied => return,
+            Err(e) => panic!("create_listener failed: {e}"),
+        };
+        let listener = TcpListener::from_std(socket.into()).unwrap();
+        let addr = listener.local_addr().unwrap();
+        let client = TcpStream::connect(addr).await.unwrap();
+        let (mut server, _) = listener.accept().await.unwrap();
+
+        let initial_response: Vec<u8> = (0..4096).map(|value| (value % 251) as u8).collect();
+        let bulk_payload = vec![0xA5; 8192];
+        let expected_len = initial_response.len() + bulk_payload.len();
+        let reader = tokio::spawn(async move {
+            let mut client = client;
+            let mut received = vec![0; expected_len];
+            client.read_exact(&mut received).await.unwrap();
+            received
+        });
+
+        let mss_before = socket2::SockRef::from(&server).tcp_mss().unwrap();
+        send_tcp_fragmented_fd(server.as_raw_fd(), &initial_response, 92)
+            .await
+            .unwrap();
+        server.write_all(&bulk_payload).await.unwrap();
+        let mss_after = socket2::SockRef::from(&server).tcp_mss().unwrap();
+
+        let mut expected = initial_response;
+        expected.extend_from_slice(&bulk_payload);
+        assert_eq!(
+            reader.await.unwrap(),
+            expected,
+            "chunked send must preserve the byte stream"
+        );
+        assert_eq!(
+            mss_after, mss_before,
+            "chunked send must not change the configured socket MSS"
+        );
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn test_chunked_send_rejects_zero_fragment_size() {
+        let error = send_tcp_fragmented_fd(-1, b"response", 0)
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind(), ErrorKind::InvalidInput);
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn test_chunked_send_rejects_invalid_fd() {
+        let error = send_tcp_fragmented_fd(-1, b"response", 92)
+            .await
+            .unwrap_err();
+        assert_eq!(error.raw_os_error(), Some(libc::EBADF));
+    }
 }