Update Cargo.toml

Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>

Cargo.toml

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

docs/CONFIG_PARAMS.en.md

@@ -91,6 +91,7 @@ This document lists all configuration keys accepted by `config.toml`.
 | upstream_connect_retry_attempts | `u32` | `2` | Must be `> 0`. | Connect attempts for selected upstream before error/fallback. |
 | upstream_connect_retry_backoff_ms | `u64` | `100` | — | Delay between upstream connect attempts (ms). |
 | upstream_connect_budget_ms | `u64` | `3000` | Must be `> 0`. | Total wall-clock budget for one upstream connect request (ms). |
+| tg_connect | `u64` | `10` | Must be `> 0`. | Per-attempt upstream TCP connect timeout to Telegram DC (seconds). |
 | upstream_unhealthy_fail_threshold | `u32` | `5` | Must be `> 0`. | Consecutive failed requests before upstream is marked unhealthy. |
 | upstream_connect_failfast_hard_errors | `bool` | `false` | — | Skips additional retries for hard non-transient connect errors. |
 | stun_iface_mismatch_ignore | `bool` | `false` | none | Reserved compatibility flag in current runtime revision. |
@@ -249,7 +250,6 @@ Note: When `server.proxy_protocol` is enabled, incoming PROXY protocol headers a
 | relay_client_idle_soft_secs | `u64` | `120` | Must be `> 0`; must be `<= relay_client_idle_hard_secs`. | Soft idle threshold for middle-relay client uplink inactivity (seconds). |
 | relay_client_idle_hard_secs | `u64` | `360` | Must be `> 0`; must be `>= relay_client_idle_soft_secs`. | Hard idle threshold for middle-relay client uplink inactivity (seconds). |
 | relay_idle_grace_after_downstream_activity_secs | `u64` | `30` | Must be `<= relay_client_idle_hard_secs`. | Extra hard-idle grace after recent downstream activity (seconds). |
-| tg_connect | `u64` | `10` | — | Upstream Telegram connect timeout. |
 | client_keepalive | `u64` | `15` | — | Client keepalive timeout. |
 | client_ack | `u64` | `90` | — | Client ACK timeout. |
 | me_one_retry | `u8` | `12` | none | Fast reconnect attempts budget for single-endpoint DC scenarios. |

src/api/mod.rs

@@ -370,7 +370,10 @@ async fn handle(
                 let mut data: Vec<UserActiveIps> = active_ips_map
                     .into_iter()
                     .filter(|(_, ips)| !ips.is_empty())
-                    .map(|(username, active_ips)| UserActiveIps { username, active_ips })
+                    .map(|(username, active_ips)| UserActiveIps {
+                        username,
+                        active_ips,
+                    })
                     .collect();
                 data.sort_by(|a, b| a.username.cmp(&b.username));
                 Ok(success_response(StatusCode::OK, data, revision))

src/api/runtime_zero.rs

@@ -50,6 +50,7 @@ pub(super) struct RuntimeGatesData {
 
 #[derive(Serialize)]
 pub(super) struct EffectiveTimeoutLimits {
+    pub(super) client_first_byte_idle_secs: u64,
     pub(super) client_handshake_secs: u64,
     pub(super) tg_connect_secs: u64,
     pub(super) client_keepalive_secs: u64,
@@ -99,6 +100,11 @@ pub(super) struct EffectiveUserIpPolicyLimits {
     pub(super) window_secs: u64,
 }
 
+#[derive(Serialize)]
+pub(super) struct EffectiveUserTcpPolicyLimits {
+    pub(super) global_each: usize,
+}
+
 #[derive(Serialize)]
 pub(super) struct EffectiveLimitsData {
     pub(super) update_every_secs: u64,
@@ -108,6 +114,7 @@ pub(super) struct EffectiveLimitsData {
     pub(super) upstream: EffectiveUpstreamLimits,
     pub(super) middle_proxy: EffectiveMiddleProxyLimits,
     pub(super) user_ip_policy: EffectiveUserIpPolicyLimits,
+    pub(super) user_tcp_policy: EffectiveUserTcpPolicyLimits,
 }
 
 #[derive(Serialize)]
@@ -227,8 +234,9 @@ pub(super) fn build_limits_effective_data(cfg: &ProxyConfig) -> EffectiveLimitsD
         me_reinit_every_secs: cfg.general.effective_me_reinit_every_secs(),
         me_pool_force_close_secs: cfg.general.effective_me_pool_force_close_secs(),
         timeouts: EffectiveTimeoutLimits {
+            client_first_byte_idle_secs: cfg.timeouts.client_first_byte_idle_secs,
             client_handshake_secs: cfg.timeouts.client_handshake,
-            tg_connect_secs: cfg.timeouts.tg_connect,
+            tg_connect_secs: cfg.general.tg_connect,
             client_keepalive_secs: cfg.timeouts.client_keepalive,
             client_ack_secs: cfg.timeouts.client_ack,
             me_one_retry: cfg.timeouts.me_one_retry,
@@ -287,6 +295,9 @@ pub(super) fn build_limits_effective_data(cfg: &ProxyConfig) -> EffectiveLimitsD
             mode: user_max_unique_ips_mode_label(cfg.access.user_max_unique_ips_mode),
             window_secs: cfg.access.user_max_unique_ips_window_secs,
         },
+        user_tcp_policy: EffectiveUserTcpPolicyLimits {
+            global_each: cfg.access.user_max_tcp_conns_global_each,
+        },
     }
 }
 

src/api/users.rs

@@ -144,7 +144,14 @@ pub(super) async fn create_user(
         .unwrap_or(UserInfo {
             username: body.username.clone(),
             user_ad_tag: None,
-            max_tcp_conns: None,
+            max_tcp_conns: cfg
+                .access
+                .user_max_tcp_conns
+                .get(&body.username)
+                .copied()
+                .filter(|limit| *limit > 0)
+                .or((cfg.access.user_max_tcp_conns_global_each > 0)
+                    .then_some(cfg.access.user_max_tcp_conns_global_each)),
             expiration_rfc3339: None,
             data_quota_bytes: None,
             max_unique_ips: updated_limit,
@@ -395,7 +402,14 @@ pub(super) async fn users_from_config(
             });
         users.push(UserInfo {
             user_ad_tag: cfg.access.user_ad_tags.get(&username).cloned(),
-            max_tcp_conns: cfg.access.user_max_tcp_conns.get(&username).copied(),
+            max_tcp_conns: cfg
+                .access
+                .user_max_tcp_conns
+                .get(&username)
+                .copied()
+                .filter(|limit| *limit > 0)
+                .or((cfg.access.user_max_tcp_conns_global_each > 0)
+                    .then_some(cfg.access.user_max_tcp_conns_global_each)),
             expiration_rfc3339: cfg
                 .access
                 .user_expirations
@@ -572,3 +586,54 @@ fn resolve_tls_domains(cfg: &ProxyConfig) -> Vec<&str> {
     }
     domains
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ip_tracker::UserIpTracker;
+    use crate::stats::Stats;
+
+    #[tokio::test]
+    async fn users_from_config_reports_effective_tcp_limit_with_global_fallback() {
+        let mut cfg = ProxyConfig::default();
+        cfg.access.users.insert(
+            "alice".to_string(),
+            "0123456789abcdef0123456789abcdef".to_string(),
+        );
+        cfg.access.user_max_tcp_conns_global_each = 7;
+
+        let stats = Stats::new();
+        let tracker = UserIpTracker::new();
+
+        let users = users_from_config(&cfg, &stats, &tracker, None, None).await;
+        let alice = users
+            .iter()
+            .find(|entry| entry.username == "alice")
+            .expect("alice must be present");
+        assert_eq!(alice.max_tcp_conns, Some(7));
+
+        cfg.access.user_max_tcp_conns.insert("alice".to_string(), 5);
+        let users = users_from_config(&cfg, &stats, &tracker, None, None).await;
+        let alice = users
+            .iter()
+            .find(|entry| entry.username == "alice")
+            .expect("alice must be present");
+        assert_eq!(alice.max_tcp_conns, Some(5));
+
+        cfg.access.user_max_tcp_conns.insert("alice".to_string(), 0);
+        let users = users_from_config(&cfg, &stats, &tracker, None, None).await;
+        let alice = users
+            .iter()
+            .find(|entry| entry.username == "alice")
+            .expect("alice must be present");
+        assert_eq!(alice.max_tcp_conns, Some(7));
+
+        cfg.access.user_max_tcp_conns_global_each = 0;
+        let users = users_from_config(&cfg, &stats, &tracker, None, None).await;
+        let alice = users
+            .iter()
+            .find(|entry| entry.username == "alice")
+            .expect("alice must be present");
+        assert_eq!(alice.max_tcp_conns, None);
+    }
+}

src/cli.rs

@@ -13,7 +13,7 @@ use std::path::{Path, PathBuf};
 use std::process::Command;
 
 #[cfg(unix)]
-use crate::daemon::{self, DaemonOptions, DEFAULT_PID_FILE};
+use crate::daemon::{self, DEFAULT_PID_FILE, DaemonOptions};
 
 /// CLI subcommand to execute.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -437,13 +437,13 @@ pub fn run_init(opts: InitOptions) -> Result<(), Box<dyn std::error::Error>> {
     eprintln!("[+] Config written to {}", config_path.display());
 
     // 5. Generate and write service file
-    let exe_path = std::env::current_exe()
-        .unwrap_or_else(|_| PathBuf::from("/usr/local/bin/telemt"));
+    let exe_path =
+        std::env::current_exe().unwrap_or_else(|_| PathBuf::from("/usr/local/bin/telemt"));
 
     let service_opts = ServiceOptions {
         exe_path: &exe_path,
         config_path: &config_path,
-        user: None,  // Let systemd/init handle user
+        user: None, // Let systemd/init handle user
         group: None,
         pid_file: "/var/run/telemt.pid",
         working_dir: Some("/var/lib/telemt"),
@@ -584,6 +584,7 @@ me_pool_drain_soft_evict_cooldown_ms = 1000
 me_bind_stale_mode = "never"
 me_pool_min_fresh_ratio = 0.8
 me_reinit_drain_timeout_secs = 90
+tg_connect = 10
 
 [network]
 ipv4 = true
@@ -609,8 +610,8 @@ ip = "0.0.0.0"
 ip = "::"
 
 [timeouts]
-client_handshake = 15
-tg_connect = 10
+client_first_byte_idle_secs = 300
+client_handshake = 60
 client_keepalive = 60
 client_ack = 300
 
@@ -622,6 +623,7 @@ fake_cert_len = 2048
 tls_full_cert_ttl_secs = 90
 
 [access]
+user_max_tcp_conns_global_each = 0
 replay_check_len = 65536
 replay_window_secs = 120
 ignore_time_skew = false

src/config/defaults.rs

@@ -110,7 +110,11 @@ pub(crate) fn default_replay_window_secs() -> u64 {
 }
 
 pub(crate) fn default_handshake_timeout() -> u64 {
-    30
+    60
+}
+
+pub(crate) fn default_client_first_byte_idle_secs() -> u64 {
+    300
 }
 
 pub(crate) fn default_relay_idle_policy_v2_enabled() -> bool {
@@ -807,6 +811,10 @@ pub(crate) fn default_user_max_unique_ips_window_secs() -> u64 {
     DEFAULT_USER_MAX_UNIQUE_IPS_WINDOW_SECS
 }
 
+pub(crate) fn default_user_max_tcp_conns_global_each() -> usize {
+    0
+}
+
 pub(crate) fn default_user_max_unique_ips_global_each() -> usize {
     0
 }

src/config/hot_reload.rs

@@ -117,6 +117,7 @@ pub struct HotFields {
     pub users: std::collections::HashMap<String, String>,
     pub user_ad_tags: std::collections::HashMap<String, String>,
     pub user_max_tcp_conns: std::collections::HashMap<String, usize>,
+    pub user_max_tcp_conns_global_each: usize,
     pub user_expirations: std::collections::HashMap<String, chrono::DateTime<chrono::Utc>>,
     pub user_data_quota: std::collections::HashMap<String, u64>,
     pub user_max_unique_ips: std::collections::HashMap<String, usize>,
@@ -240,6 +241,7 @@ impl HotFields {
             users: cfg.access.users.clone(),
             user_ad_tags: cfg.access.user_ad_tags.clone(),
             user_max_tcp_conns: cfg.access.user_max_tcp_conns.clone(),
+            user_max_tcp_conns_global_each: cfg.access.user_max_tcp_conns_global_each,
             user_expirations: cfg.access.user_expirations.clone(),
             user_data_quota: cfg.access.user_data_quota.clone(),
             user_max_unique_ips: cfg.access.user_max_unique_ips.clone(),
@@ -530,6 +532,7 @@ fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig {
     cfg.access.users = new.access.users.clone();
     cfg.access.user_ad_tags = new.access.user_ad_tags.clone();
     cfg.access.user_max_tcp_conns = new.access.user_max_tcp_conns.clone();
+    cfg.access.user_max_tcp_conns_global_each = new.access.user_max_tcp_conns_global_each;
     cfg.access.user_expirations = new.access.user_expirations.clone();
     cfg.access.user_data_quota = new.access.user_data_quota.clone();
     cfg.access.user_max_unique_ips = new.access.user_max_unique_ips.clone();
@@ -696,6 +699,7 @@ fn warn_non_hot_changes(old: &ProxyConfig, new: &ProxyConfig, non_hot_changed: b
     if old.general.upstream_connect_retry_attempts != new.general.upstream_connect_retry_attempts
         || old.general.upstream_connect_retry_backoff_ms
             != new.general.upstream_connect_retry_backoff_ms
+        || old.general.tg_connect != new.general.tg_connect
         || old.general.upstream_unhealthy_fail_threshold
             != new.general.upstream_unhealthy_fail_threshold
         || old.general.upstream_connect_failfast_hard_errors
@@ -1144,6 +1148,12 @@ fn log_changes(
             new_hot.user_max_tcp_conns.len()
         );
     }
+    if old_hot.user_max_tcp_conns_global_each != new_hot.user_max_tcp_conns_global_each {
+        info!(
+            "config reload: user_max_tcp_conns policy global_each={}",
+            new_hot.user_max_tcp_conns_global_each
+        );
+    }
     if old_hot.user_expirations != new_hot.user_expirations {
         info!(
             "config reload: user_expirations updated ({} entries)",

src/config/load.rs

@@ -346,6 +346,12 @@ impl ProxyConfig {
             ));
         }
 
+        if config.general.tg_connect == 0 {
+            return Err(ProxyError::Config(
+                "general.tg_connect must be > 0".to_string(),
+            ));
+        }
+
         if config.general.upstream_unhealthy_fail_threshold == 0 {
             return Err(ProxyError::Config(
                 "general.upstream_unhealthy_fail_threshold must be > 0".to_string(),
@@ -1322,6 +1328,10 @@ mod tests {
             default_api_runtime_edge_events_capacity()
         );
         assert_eq!(cfg.access.users, default_access_users());
+        assert_eq!(
+            cfg.access.user_max_tcp_conns_global_each,
+            default_user_max_tcp_conns_global_each()
+        );
         assert_eq!(
             cfg.access.user_max_unique_ips_mode,
             UserMaxUniqueIpsMode::default()
@@ -1465,6 +1475,10 @@ mod tests {
 
         let access = AccessConfig::default();
         assert_eq!(access.users, default_access_users());
+        assert_eq!(
+            access.user_max_tcp_conns_global_each,
+            default_user_max_tcp_conns_global_each()
+        );
     }
 
     #[test]
@@ -1907,6 +1921,26 @@ mod tests {
         let _ = std::fs::remove_file(path);
     }
 
+    #[test]
+    fn tg_connect_zero_is_rejected() {
+        let toml = r#"
+            [general]
+            tg_connect = 0
+
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+        "#;
+        let dir = std::env::temp_dir();
+        let path = dir.join("telemt_tg_connect_zero_test.toml");
+        std::fs::write(&path, toml).unwrap();
+        let err = ProxyConfig::load(&path).unwrap_err().to_string();
+        assert!(err.contains("general.tg_connect must be > 0"));
+        let _ = std::fs::remove_file(path);
+    }
+
     #[test]
     fn rpc_proxy_req_every_out_of_range_is_rejected() {
         let toml = r#"

src/config/tests/load_idle_policy_tests.rs

@@ -17,6 +17,28 @@ fn remove_temp_config(path: &PathBuf) {
     let _ = fs::remove_file(path);
 }
 
+#[test]
+fn default_timeouts_enable_apple_compatible_handshake_profile() {
+    let cfg = ProxyConfig::default();
+    assert_eq!(cfg.timeouts.client_first_byte_idle_secs, 300);
+    assert_eq!(cfg.timeouts.client_handshake, 60);
+}
+
+#[test]
+fn load_accepts_zero_first_byte_idle_timeout_as_legacy_opt_out() {
+    let path = write_temp_config(
+        r#"
+[timeouts]
+client_first_byte_idle_secs = 0
+"#,
+    );
+
+    let cfg = ProxyConfig::load(&path).expect("config with zero first-byte idle timeout must load");
+    assert_eq!(cfg.timeouts.client_first_byte_idle_secs, 0);
+
+    remove_temp_config(&path);
+}
+
 #[test]
 fn load_rejects_relay_hard_idle_smaller_than_soft_idle_with_clear_error() {
     let path = write_temp_config(

src/config/types.rs

@@ -663,6 +663,10 @@ pub struct GeneralConfig {
     #[serde(default = "default_upstream_connect_budget_ms")]
     pub upstream_connect_budget_ms: u64,
 
+    /// Per-attempt TCP connect timeout to Telegram DC (seconds).
+    #[serde(default = "default_connect_timeout")]
+    pub tg_connect: u64,
+
     /// Consecutive failed requests before upstream is marked unhealthy.
     #[serde(default = "default_upstream_unhealthy_fail_threshold")]
     pub upstream_unhealthy_fail_threshold: u32,
@@ -1007,6 +1011,7 @@ impl Default for GeneralConfig {
             upstream_connect_retry_attempts: default_upstream_connect_retry_attempts(),
             upstream_connect_retry_backoff_ms: default_upstream_connect_retry_backoff_ms(),
             upstream_connect_budget_ms: default_upstream_connect_budget_ms(),
+            tg_connect: default_connect_timeout(),
             upstream_unhealthy_fail_threshold: default_upstream_unhealthy_fail_threshold(),
             upstream_connect_failfast_hard_errors: default_upstream_connect_failfast_hard_errors(),
             stun_iface_mismatch_ignore: false,
@@ -1314,6 +1319,12 @@ impl Default for ServerConfig {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TimeoutsConfig {
+    /// Maximum idle wait in seconds for the first client byte before handshake parsing starts.
+    /// `0` disables the separate idle phase and keeps legacy timeout behavior.
+    #[serde(default = "default_client_first_byte_idle_secs")]
+    pub client_first_byte_idle_secs: u64,
+
+    /// Maximum active handshake duration in seconds after the first client byte is received.
     #[serde(default = "default_handshake_timeout")]
     pub client_handshake: u64,
 
@@ -1335,9 +1346,6 @@ pub struct TimeoutsConfig {
     #[serde(default = "default_relay_idle_grace_after_downstream_activity_secs")]
     pub relay_idle_grace_after_downstream_activity_secs: u64,
 
-    #[serde(default = "default_connect_timeout")]
-    pub tg_connect: u64,
-
     #[serde(default = "default_keepalive")]
     pub client_keepalive: u64,
 
@@ -1356,13 +1364,13 @@ pub struct TimeoutsConfig {
 impl Default for TimeoutsConfig {
     fn default() -> Self {
         Self {
+            client_first_byte_idle_secs: default_client_first_byte_idle_secs(),
             client_handshake: default_handshake_timeout(),
             relay_idle_policy_v2_enabled: default_relay_idle_policy_v2_enabled(),
             relay_client_idle_soft_secs: default_relay_client_idle_soft_secs(),
             relay_client_idle_hard_secs: default_relay_client_idle_hard_secs(),
             relay_idle_grace_after_downstream_activity_secs:
                 default_relay_idle_grace_after_downstream_activity_secs(),
-            tg_connect: default_connect_timeout(),
             client_keepalive: default_keepalive(),
             client_ack: default_ack_timeout(),
             me_one_retry: default_me_one_retry(),
@@ -1625,6 +1633,12 @@ pub struct AccessConfig {
     #[serde(default)]
     pub user_max_tcp_conns: HashMap<String, usize>,
 
+    /// Global per-user TCP connection limit applied when a user has no
+    /// positive individual override.
+    /// `0` disables the inherited limit.
+    #[serde(default = "default_user_max_tcp_conns_global_each")]
+    pub user_max_tcp_conns_global_each: usize,
+
     #[serde(default)]
     pub user_expirations: HashMap<String, DateTime<Utc>>,
 
@@ -1661,6 +1675,7 @@ impl Default for AccessConfig {
             users: default_access_users(),
             user_ad_tags: HashMap::new(),
             user_max_tcp_conns: HashMap::new(),
+            user_max_tcp_conns_global_each: default_user_max_tcp_conns_global_each(),
             user_expirations: HashMap::new(),
             user_data_quota: HashMap::new(),
             user_max_unique_ips: HashMap::new(),

src/daemon/mod.rs

@@ -206,7 +206,9 @@ impl PidFile {
         let mut contents = String::new();
         File::open(&self.path)
             .and_then(|mut f| f.read_to_string(&mut contents))
-            .map_err(|e| DaemonError::PidFile(format!("cannot read {}: {}", self.path.display(), e)))?;
+            .map_err(|e| {
+                DaemonError::PidFile(format!("cannot read {}: {}", self.path.display(), e))
+            })?;
 
         let pid: i32 = contents
             .trim()
@@ -269,12 +271,16 @@ impl PidFile {
 
         // Write our PID
         let pid = getpid();
-        let mut file = flock.unlock().map_err(|(_, errno)| {
-            DaemonError::PidFile(format!("unlock failed: {}", errno))
-        })?;
+        let mut file = flock
+            .unlock()
+            .map_err(|(_, errno)| DaemonError::PidFile(format!("unlock failed: {}", errno)))?;
 
         writeln!(file, "{}", pid).map_err(|e| {
-            DaemonError::PidFile(format!("cannot write PID to {}: {}", self.path.display(), e))
+            DaemonError::PidFile(format!(
+                "cannot write PID to {}: {}",
+                self.path.display(),
+                e
+            ))
         })?;
 
         // Re-acquire lock and keep it
@@ -373,7 +379,8 @@ pub fn drop_privileges(user: Option<&str>, group: Option<&str>) -> Result<(), Da
 /// Looks up a user by name and returns their UID.
 fn lookup_user(name: &str) -> Result<Uid, DaemonError> {
     // Use libc getpwnam
-    let c_name = std::ffi::CString::new(name).map_err(|_| DaemonError::UserNotFound(name.to_string()))?;
+    let c_name =
+        std::ffi::CString::new(name).map_err(|_| DaemonError::UserNotFound(name.to_string()))?;
 
     unsafe {
         let pwd = libc::getpwnam(c_name.as_ptr());
@@ -387,7 +394,8 @@ fn lookup_user(name: &str) -> Result<Uid, DaemonError> {
 
 /// Looks up a user's primary GID by username.
 fn lookup_user_primary_gid(name: &str) -> Result<Gid, DaemonError> {
-    let c_name = std::ffi::CString::new(name).map_err(|_| DaemonError::UserNotFound(name.to_string()))?;
+    let c_name =
+        std::ffi::CString::new(name).map_err(|_| DaemonError::UserNotFound(name.to_string()))?;
 
     unsafe {
         let pwd = libc::getpwnam(c_name.as_ptr());
@@ -401,7 +409,8 @@ fn lookup_user_primary_gid(name: &str) -> Result<Gid, DaemonError> {
 
 /// Looks up a group by name and returns its GID.
 fn lookup_group(name: &str) -> Result<Gid, DaemonError> {
-    let c_name = std::ffi::CString::new(name).map_err(|_| DaemonError::GroupNotFound(name.to_string()))?;
+    let c_name =
+        std::ffi::CString::new(name).map_err(|_| DaemonError::GroupNotFound(name.to_string()))?;
 
     unsafe {
         let grp = libc::getgrnam(c_name.as_ptr());
@@ -444,9 +453,8 @@ pub fn signal_pid_file<P: AsRef<Path>>(
         )));
     }
 
-    nix::sys::signal::kill(Pid::from_raw(pid), signal).map_err(|e| {
-        DaemonError::PidFile(format!("cannot signal process {}: {}", pid, e))
-    })?;
+    nix::sys::signal::kill(Pid::from_raw(pid), signal)
+        .map_err(|e| DaemonError::PidFile(format!("cannot signal process {}: {}", pid, e)))?;
 
     Ok(())
 }

src/logging.rs

@@ -63,7 +63,10 @@ impl LoggingGuard {
 pub fn init_logging(
     opts: &LoggingOptions,
     initial_filter: &str,
-) -> (reload::Handle<EnvFilter, impl tracing::Subscriber + Send + Sync>, LoggingGuard) {
+) -> (
+    reload::Handle<EnvFilter, impl tracing::Subscriber + Send + Sync>,
+    LoggingGuard,
+) {
     let (filter_layer, filter_handle) = reload::Layer::new(EnvFilter::new(initial_filter));
 
     match &opts.destination {
@@ -101,7 +104,8 @@ pub fn init_logging(
                 // Extract directory and filename prefix
                 let path = Path::new(path);
                 let dir = path.parent().unwrap_or(Path::new("/var/log"));
-                let prefix = path.file_name()
+                let prefix = path
+                    .file_name()
                     .and_then(|s| s.to_str())
                     .unwrap_or("telemt");
 
@@ -182,7 +186,11 @@ impl std::io::Write for SyslogWriter {
             .unwrap_or_else(|_| std::ffi::CString::new("(invalid utf8)").unwrap());
 
         unsafe {
-            libc::syslog(priority, b"%s\0".as_ptr() as *const libc::c_char, c_msg.as_ptr());
+            libc::syslog(
+                priority,
+                b"%s\0".as_ptr() as *const libc::c_char,
+                c_msg.as_ptr(),
+            );
         }
 
         Ok(buf.len())
@@ -255,7 +263,10 @@ mod tests {
     #[test]
     fn test_parse_log_destination_default() {
         let args: Vec<String> = vec![];
-        assert!(matches!(parse_log_destination(&args), LogDestination::Stderr));
+        assert!(matches!(
+            parse_log_destination(&args),
+            LogDestination::Stderr
+        ));
     }
 
     #[test]
@@ -286,6 +297,9 @@ mod tests {
     #[test]
     fn test_parse_log_destination_syslog() {
         let args = vec!["--syslog".to_string()];
-        assert!(matches!(parse_log_destination(&args), LogDestination::Syslog));
+        assert!(matches!(
+            parse_log_destination(&args),
+            LogDestination::Syslog
+        ));
     }
 }

src/maestro/helpers.rs

@@ -149,7 +149,9 @@ fn print_help() {
     }
     eprintln!();
     eprintln!("Options:");
-    eprintln!("  --data-path <DIR>       Set data directory (absolute path; overrides config value)");
+    eprintln!(
+        "  --data-path <DIR>       Set data directory (absolute path; overrides config value)"
+    );
     eprintln!("  --silent, -s            Suppress info logs");
     eprintln!("  --log-level <LEVEL>     debug|verbose|normal|silent");
     eprintln!("  --help, -h              Show this help");
@@ -173,16 +175,10 @@ fn print_help() {
         eprintln!();
     }
     eprintln!("Setup (fire-and-forget):");
-    eprintln!(
-        "  --init                  Generate config, install systemd service, start"
-    );
+    eprintln!("  --init                  Generate config, install systemd service, start");
     eprintln!("    --port <PORT>          Listen port (default: 443)");
-    eprintln!(
-        "    --domain <DOMAIN>      TLS domain for masking (default: www.google.com)"
-    );
-    eprintln!(
-        "    --secret <HEX>         32-char hex secret (auto-generated if omitted)"
-    );
+    eprintln!("    --domain <DOMAIN>      TLS domain for masking (default: www.google.com)");
+    eprintln!("    --secret <HEX>         32-char hex secret (auto-generated if omitted)");
     eprintln!("    --user <NAME>          Username (default: user)");
     eprintln!("    --config-dir <DIR>     Config directory (default: /etc/telemt)");
     eprintln!("    --no-start             Don't start the service after install");

src/maestro/mod.rs

@@ -83,7 +83,6 @@ pub async fn run() -> std::result::Result<(), Box<dyn std::error::Error>> {
 async fn run_inner(
     daemon_opts: DaemonOptions,
 ) -> std::result::Result<(), Box<dyn std::error::Error>> {
-
     // Acquire PID file if daemonizing or if explicitly requested
     // Keep it alive until shutdown (underscore prefix = intentionally kept for RAII cleanup)
     let _pid_file = if daemon_opts.daemonize || daemon_opts.pid_file.is_some() {
@@ -302,6 +301,7 @@ async fn run_inner(
         config.general.upstream_connect_retry_attempts,
         config.general.upstream_connect_retry_backoff_ms,
         config.general.upstream_connect_budget_ms,
+        config.general.tg_connect,
         config.general.upstream_unhealthy_fail_threshold,
         config.general.upstream_connect_failfast_hard_errors,
         stats.clone(),
@@ -664,10 +664,7 @@ async fn run_inner(
 
     // Drop privileges after binding sockets (which may require root for port < 1024)
     if daemon_opts.user.is_some() || daemon_opts.group.is_some() {
-        if let Err(e) = drop_privileges(
-            daemon_opts.user.as_deref(),
-            daemon_opts.group.as_deref(),
-        ) {
+        if let Err(e) = drop_privileges(daemon_opts.user.as_deref(), daemon_opts.group.as_deref()) {
             error!(error = %e, "Failed to drop privileges");
             std::process::exit(1);
         }

src/maestro/shutdown.rs

@@ -11,10 +11,10 @@
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-#[cfg(unix)]
-use tokio::signal::unix::{SignalKind, signal};
 #[cfg(not(unix))]
 use tokio::signal;
+#[cfg(unix)]
+use tokio::signal::unix::{SignalKind, signal};
 use tracing::{info, warn};
 
 use crate::stats::Stats;
@@ -94,7 +94,8 @@ async fn perform_shutdown(
 
     // Graceful ME pool shutdown
     if let Some(pool) = &me_pool {
-        match tokio::time::timeout(Duration::from_secs(2), pool.shutdown_send_close_conn_all()).await
+        match tokio::time::timeout(Duration::from_secs(2), pool.shutdown_send_close_conn_all())
+            .await
         {
             Ok(total) => {
                 info!(
@@ -159,15 +160,12 @@ fn dump_stats(stats: &Stats, process_started_at: Instant) {
 /// - SIGUSR1: Log rotation acknowledgment (for external log rotation tools)
 /// - SIGUSR2: Dump runtime status to log
 #[cfg(unix)]
-pub(crate) fn spawn_signal_handlers(
-    stats: Arc<Stats>,
-    process_started_at: Instant,
-) {
+pub(crate) fn spawn_signal_handlers(stats: Arc<Stats>, process_started_at: Instant) {
     tokio::spawn(async move {
-        let mut sigusr1 = signal(SignalKind::user_defined1())
-            .expect("Failed to register SIGUSR1 handler");
-        let mut sigusr2 = signal(SignalKind::user_defined2())
-            .expect("Failed to register SIGUSR2 handler");
+        let mut sigusr1 =
+            signal(SignalKind::user_defined1()).expect("Failed to register SIGUSR1 handler");
+        let mut sigusr2 =
+            signal(SignalKind::user_defined2()).expect("Failed to register SIGUSR2 handler");
 
         loop {
             tokio::select! {
@@ -184,10 +182,7 @@ pub(crate) fn spawn_signal_handlers(
 
 /// No-op on non-Unix platforms.
 #[cfg(not(unix))]
-pub(crate) fn spawn_signal_handlers(
-    _stats: Arc<Stats>,
-    _process_started_at: Instant,
-) {
+pub(crate) fn spawn_signal_handlers(_stats: Arc<Stats>, _process_started_at: Instant) {
     // No SIGUSR1/SIGUSR2 on non-Unix
 }
 

src/main.rs

@@ -8,8 +8,6 @@ mod crypto;
 mod daemon;
 mod error;
 mod ip_tracker;
-mod logging;
-mod service;
 #[cfg(test)]
 #[path = "tests/ip_tracker_encapsulation_adversarial_tests.rs"]
 mod ip_tracker_encapsulation_adversarial_tests;
@@ -19,11 +17,13 @@ mod ip_tracker_hotpath_adversarial_tests;
 #[cfg(test)]
 #[path = "tests/ip_tracker_regression_tests.rs"]
 mod ip_tracker_regression_tests;
+mod logging;
 mod maestro;
 mod metrics;
 mod network;
 mod protocol;
 mod proxy;
+mod service;
 mod startup;
 mod stats;
 mod stream;

src/proxy/client.rs

@@ -416,16 +416,68 @@ where
 
     debug!(peer = %real_peer, "New connection (generic stream)");
 
+    let first_byte = if config.timeouts.client_first_byte_idle_secs == 0 {
+        None
+    } else {
+        let idle_timeout = Duration::from_secs(config.timeouts.client_first_byte_idle_secs);
+        let mut first_byte = [0u8; 1];
+        match timeout(idle_timeout, stream.read(&mut first_byte)).await {
+            Ok(Ok(0)) => {
+                debug!(peer = %real_peer, "Connection closed before first client byte");
+                return Ok(());
+            }
+            Ok(Ok(_)) => Some(first_byte[0]),
+            Ok(Err(e))
+                if matches!(
+                    e.kind(),
+                    std::io::ErrorKind::UnexpectedEof
+                        | std::io::ErrorKind::ConnectionReset
+                        | std::io::ErrorKind::ConnectionAborted
+                        | std::io::ErrorKind::BrokenPipe
+                        | std::io::ErrorKind::NotConnected
+                ) =>
+            {
+                debug!(
+                    peer = %real_peer,
+                    error = %e,
+                    "Connection closed before first client byte"
+                );
+                return Ok(());
+            }
+            Ok(Err(e)) => {
+                debug!(
+                    peer = %real_peer,
+                    error = %e,
+                    "Failed while waiting for first client byte"
+                );
+                return Err(ProxyError::Io(e));
+            }
+            Err(_) => {
+                debug!(
+                    peer = %real_peer,
+                    idle_secs = config.timeouts.client_first_byte_idle_secs,
+                    "Closing idle pooled connection before first client byte"
+                );
+                return Ok(());
+            }
+        }
+    };
+
     let handshake_timeout = handshake_timeout_with_mask_grace(&config);
     let stats_for_timeout = stats.clone();
     let config_for_timeout = config.clone();
     let beobachten_for_timeout = beobachten.clone();
     let peer_for_timeout = real_peer.ip();
 
-    // Phase 1: handshake (with timeout)
+    // Phase 2: active handshake (with timeout after the first client byte)
     let outcome = match timeout(handshake_timeout, async {
         let mut first_bytes = [0u8; 5];
-        stream.read_exact(&mut first_bytes).await?;
+        if let Some(first_byte) = first_byte {
+            first_bytes[0] = first_byte;
+            stream.read_exact(&mut first_bytes[1..]).await?;
+        } else {
+            stream.read_exact(&mut first_bytes).await?;
+        }
 
         let is_tls = tls::is_tls_handshake(&first_bytes[..3]);
         debug!(peer = %real_peer, is_tls = is_tls, "Handshake type detected");
@@ -736,36 +788,9 @@ impl RunningClientHandler {
             debug!(peer = %peer, error = %e, "Failed to configure client socket");
         }
 
-        let handshake_timeout = handshake_timeout_with_mask_grace(&self.config);
-        let stats = self.stats.clone();
-        let config_for_timeout = self.config.clone();
-        let beobachten_for_timeout = self.beobachten.clone();
-        let peer_for_timeout = peer.ip();
-
-        // Phase 1: handshake (with timeout)
-        let outcome = match timeout(handshake_timeout, self.do_handshake()).await {
-            Ok(Ok(outcome)) => outcome,
-            Ok(Err(e)) => {
-                debug!(peer = %peer, error = %e, "Handshake failed");
-                record_handshake_failure_class(
-                    &beobachten_for_timeout,
-                    &config_for_timeout,
-                    peer_for_timeout,
-                    &e,
-                );
-                return Err(e);
-            }
-            Err(_) => {
-                stats.increment_handshake_timeouts();
-                debug!(peer = %peer, "Handshake timeout");
-                record_beobachten_class(
-                    &beobachten_for_timeout,
-                    &config_for_timeout,
-                    peer_for_timeout,
-                    "other",
-                );
-                return Err(ProxyError::TgHandshakeTimeout);
-            }
+        let outcome = match self.do_handshake().await? {
+            Some(outcome) => outcome,
+            None => return Ok(()),
         };
 
         // Phase 2: relay (WITHOUT handshake timeout — relay has its own activity timeouts)
@@ -774,7 +799,7 @@ impl RunningClientHandler {
         }
     }
 
-    async fn do_handshake(mut self) -> Result<HandshakeOutcome> {
+    async fn do_handshake(mut self) -> Result<Option<HandshakeOutcome>> {
         let mut local_addr = self.stream.local_addr().map_err(ProxyError::Io)?;
 
         if self.proxy_protocol_enabled {
@@ -849,19 +874,108 @@ impl RunningClientHandler {
             }
         }
 
-        let mut first_bytes = [0u8; 5];
-        self.stream.read_exact(&mut first_bytes).await?;
-
-        let is_tls = tls::is_tls_handshake(&first_bytes[..3]);
-        let peer = self.peer;
-
-        debug!(peer = %peer, is_tls = is_tls, "Handshake type detected");
-
-        if is_tls {
-            self.handle_tls_client(first_bytes, local_addr).await
+        let first_byte = if self.config.timeouts.client_first_byte_idle_secs == 0 {
+            None
         } else {
-            self.handle_direct_client(first_bytes, local_addr).await
-        }
+            let idle_timeout =
+                Duration::from_secs(self.config.timeouts.client_first_byte_idle_secs);
+            let mut first_byte = [0u8; 1];
+            match timeout(idle_timeout, self.stream.read(&mut first_byte)).await {
+                Ok(Ok(0)) => {
+                    debug!(peer = %self.peer, "Connection closed before first client byte");
+                    return Ok(None);
+                }
+                Ok(Ok(_)) => Some(first_byte[0]),
+                Ok(Err(e))
+                    if matches!(
+                        e.kind(),
+                        std::io::ErrorKind::UnexpectedEof
+                            | std::io::ErrorKind::ConnectionReset
+                            | std::io::ErrorKind::ConnectionAborted
+                            | std::io::ErrorKind::BrokenPipe
+                            | std::io::ErrorKind::NotConnected
+                    ) =>
+                {
+                    debug!(
+                        peer = %self.peer,
+                        error = %e,
+                        "Connection closed before first client byte"
+                    );
+                    return Ok(None);
+                }
+                Ok(Err(e)) => {
+                    debug!(
+                        peer = %self.peer,
+                        error = %e,
+                        "Failed while waiting for first client byte"
+                    );
+                    return Err(ProxyError::Io(e));
+                }
+                Err(_) => {
+                    debug!(
+                        peer = %self.peer,
+                        idle_secs = self.config.timeouts.client_first_byte_idle_secs,
+                        "Closing idle pooled connection before first client byte"
+                    );
+                    return Ok(None);
+                }
+            }
+        };
+
+        let handshake_timeout = handshake_timeout_with_mask_grace(&self.config);
+        let stats = self.stats.clone();
+        let config_for_timeout = self.config.clone();
+        let beobachten_for_timeout = self.beobachten.clone();
+        let peer_for_timeout = self.peer.ip();
+        let peer_for_log = self.peer;
+
+        let outcome = match timeout(handshake_timeout, async {
+            let mut first_bytes = [0u8; 5];
+            if let Some(first_byte) = first_byte {
+                first_bytes[0] = first_byte;
+                self.stream.read_exact(&mut first_bytes[1..]).await?;
+            } else {
+                self.stream.read_exact(&mut first_bytes).await?;
+            }
+
+            let is_tls = tls::is_tls_handshake(&first_bytes[..3]);
+            let peer = self.peer;
+
+            debug!(peer = %peer, is_tls = is_tls, "Handshake type detected");
+
+            if is_tls {
+                self.handle_tls_client(first_bytes, local_addr).await
+            } else {
+                self.handle_direct_client(first_bytes, local_addr).await
+            }
+        })
+        .await
+        {
+            Ok(Ok(outcome)) => outcome,
+            Ok(Err(e)) => {
+                debug!(peer = %peer_for_log, error = %e, "Handshake failed");
+                record_handshake_failure_class(
+                    &beobachten_for_timeout,
+                    &config_for_timeout,
+                    peer_for_timeout,
+                    &e,
+                );
+                return Err(e);
+            }
+            Err(_) => {
+                stats.increment_handshake_timeouts();
+                debug!(peer = %peer_for_log, "Handshake timeout");
+                record_beobachten_class(
+                    &beobachten_for_timeout,
+                    &config_for_timeout,
+                    peer_for_timeout,
+                    "other",
+                );
+                return Err(ProxyError::TgHandshakeTimeout);
+            }
+        };
+
+        Ok(Some(outcome))
     }
 
     async fn handle_tls_client(
@@ -1252,7 +1366,11 @@ impl RunningClientHandler {
             .access
             .user_max_tcp_conns
             .get(user)
-            .map(|v| *v as u64);
+            .copied()
+            .filter(|limit| *limit > 0)
+            .or((config.access.user_max_tcp_conns_global_each > 0)
+                .then_some(config.access.user_max_tcp_conns_global_each))
+            .map(|v| v as u64);
         if !stats.try_acquire_user_curr_connects(user, limit) {
             return Err(ProxyError::ConnectionLimitExceeded {
                 user: user.to_string(),
@@ -1311,7 +1429,11 @@ impl RunningClientHandler {
             .access
             .user_max_tcp_conns
             .get(user)
-            .map(|v| *v as u64);
+            .copied()
+            .filter(|limit| *limit > 0)
+            .or((config.access.user_max_tcp_conns_global_each > 0)
+                .then_some(config.access.user_max_tcp_conns_global_each))
+            .map(|v| v as u64);
         if !stats.try_acquire_user_curr_connects(user, limit) {
             return Err(ProxyError::ConnectionLimitExceeded {
                 user: user.to_string(),

src/proxy/tests/client_clever_advanced_tests.rs

@@ -94,6 +94,7 @@ async fn adversarial_tls_handshake_timeout_during_masking_delay() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -141,6 +142,7 @@ async fn blackhat_proxy_protocol_slowloris_timeout() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -193,6 +195,7 @@ async fn negative_proxy_protocol_enabled_but_client_sends_tls_hello() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -239,6 +242,7 @@ async fn edge_client_stream_exactly_4_bytes_eof() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -282,6 +286,7 @@ async fn edge_client_stream_tls_header_valid_but_body_1_byte_short_eof() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -328,6 +333,7 @@ async fn integration_non_tls_modes_disabled_immediately_masks() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),

src/proxy/tests/client_deep_invariants_tests.rs

@@ -47,6 +47,7 @@ async fn invariant_tls_clienthello_truncation_exact_boundary_triggers_masking()
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -177,6 +178,7 @@ async fn invariant_direct_mode_partial_header_eof_is_error_not_bad_connect() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),

src/proxy/tests/client_masking_blackhat_campaign_tests.rs

@@ -40,6 +40,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_budget_security_tests.rs

@@ -36,6 +36,7 @@ fn build_harness(config: ProxyConfig) -> PipelineHarness {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),

src/proxy/tests/client_masking_diagnostics_security_tests.rs

@@ -20,6 +20,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_fragmented_classifier_security_tests.rs

@@ -20,6 +20,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_hard_adversarial_tests.rs

@@ -34,6 +34,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_http2_fragmented_preface_security_tests.rs

@@ -20,6 +20,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_prefetch_config_pipeline_integration_security_tests.rs

@@ -20,6 +20,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_prefetch_invariant_security_tests.rs

@@ -47,6 +47,7 @@ fn build_harness(secret_hex: &str, mask_port: u16) -> PipelineHarness {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),

src/proxy/tests/client_masking_probe_evasion_blackhat_tests.rs

@@ -25,6 +25,7 @@ fn make_test_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_redteam_expected_fail_tests.rs

@@ -48,6 +48,7 @@ fn build_harness(secret_hex: &str, mask_port: u16) -> RedTeamHarness {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -237,6 +238,7 @@ async fn redteam_03_masking_duration_must_be_less_than_1ms_when_backend_down() {
             1,
             1,
             1,
+            10,
             1,
             false,
             Arc::new(Stats::new()),
@@ -477,6 +479,7 @@ async fn measure_invalid_probe_duration_ms(delay_ms: u64, tls_len: u16, body_sen
             1,
             1,
             1,
+            10,
             1,
             false,
             Arc::new(Stats::new()),
@@ -550,6 +553,7 @@ async fn capture_forwarded_probe_len(tls_len: u16, body_sent: usize) -> usize {
             1,
             1,
             1,
+            10,
             1,
             false,
             Arc::new(Stats::new()),

src/proxy/tests/client_masking_replay_timing_security_tests.rs

@@ -22,6 +22,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_shape_classifier_fuzz_redteam_expected_fail_tests.rs

@@ -20,6 +20,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_shape_hardening_adversarial_tests.rs

@@ -20,6 +20,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_shape_hardening_redteam_expected_fail_tests.rs

@@ -20,6 +20,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_shape_hardening_security_tests.rs

@@ -20,6 +20,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_masking_stress_adversarial_tests.rs

@@ -34,6 +34,7 @@ fn new_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_more_advanced_tests.rs

@@ -100,6 +100,7 @@ async fn blackhat_proxy_protocol_massive_garbage_rejected_quickly() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -146,6 +147,7 @@ async fn edge_tls_body_immediate_eof_triggers_masking_and_bad_connect() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -195,6 +197,7 @@ async fn security_classic_mode_disabled_masks_valid_length_payload() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),

src/proxy/tests/client_security_tests.rs

@@ -1,8 +1,10 @@
 use super::*;
 use crate::config::{UpstreamConfig, UpstreamType};
-use crate::crypto::AesCtr;
-use crate::crypto::sha256_hmac;
-use crate::protocol::constants::ProtoTag;
+use crate::crypto::{AesCtr, sha256, sha256_hmac};
+use crate::protocol::constants::{
+    DC_IDX_POS, HANDSHAKE_LEN, IV_LEN, PREKEY_LEN, PROTO_TAG_POS, ProtoTag, SKIP_LEN,
+    TLS_RECORD_CHANGE_CIPHER,
+};
 use crate::protocol::tls;
 use crate::proxy::handshake::HandshakeSuccess;
 use crate::stream::{CryptoReader, CryptoWriter};
@@ -339,6 +341,7 @@ async fn relay_task_abort_releases_user_gate_and_ip_reservation() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -452,6 +455,7 @@ async fn relay_cutover_releases_user_gate_and_ip_reservation() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -575,6 +579,7 @@ async fn integration_route_cutover_and_quota_overlap_fails_closed_and_releases_s
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -744,6 +749,7 @@ async fn proxy_protocol_header_is_rejected_when_trust_list_is_empty() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -820,6 +826,7 @@ async fn proxy_protocol_header_from_untrusted_peer_range_is_rejected_under_load(
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -979,6 +986,7 @@ async fn short_tls_probe_is_masked_through_client_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1066,6 +1074,7 @@ async fn tls12_record_probe_is_masked_through_client_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1151,6 +1160,7 @@ async fn handle_client_stream_increments_connects_all_exactly_once() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1243,6 +1253,7 @@ async fn running_client_handler_increments_connects_all_exactly_once() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1310,6 +1321,163 @@ async fn running_client_handler_increments_connects_all_exactly_once() {
     );
 }
 
+#[tokio::test(start_paused = true)]
+async fn idle_pooled_connection_closes_cleanly_in_generic_stream_path() {
+    let mut cfg = ProxyConfig::default();
+    cfg.general.beobachten = false;
+    cfg.timeouts.client_first_byte_idle_secs = 1;
+
+    let config = Arc::new(cfg);
+    let stats = Arc::new(Stats::new());
+    let upstream_manager = Arc::new(UpstreamManager::new(
+        vec![UpstreamConfig {
+            upstream_type: UpstreamType::Direct {
+                interface: None,
+                bind_addresses: None,
+            },
+            weight: 1,
+            enabled: true,
+            scopes: String::new(),
+            selected_scope: String::new(),
+        }],
+        1,
+        1,
+        1,
+        10,
+        1,
+        false,
+        stats.clone(),
+    ));
+    let replay_checker = Arc::new(ReplayChecker::new(128, Duration::from_secs(60)));
+    let buffer_pool = Arc::new(BufferPool::new());
+    let rng = Arc::new(SecureRandom::new());
+    let route_runtime = Arc::new(RouteRuntimeController::new(RelayRouteMode::Direct));
+    let ip_tracker = Arc::new(UserIpTracker::new());
+    let beobachten = Arc::new(BeobachtenStore::new());
+
+    let (server_side, _client_side) = duplex(4096);
+    let peer: SocketAddr = "198.51.100.169:55200".parse().unwrap();
+
+    let handler = tokio::spawn(handle_client_stream(
+        server_side,
+        peer,
+        config,
+        stats.clone(),
+        upstream_manager,
+        replay_checker,
+        buffer_pool,
+        rng,
+        None,
+        route_runtime,
+        None,
+        ip_tracker,
+        beobachten,
+        false,
+    ));
+
+    // Let the spawned handler arm the idle-phase timeout before advancing paused time.
+    tokio::task::yield_now().await;
+    tokio::time::advance(Duration::from_secs(2)).await;
+    tokio::task::yield_now().await;
+
+    let result = tokio::time::timeout(Duration::from_secs(1), handler)
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(result.is_ok());
+    assert_eq!(stats.get_handshake_timeouts(), 0);
+    assert_eq!(stats.get_connects_bad(), 0);
+}
+
+#[tokio::test(start_paused = true)]
+async fn idle_pooled_connection_closes_cleanly_in_client_handler_path() {
+    let front_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let front_addr = front_listener.local_addr().unwrap();
+
+    let mut cfg = ProxyConfig::default();
+    cfg.general.beobachten = false;
+    cfg.timeouts.client_first_byte_idle_secs = 1;
+
+    let config = Arc::new(cfg);
+    let stats = Arc::new(Stats::new());
+    let upstream_manager = Arc::new(UpstreamManager::new(
+        vec![UpstreamConfig {
+            upstream_type: UpstreamType::Direct {
+                interface: None,
+                bind_addresses: None,
+            },
+            weight: 1,
+            enabled: true,
+            scopes: String::new(),
+            selected_scope: String::new(),
+        }],
+        1,
+        1,
+        1,
+        10,
+        1,
+        false,
+        stats.clone(),
+    ));
+    let replay_checker = Arc::new(ReplayChecker::new(128, Duration::from_secs(60)));
+    let buffer_pool = Arc::new(BufferPool::new());
+    let rng = Arc::new(SecureRandom::new());
+    let route_runtime = Arc::new(RouteRuntimeController::new(RelayRouteMode::Direct));
+    let ip_tracker = Arc::new(UserIpTracker::new());
+    let beobachten = Arc::new(BeobachtenStore::new());
+
+    let server_task = {
+        let config = config.clone();
+        let stats = stats.clone();
+        let upstream_manager = upstream_manager.clone();
+        let replay_checker = replay_checker.clone();
+        let buffer_pool = buffer_pool.clone();
+        let rng = rng.clone();
+        let route_runtime = route_runtime.clone();
+        let ip_tracker = ip_tracker.clone();
+        let beobachten = beobachten.clone();
+
+        tokio::spawn(async move {
+            let (stream, peer) = front_listener.accept().await.unwrap();
+            let real_peer_report = Arc::new(std::sync::Mutex::new(None));
+            ClientHandler::new(
+                stream,
+                peer,
+                config,
+                stats,
+                upstream_manager,
+                replay_checker,
+                buffer_pool,
+                rng,
+                None,
+                route_runtime,
+                None,
+                ip_tracker,
+                beobachten,
+                false,
+                real_peer_report,
+            )
+            .run()
+            .await
+        })
+    };
+
+    let _client = TcpStream::connect(front_addr).await.unwrap();
+
+    // Let the accepted connection reach the idle wait before advancing paused time.
+    tokio::task::yield_now().await;
+    tokio::time::advance(Duration::from_secs(2)).await;
+    tokio::task::yield_now().await;
+
+    let result = tokio::time::timeout(Duration::from_secs(1), server_task)
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(result.is_ok());
+    assert_eq!(stats.get_handshake_timeouts(), 0);
+    assert_eq!(stats.get_connects_bad(), 0);
+}
+
 #[tokio::test]
 async fn partial_tls_header_stall_triggers_handshake_timeout() {
     let mut cfg = ProxyConfig::default();
@@ -1332,6 +1500,7 @@ async fn partial_tls_header_stall_triggers_handshake_timeout() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1477,6 +1646,148 @@ fn wrap_tls_application_data(payload: &[u8]) -> Vec<u8> {
     record
 }
 
+fn wrap_tls_ccs_record() -> Vec<u8> {
+    let mut record = Vec::with_capacity(6);
+    record.push(TLS_RECORD_CHANGE_CIPHER);
+    record.extend_from_slice(&[0x03, 0x03]);
+    record.extend_from_slice(&1u16.to_be_bytes());
+    record.push(0x01);
+    record
+}
+
+fn make_valid_mtproto_handshake(
+    secret_hex: &str,
+    proto_tag: ProtoTag,
+    dc_idx: i16,
+) -> [u8; HANDSHAKE_LEN] {
+    let secret = hex::decode(secret_hex).expect("secret hex must decode for mtproto test helper");
+
+    let mut handshake = [0x5Au8; HANDSHAKE_LEN];
+    for (idx, b) in handshake[SKIP_LEN..SKIP_LEN + PREKEY_LEN + IV_LEN]
+        .iter_mut()
+        .enumerate()
+    {
+        *b = (idx as u8).wrapping_add(1);
+    }
+
+    let dec_prekey = &handshake[SKIP_LEN..SKIP_LEN + PREKEY_LEN];
+    let dec_iv_bytes = &handshake[SKIP_LEN + PREKEY_LEN..SKIP_LEN + PREKEY_LEN + IV_LEN];
+
+    let mut dec_key_input = Vec::with_capacity(PREKEY_LEN + secret.len());
+    dec_key_input.extend_from_slice(dec_prekey);
+    dec_key_input.extend_from_slice(&secret);
+    let dec_key = sha256(&dec_key_input);
+
+    let mut dec_iv_arr = [0u8; IV_LEN];
+    dec_iv_arr.copy_from_slice(dec_iv_bytes);
+    let dec_iv = u128::from_be_bytes(dec_iv_arr);
+
+    let mut stream = AesCtr::new(&dec_key, dec_iv);
+    let keystream = stream.encrypt(&[0u8; HANDSHAKE_LEN]);
+
+    let mut target_plain = [0u8; HANDSHAKE_LEN];
+    target_plain[PROTO_TAG_POS..PROTO_TAG_POS + 4].copy_from_slice(&proto_tag.to_bytes());
+    target_plain[DC_IDX_POS..DC_IDX_POS + 2].copy_from_slice(&dc_idx.to_le_bytes());
+
+    for idx in PROTO_TAG_POS..HANDSHAKE_LEN {
+        handshake[idx] = target_plain[idx] ^ keystream[idx];
+    }
+
+    handshake
+}
+
+#[tokio::test]
+async fn fragmented_tls_mtproto_with_interleaved_ccs_is_accepted() {
+    let secret_hex = "55555555555555555555555555555555";
+    let secret = [0x55u8; 16];
+    let client_hello = make_valid_tls_client_hello(&secret, 0);
+    let mtproto_handshake = make_valid_mtproto_handshake(secret_hex, ProtoTag::Secure, 2);
+
+    let mut cfg = ProxyConfig::default();
+    cfg.general.beobachten = false;
+    cfg.access.ignore_time_skew = true;
+    cfg.access
+        .users
+        .insert("user".to_string(), secret_hex.to_string());
+
+    let config = Arc::new(cfg);
+    let replay_checker = Arc::new(ReplayChecker::new(128, Duration::from_secs(60)));
+    let rng = SecureRandom::new();
+
+    let (server_side, mut client_side) = duplex(131072);
+    let peer: SocketAddr = "198.51.100.85:55007".parse().unwrap();
+    let (read_half, write_half) = tokio::io::split(server_side);
+
+    let (mut tls_reader, tls_writer, tls_user) = match handle_tls_handshake(
+        &client_hello,
+        read_half,
+        write_half,
+        peer,
+        &config,
+        &replay_checker,
+        &rng,
+        None,
+    )
+    .await
+    {
+        HandshakeResult::Success(result) => result,
+        _ => panic!("expected successful TLS handshake"),
+    };
+
+    let mut tls_response_head = [0u8; 5];
+    client_side
+        .read_exact(&mut tls_response_head)
+        .await
+        .unwrap();
+    assert_eq!(tls_response_head[0], 0x16);
+    let tls_response_len =
+        u16::from_be_bytes([tls_response_head[3], tls_response_head[4]]) as usize;
+    let mut tls_response_body = vec![0u8; tls_response_len];
+    client_side
+        .read_exact(&mut tls_response_body)
+        .await
+        .unwrap();
+
+    client_side
+        .write_all(&wrap_tls_application_data(&mtproto_handshake[..13]))
+        .await
+        .unwrap();
+    client_side.write_all(&wrap_tls_ccs_record()).await.unwrap();
+    client_side
+        .write_all(&wrap_tls_application_data(&mtproto_handshake[13..37]))
+        .await
+        .unwrap();
+    client_side.write_all(&wrap_tls_ccs_record()).await.unwrap();
+    client_side
+        .write_all(&wrap_tls_application_data(&mtproto_handshake[37..]))
+        .await
+        .unwrap();
+
+    let mtproto_data = tls_reader.read_exact(HANDSHAKE_LEN).await.unwrap();
+    assert_eq!(&mtproto_data[..], &mtproto_handshake);
+
+    let mtproto_handshake: [u8; HANDSHAKE_LEN] = mtproto_data[..].try_into().unwrap();
+    let (_, _, success) = match handle_mtproto_handshake(
+        &mtproto_handshake,
+        tls_reader,
+        tls_writer,
+        peer,
+        &config,
+        &replay_checker,
+        true,
+        Some(tls_user.as_str()),
+    )
+    .await
+    {
+        HandshakeResult::Success(result) => result,
+        _ => panic!("expected successful MTProto handshake"),
+    };
+
+    assert_eq!(success.user, "user");
+    assert_eq!(success.proto_tag, ProtoTag::Secure);
+    assert_eq!(success.dc_idx, 2);
+}
+
 #[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();
@@ -1514,6 +1825,7 @@ async fn valid_tls_path_does_not_fall_back_to_mask_backend() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1622,6 +1934,7 @@ async fn valid_tls_with_invalid_mtproto_falls_back_to_mask_backend() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1728,6 +2041,7 @@ async fn client_handler_tls_bad_mtproto_is_forwarded_to_mask_backend() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1849,6 +2163,7 @@ async fn alpn_mismatch_tls_probe_is_masked_through_client_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1941,6 +2256,7 @@ async fn invalid_hmac_tls_probe_is_masked_through_client_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -2039,6 +2355,7 @@ async fn burst_invalid_tls_probes_are_masked_verbatim() {
             1,
             1,
             1,
+            10,
             1,
             false,
             stats.clone(),
@@ -2217,14 +2534,16 @@ async fn tcp_limit_rejection_does_not_reserve_ip_or_trigger_rollback() {
 }
 
 #[tokio::test]
-async fn zero_tcp_limit_rejects_without_ip_or_counter_side_effects() {
+async fn zero_tcp_limit_uses_global_fallback_and_rejects_without_side_effects() {
     let mut config = ProxyConfig::default();
     config
         .access
         .user_max_tcp_conns
         .insert("user".to_string(), 0);
+    config.access.user_max_tcp_conns_global_each = 1;
 
     let stats = Stats::new();
+    stats.increment_user_curr_connects("user");
     let ip_tracker = UserIpTracker::new();
     let peer_addr: SocketAddr = "198.51.100.211:50001".parse().unwrap();
 
@@ -2241,10 +2560,75 @@ async fn zero_tcp_limit_rejects_without_ip_or_counter_side_effects() {
         result,
         Err(ProxyError::ConnectionLimitExceeded { user }) if user == "user"
     ));
+    assert_eq!(
+        stats.get_user_curr_connects("user"),
+        1,
+        "TCP-limit rejection must keep pre-existing in-flight connection count unchanged"
+    );
+    assert_eq!(ip_tracker.get_active_ip_count("user").await, 0);
+}
+
+#[tokio::test]
+async fn zero_tcp_limit_with_disabled_global_fallback_is_unlimited() {
+    let mut config = ProxyConfig::default();
+    config
+        .access
+        .user_max_tcp_conns
+        .insert("user".to_string(), 0);
+    config.access.user_max_tcp_conns_global_each = 0;
+
+    let stats = Stats::new();
+    let ip_tracker = UserIpTracker::new();
+    let peer_addr: SocketAddr = "198.51.100.212:50002".parse().unwrap();
+
+    let result = RunningClientHandler::check_user_limits_static(
+        "user",
+        &config,
+        &stats,
+        peer_addr,
+        &ip_tracker,
+    )
+    .await;
+
+    assert!(
+        result.is_ok(),
+        "per-user zero with global fallback disabled must not enforce a TCP limit"
+    );
     assert_eq!(stats.get_user_curr_connects("user"), 0);
     assert_eq!(ip_tracker.get_active_ip_count("user").await, 0);
 }
 
+#[tokio::test]
+async fn global_tcp_fallback_applies_when_per_user_limit_is_missing() {
+    let mut config = ProxyConfig::default();
+    config.access.user_max_tcp_conns_global_each = 1;
+
+    let stats = Stats::new();
+    stats.increment_user_curr_connects("user");
+    let ip_tracker = UserIpTracker::new();
+    let peer_addr: SocketAddr = "198.51.100.213:50003".parse().unwrap();
+
+    let result = RunningClientHandler::check_user_limits_static(
+        "user",
+        &config,
+        &stats,
+        peer_addr,
+        &ip_tracker,
+    )
+    .await;
+
+    assert!(matches!(
+        result,
+        Err(ProxyError::ConnectionLimitExceeded { user }) if user == "user"
+    ));
+    assert_eq!(
+        stats.get_user_curr_connects("user"),
+        1,
+        "Global fallback TCP-limit rejection must keep pre-existing counter unchanged"
+    );
+    assert_eq!(ip_tracker.get_active_ip_count("user").await, 0);
+}
+
 #[tokio::test]
 async fn check_user_limits_static_success_does_not_leak_counter_or_ip_reservation() {
     let user = "check-helper-user";
@@ -2876,6 +3260,7 @@ async fn relay_connect_error_releases_user_and_ip_before_return() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -3436,6 +3821,7 @@ async fn untrusted_proxy_header_source_is_rejected() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -3505,6 +3891,7 @@ async fn empty_proxy_trusted_cidrs_rejects_proxy_header_by_default() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -3601,6 +3988,7 @@ async fn oversized_tls_record_is_masked_in_generic_stream_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -3703,6 +4091,7 @@ async fn oversized_tls_record_is_masked_in_client_handler_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -3819,6 +4208,7 @@ async fn tls_record_len_min_minus_1_is_rejected_in_generic_stream_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -3921,6 +4311,7 @@ async fn tls_record_len_min_minus_1_is_rejected_in_client_handler_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -4026,6 +4417,7 @@ async fn tls_record_len_16384_is_accepted_in_generic_stream_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -4126,6 +4518,7 @@ async fn tls_record_len_16384_is_accepted_in_client_handler_pipeline() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),

src/proxy/tests/client_timing_profile_adversarial_tests.rs

@@ -33,6 +33,7 @@ fn make_test_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_tls_clienthello_size_security_tests.rs

@@ -35,6 +35,7 @@ fn make_test_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_tls_clienthello_truncation_adversarial_tests.rs

@@ -36,6 +36,7 @@ fn make_test_upstream_manager(stats: Arc<Stats>) -> Arc<UpstreamManager> {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats,

src/proxy/tests/client_tls_mtproto_fallback_security_tests.rs

@@ -50,6 +50,7 @@ fn build_harness(secret_hex: &str, mask_port: u16) -> PipelineHarness {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),

src/proxy/tests/direct_relay_security_tests.rs

@@ -1302,6 +1302,7 @@ async fn direct_relay_abort_midflight_releases_route_gauge() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1408,6 +1409,7 @@ async fn direct_relay_cutover_midflight_releases_route_gauge() {
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1529,6 +1531,7 @@ async fn direct_relay_cutover_storm_multi_session_keeps_generic_errors_and_relea
         1,
         1,
         1,
+        10,
         1,
         false,
         stats.clone(),
@@ -1761,6 +1764,7 @@ async fn negative_direct_relay_dc_connection_refused_fails_fast() {
         1,
         100,
         5000,
+        10,
         3,
         false,
         stats.clone(),
@@ -1851,6 +1855,7 @@ async fn adversarial_direct_relay_cutover_integrity() {
         1,
         100,
         5000,
+        10,
         3,
         false,
         stats.clone(),

src/service/mod.rs

@@ -111,11 +111,17 @@ pub fn generate_service_file(init_system: InitSystem, opts: &ServiceOptions) ->
 /// Generates an enhanced systemd unit file.
 fn generate_systemd_unit(opts: &ServiceOptions) -> String {
     let user_line = opts.user.map(|u| format!("User={}", u)).unwrap_or_default();
-    let group_line = opts.group.map(|g| format!("Group={}", g)).unwrap_or_default();
-    let working_dir = opts.working_dir.map(|d| format!("WorkingDirectory={}", d)).unwrap_or_default();
+    let group_line = opts
+        .group
+        .map(|g| format!("Group={}", g))
+        .unwrap_or_default();
+    let working_dir = opts
+        .working_dir
+        .map(|d| format!("WorkingDirectory={}", d))
+        .unwrap_or_default();
 
     format!(
-r#"[Unit]
+        r#"[Unit]
 Description={description}
 Documentation=https://github.com/telemt/telemt
 After=network-online.target
@@ -176,7 +182,7 @@ fn generate_openrc_script(opts: &ServiceOptions) -> String {
     let group = opts.group.unwrap_or("root");
 
     format!(
-r#"#!/sbin/openrc-run
+        r#"#!/sbin/openrc-run
 # OpenRC init script for telemt
 
 description="{description}"
@@ -218,7 +224,7 @@ fn generate_freebsd_rc_script(opts: &ServiceOptions) -> String {
     let group = opts.group.unwrap_or("wheel");
 
     format!(
-r#"#!/bin/sh
+        r#"#!/bin/sh
 #
 # PROVIDE: telemt
 # REQUIRE: LOGIN NETWORKING
@@ -284,7 +290,7 @@ run_rc_command "$1"
 pub fn installation_instructions(init_system: InitSystem) -> &'static str {
     match init_system {
         InitSystem::Systemd => {
-r#"To install and enable the service:
+            r#"To install and enable the service:
   sudo systemctl daemon-reload
   sudo systemctl enable telemt
   sudo systemctl start telemt
@@ -300,7 +306,7 @@ To reload configuration:
 "#
         }
         InitSystem::OpenRC => {
-r#"To install and enable the service:
+            r#"To install and enable the service:
   sudo chmod +x /etc/init.d/telemt
   sudo rc-update add telemt default
   sudo rc-service telemt start
@@ -313,7 +319,7 @@ To reload configuration:
 "#
         }
         InitSystem::FreeBSDRc => {
-r#"To install and enable the service:
+            r#"To install and enable the service:
   sudo chmod +x /usr/local/etc/rc.d/telemt
   sudo sysrc telemt_enable="YES"
   sudo service telemt start
@@ -326,7 +332,7 @@ To reload configuration:
 "#
         }
         InitSystem::Unknown => {
-r#"No supported init system detected.
+            r#"No supported init system detected.
 You may need to create a service file manually or run telemt directly:
   telemt start /etc/telemt/config.toml
 "#
@@ -369,8 +375,14 @@ mod tests {
 
     #[test]
     fn test_service_file_paths() {
-        assert_eq!(service_file_path(InitSystem::Systemd), "/etc/systemd/system/telemt.service");
+        assert_eq!(
+            service_file_path(InitSystem::Systemd),
+            "/etc/systemd/system/telemt.service"
+        );
         assert_eq!(service_file_path(InitSystem::OpenRC), "/etc/init.d/telemt");
-        assert_eq!(service_file_path(InitSystem::FreeBSDRc), "/usr/local/etc/rc.d/telemt");
+        assert_eq!(
+            service_file_path(InitSystem::FreeBSDRc),
+            "/usr/local/etc/rc.d/telemt"
+        );
     }
 }

src/transport/upstream.rs

@@ -34,8 +34,6 @@ const NUM_DCS: usize = 5;
 
 /// Timeout for individual DC ping attempt
 const DC_PING_TIMEOUT_SECS: u64 = 5;
-/// Timeout for direct TG DC TCP connect readiness.
-const DIRECT_CONNECT_TIMEOUT_SECS: u64 = 10;
 /// Interval between upstream health-check cycles.
 const HEALTH_CHECK_INTERVAL_SECS: u64 = 30;
 /// Timeout for a single health-check connect attempt.
@@ -319,6 +317,8 @@ pub struct UpstreamManager {
     connect_retry_attempts: u32,
     connect_retry_backoff: Duration,
     connect_budget: Duration,
+    /// Per-attempt TCP connect timeout to Telegram DC (`[general] tg_connect`, seconds).
+    tg_connect_timeout_secs: u64,
     unhealthy_fail_threshold: u32,
     connect_failfast_hard_errors: bool,
     no_upstreams_warn_epoch_ms: Arc<AtomicU64>,
@@ -332,6 +332,7 @@ impl UpstreamManager {
         connect_retry_attempts: u32,
         connect_retry_backoff_ms: u64,
         connect_budget_ms: u64,
+        tg_connect_timeout_secs: u64,
         unhealthy_fail_threshold: u32,
         connect_failfast_hard_errors: bool,
         stats: Arc<Stats>,
@@ -347,6 +348,7 @@ impl UpstreamManager {
             connect_retry_attempts: connect_retry_attempts.max(1),
             connect_retry_backoff: Duration::from_millis(connect_retry_backoff_ms),
             connect_budget: Duration::from_millis(connect_budget_ms.max(1)),
+            tg_connect_timeout_secs: tg_connect_timeout_secs.max(1),
             unhealthy_fail_threshold: unhealthy_fail_threshold.max(1),
             connect_failfast_hard_errors,
             no_upstreams_warn_epoch_ms: Arc::new(AtomicU64::new(0)),
@@ -798,7 +800,7 @@ impl UpstreamManager {
             }
             let remaining_budget = self.connect_budget.saturating_sub(elapsed);
             let attempt_timeout =
-                Duration::from_secs(DIRECT_CONNECT_TIMEOUT_SECS).min(remaining_budget);
+                Duration::from_secs(self.tg_connect_timeout_secs).min(remaining_budget);
             if attempt_timeout.is_zero() {
                 last_error = Some(ProxyError::ConnectionTimeout {
                     addr: target.to_string(),
@@ -1901,6 +1903,7 @@ mod tests {
             1,
             100,
             1000,
+            10,
             1,
             false,
             Arc::new(Stats::new()),