Docker Health-Check

Fixes in TLS-F

Cargo.lock

@@ -2780,7 +2780,7 @@ checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
 
 [[package]]
 name = "telemt"
-version = "3.4.1"
+version = "3.4.2"
 dependencies = [
  "aes",
  "anyhow",

Cargo.toml

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

Dockerfile

@@ -77,6 +77,8 @@ COPY config.toml /app/config.toml
 
 EXPOSE 443 9090 9091
 
+HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 CMD ["/app/telemt", "healthcheck", "/app/config.toml", "--mode", "liveness"]
+
 ENTRYPOINT ["/app/telemt"]
 CMD ["config.toml"]
 
@@ -94,5 +96,7 @@ USER nonroot:nonroot
 
 EXPOSE 443 9090 9091
 
+HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 CMD ["/app/telemt", "healthcheck", "/app/config.toml", "--mode", "liveness"]
+
 ENTRYPOINT ["/app/telemt"]
 CMD ["config.toml"]

README.md

@@ -1,6 +1,6 @@
 # Telemt - MTProxy on Rust + Tokio
 
-![Latest Release](https://img.shields.io/github/v/release/telemt/telemt?color=neon) ![Stars](https://img.shields.io/github/stars/telemt/telemt?style=social) ![Forks](https://img.shields.io/github/forks/telemt/telemt?style=social) [![Telegram](https://img.shields.io/badge/Telegram-Chat-24a1de?logo=telegram&logoColor=24a1de)](https://t.me/telemtrs)
+[![Latest Release](https://img.shields.io/github/v/release/telemt/telemt?color=neon)](https://github.com/telemt/telemt/releases/latest) [![Stars](https://img.shields.io/github/stars/telemt/telemt?style=social)](https://github.com/telemt/telemt/stargazers) [![Forks](https://img.shields.io/github/forks/telemt/telemt?style=social)](https://github.com/telemt/telemt/network/members) [![Telegram](https://img.shields.io/badge/Telegram-Chat-24a1de?logo=telegram&logoColor=24a1de)](https://t.me/telemtrs)
 
 [🇷🇺 README на русском](https://github.com/telemt/telemt/blob/main/README.ru.md)
 

README.ru.md

@@ -1,6 +1,6 @@
 # Telemt — MTProxy на Rust + Tokio
 
-![Latest Release](https://img.shields.io/github/v/release/telemt/telemt?color=neon) ![Stars](https://img.shields.io/github/stars/telemt/telemt?style=social) ![Forks](https://img.shields.io/github/forks/telemt/telemt?style=social) [![Telegram](https://img.shields.io/badge/Telegram-Chat-24a1de?logo=telegram&logoColor=24a1de)](https://t.me/telemtrs)
+[![Latest Release](https://img.shields.io/github/v/release/telemt/telemt?color=neon)](https://github.com/telemt/telemt/releases/latest) [![Stars](https://img.shields.io/github/stars/telemt/telemt?style=social)](https://github.com/telemt/telemt/stargazers) [![Forks](https://img.shields.io/github/forks/telemt/telemt?style=social)](https://github.com/telemt/telemt/network/members) [![Telegram](https://img.shields.io/badge/Telegram-Chat-24a1de?logo=telegram&logoColor=24a1de)](https://t.me/telemtrs)
 
 ***Решает проблемы раньше, чем другие узнают об их существовании***
 

docker-compose.yml

@@ -16,6 +16,12 @@ services:
       - /etc/telemt:rw,mode=1777,size=4m
     environment:
       - RUST_LOG=info
+    healthcheck:
+      test: [ "CMD", "/app/telemt", "healthcheck", "/etc/telemt/config.toml", "--mode", "liveness" ]
+      interval: 30s
+      timeout: 5s
+      retries: 3
+      start_period: 20s
     # Uncomment this line if you want to use host network for IPv6, but bridge is default and usually better
     # network_mode: host
     cap_drop:

docs/Setup_examples/XRAY_DOUBLE_HOP.en.md

@@ -37,13 +37,13 @@ xray x25519
 ```
 3. **Short ID (Reality identifier):**
 ```bash
-openssl rand -hex 16
-# Save the output (e.g.: 0123456789abcdef0123456789abcdef) — this is <SHORT_ID>
+openssl rand -hex 8
+# Save the output (e.g.: abc123def456) — this is <SHORT_ID>
 ```
 4. **Random Path (for xhttp):**
 ```bash
-openssl rand -hex 8
-# Save the output (e.g., abc123def456) to replace <YOUR_RANDOM_PATH> in configs
+openssl rand -hex 16
+# Save the output (e.g., 0123456789abcdef0123456789abcdef) to replace <YOUR_RANDOM_PATH> in configs
 ```
 
 ---

docs/Setup_examples/XRAY_DOUBLE_HOP.ru.md

@@ -37,13 +37,13 @@ xray x25519
 ```
 3. **Short ID (идентификатор Reality):**
 ```bash
-openssl rand -hex 16
-# Сохраните вывод (например: 0123456789abcdef0123456789abcdef) — это <SHORT_ID>
+openssl rand -hex 8
+# Сохраните вывод (например: abc123def456) — это <SHORT_ID>
 ```
 4. **Random Path (путь для xhttp):**
 ```bash
-openssl rand -hex 8
-# Сохраните вывод (например, abc123def456), чтобы заменить <YOUR_RANDOM_PATH> в конфигах
+openssl rand -hex 16
+# Сохраните вывод (например, 0123456789abcdef0123456789abcdef), чтобы заменить <YOUR_RANDOM_PATH> в конфигах
 ```
 
 ---

src/api/mod.rs

@@ -41,7 +41,7 @@ use config_store::{current_revision, load_config_from_disk, parse_if_match};
 use events::ApiEventStore;
 use http_utils::{error_response, read_json, read_optional_json, success_response};
 use model::{
-    ApiFailure, CreateUserRequest, DeleteUserResponse, HealthData, PatchUserRequest,
+    ApiFailure, CreateUserRequest, DeleteUserResponse, HealthData, HealthReadyData, PatchUserRequest,
     RotateSecretRequest, SummaryData, UserActiveIps,
 };
 use runtime_edge::{
@@ -275,6 +275,33 @@ async fn handle(
                 };
                 Ok(success_response(StatusCode::OK, data, revision))
             }
+            ("GET", "/v1/health/ready") => {
+                let revision = current_revision(&shared.config_path).await?;
+                let admission_open = shared.runtime_state.admission_open.load(Ordering::Relaxed);
+                let upstream_health = shared.upstream_manager.api_health_summary().await;
+                let ready = admission_open && upstream_health.healthy_total > 0;
+                let reason = if ready {
+                    None
+                } else if !admission_open {
+                    Some("admission_closed")
+                } else {
+                    Some("no_healthy_upstreams")
+                };
+                let data = HealthReadyData {
+                    ready,
+                    status: if ready { "ready" } else { "not_ready" },
+                    reason,
+                    admission_open,
+                    healthy_upstreams: upstream_health.healthy_total,
+                    total_upstreams: upstream_health.configured_total,
+                };
+                let status_code = if ready {
+                    StatusCode::OK
+                } else {
+                    StatusCode::SERVICE_UNAVAILABLE
+                };
+                Ok(success_response(status_code, data, revision))
+            }
             ("GET", "/v1/system/info") => {
                 let revision = current_revision(&shared.config_path).await?;
                 let data = build_system_info_data(shared.as_ref(), cfg.as_ref(), &revision);

src/api/model.rs

@@ -60,6 +60,17 @@ pub(super) struct HealthData {
     pub(super) read_only: bool,
 }
 
+#[derive(Serialize)]
+pub(super) struct HealthReadyData {
+    pub(super) ready: bool,
+    pub(super) status: &'static str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) reason: Option<&'static str>,
+    pub(super) admission_open: bool,
+    pub(super) healthy_upstreams: usize,
+    pub(super) total_upstreams: usize,
+}
+
 #[derive(Serialize)]
 pub(super) struct SummaryData {
     pub(super) uptime_seconds: f64,

src/cli.rs

@@ -6,12 +6,15 @@
 //! - `reload [--pid-file PATH]` - Reload configuration (SIGHUP)
 //! - `status [--pid-file PATH]` - Check daemon status
 //! - `run [OPTIONS] [config.toml]` - Run in foreground (default behavior)
+//! - `healthcheck [OPTIONS] [config.toml]` - Run control-plane health probe
 
 use rand::RngExt;
 use std::fs;
 use std::path::{Path, PathBuf};
 use std::process::Command;
 
+use crate::healthcheck::{self, HealthcheckMode};
+
 #[cfg(unix)]
 use crate::daemon::{self, DEFAULT_PID_FILE, DaemonOptions};
 
@@ -28,6 +31,8 @@ pub enum Subcommand {
     Reload,
     /// Check daemon status (`status` subcommand).
     Status,
+    /// Run health probe and exit with status code.
+    Healthcheck,
     /// Fire-and-forget setup (`--init`).
     Init,
 }
@@ -38,6 +43,8 @@ pub struct ParsedCommand {
     pub subcommand: Subcommand,
     pub pid_file: PathBuf,
     pub config_path: String,
+    pub healthcheck_mode: HealthcheckMode,
+    pub healthcheck_mode_invalid: Option<String>,
     #[cfg(unix)]
     pub daemon_opts: DaemonOptions,
     pub init_opts: Option<InitOptions>,
@@ -52,6 +59,8 @@ impl Default for ParsedCommand {
             #[cfg(not(unix))]
             pid_file: PathBuf::from("/var/run/telemt.pid"),
             config_path: "config.toml".to_string(),
+            healthcheck_mode: HealthcheckMode::Liveness,
+            healthcheck_mode_invalid: None,
             #[cfg(unix)]
             daemon_opts: DaemonOptions::default(),
             init_opts: None,
@@ -91,6 +100,9 @@ pub fn parse_command(args: &[String]) -> ParsedCommand {
             "status" => {
                 cmd.subcommand = Subcommand::Status;
             }
+            "healthcheck" => {
+                cmd.subcommand = Subcommand::Healthcheck;
+            }
             "run" => {
                 cmd.subcommand = Subcommand::Run;
                 #[cfg(unix)]
@@ -113,7 +125,35 @@ pub fn parse_command(args: &[String]) -> ParsedCommand {
     while i < args.len() {
         match args[i].as_str() {
             // Skip subcommand names
-            "start" | "stop" | "reload" | "status" | "run" => {}
+            "start" | "stop" | "reload" | "status" | "run" | "healthcheck" => {}
+            "--mode" => {
+                i += 1;
+                if i < args.len() {
+                    match HealthcheckMode::from_cli_arg(&args[i]) {
+                        Some(mode) => {
+                            cmd.healthcheck_mode = mode;
+                            cmd.healthcheck_mode_invalid = None;
+                        }
+                        None => {
+                            cmd.healthcheck_mode_invalid = Some(args[i].clone());
+                        }
+                    }
+                } else {
+                    cmd.healthcheck_mode_invalid = Some(String::new());
+                }
+            }
+            s if s.starts_with("--mode=") => {
+                let raw = s.trim_start_matches("--mode=");
+                match HealthcheckMode::from_cli_arg(raw) {
+                    Some(mode) => {
+                        cmd.healthcheck_mode = mode;
+                        cmd.healthcheck_mode_invalid = None;
+                    }
+                    None => {
+                        cmd.healthcheck_mode_invalid = Some(raw.to_string());
+                    }
+                }
+            }
             // PID file option (for stop/reload/status)
             "--pid-file" => {
                 i += 1;
@@ -152,6 +192,20 @@ pub fn execute_subcommand(cmd: &ParsedCommand) -> Option<i32> {
         Subcommand::Stop => Some(cmd_stop(&cmd.pid_file)),
         Subcommand::Reload => Some(cmd_reload(&cmd.pid_file)),
         Subcommand::Status => Some(cmd_status(&cmd.pid_file)),
+        Subcommand::Healthcheck => {
+            if let Some(invalid_mode) = cmd.healthcheck_mode_invalid.as_ref() {
+                if invalid_mode.is_empty() {
+                    eprintln!("[telemt] Missing value for --mode (supported: liveness, ready)");
+                } else {
+                    eprintln!(
+                        "[telemt] Invalid --mode value '{invalid_mode}' (supported: liveness, ready)"
+                    );
+                }
+                Some(2)
+            } else {
+                Some(healthcheck::run(&cmd.config_path, cmd.healthcheck_mode))
+            }
+        }
         Subcommand::Init => {
             if let Some(opts) = cmd.init_opts.clone() {
                 match run_init(opts) {
@@ -177,6 +231,20 @@ pub fn execute_subcommand(cmd: &ParsedCommand) -> Option<i32> {
             eprintln!("[telemt] Subcommand not supported on this platform");
             Some(1)
         }
+        Subcommand::Healthcheck => {
+            if let Some(invalid_mode) = cmd.healthcheck_mode_invalid.as_ref() {
+                if invalid_mode.is_empty() {
+                    eprintln!("[telemt] Missing value for --mode (supported: liveness, ready)");
+                } else {
+                    eprintln!(
+                        "[telemt] Invalid --mode value '{invalid_mode}' (supported: liveness, ready)"
+                    );
+                }
+                Some(2)
+            } else {
+                Some(healthcheck::run(&cmd.config_path, cmd.healthcheck_mode))
+            }
+        }
         Subcommand::Init => {
             if let Some(opts) = cmd.init_opts.clone() {
                 match run_init(opts) {

src/config/hot_reload.rs

@@ -122,7 +122,8 @@ pub struct HotFields {
     pub user_expirations: std::collections::HashMap<String, chrono::DateTime<chrono::Utc>>,
     pub user_data_quota: std::collections::HashMap<String, u64>,
     pub user_rate_limits: std::collections::HashMap<String, crate::config::RateLimitBps>,
-    pub cidr_rate_limits: std::collections::HashMap<ipnetwork::IpNetwork, crate::config::RateLimitBps>,
+    pub cidr_rate_limits:
+        std::collections::HashMap<ipnetwork::IpNetwork, crate::config::RateLimitBps>,
     pub user_max_unique_ips: std::collections::HashMap<String, usize>,
     pub user_max_unique_ips_global_each: usize,
     pub user_max_unique_ips_mode: crate::config::UserMaxUniqueIpsMode,

src/daemon/mod.rs

@@ -8,8 +8,8 @@ use std::io::{self, Read, Write};
 use std::os::unix::fs::OpenOptionsExt;
 use std::path::{Path, PathBuf};
 
-use nix::fcntl::{Flock, FlockArg};
 use nix::errno::Errno;
+use nix::fcntl::{Flock, FlockArg};
 use nix::unistd::{self, ForkResult, Gid, Pid, Uid, chdir, close, fork, getpid, setsid};
 use tracing::{debug, info, warn};
 
@@ -158,15 +158,15 @@ fn redirect_stdio_to_devnull() -> Result<(), DaemonError> {
     unsafe {
         // Redirect stdin (fd 0)
         if libc::dup2(devnull_fd, 0) < 0 {
-            return Err(DaemonError::RedirectFailed(nix::errno::Errno::last()));
+            return Err(DaemonError::RedirectFailed(Errno::last()));
         }
         // Redirect stdout (fd 1)
         if libc::dup2(devnull_fd, 1) < 0 {
-            return Err(DaemonError::RedirectFailed(nix::errno::Errno::last()));
+            return Err(DaemonError::RedirectFailed(Errno::last()));
         }
         // Redirect stderr (fd 2)
         if libc::dup2(devnull_fd, 2) < 0 {
-            return Err(DaemonError::RedirectFailed(nix::errno::Errno::last()));
+            return Err(DaemonError::RedirectFailed(Errno::last()));
         }
     }
 
@@ -350,11 +350,7 @@ fn set_supplementary_groups(gid: Gid) -> Result<(), nix::Error> {
                 groups.as_ptr(),
             )
         };
-        if rc == 0 {
-            Ok(())
-        } else {
-            Err(Errno::last())
-        }
+        if rc == 0 { Ok(()) } else { Err(Errno::last()) }
     }
 
     #[cfg(not(target_os = "macos"))]

src/healthcheck.rs

@@ -0,0 +1,208 @@
+use std::io::{Read, Write};
+use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, TcpStream};
+use std::time::Duration;
+
+use serde_json::Value;
+
+use crate::config::ProxyConfig;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum HealthcheckMode {
+    Liveness,
+    Ready,
+}
+
+impl HealthcheckMode {
+    pub(crate) fn from_cli_arg(value: &str) -> Option<Self> {
+        match value {
+            "liveness" => Some(Self::Liveness),
+            "ready" => Some(Self::Ready),
+            _ => None,
+        }
+    }
+
+    fn request_path(self) -> &'static str {
+        match self {
+            Self::Liveness => "/v1/health",
+            Self::Ready => "/v1/health/ready",
+        }
+    }
+}
+
+pub(crate) fn run(config_path: &str, mode: HealthcheckMode) -> i32 {
+    match run_inner(config_path, mode) {
+        Ok(()) => 0,
+        Err(error) => {
+            eprintln!("[telemt] healthcheck failed: {error}");
+            1
+        }
+    }
+}
+
+fn run_inner(config_path: &str, mode: HealthcheckMode) -> Result<(), String> {
+    let config =
+        ProxyConfig::load(config_path).map_err(|error| format!("config load failed: {error}"))?;
+    let api_cfg = &config.server.api;
+    if !api_cfg.enabled {
+        return Ok(());
+    }
+
+    let listen: SocketAddr = api_cfg
+        .listen
+        .parse()
+        .map_err(|_| format!("invalid API listen address: {}", api_cfg.listen))?;
+    if listen.port() == 0 {
+        return Err("API listen port is 0".to_string());
+    }
+    let target = probe_target(listen);
+
+    let mut stream = TcpStream::connect_timeout(&target, Duration::from_secs(2))
+        .map_err(|error| format!("connect {target} failed: {error}"))?;
+    stream
+        .set_read_timeout(Some(Duration::from_secs(2)))
+        .map_err(|error| format!("set read timeout failed: {error}"))?;
+    stream
+        .set_write_timeout(Some(Duration::from_secs(2)))
+        .map_err(|error| format!("set write timeout failed: {error}"))?;
+
+    let request = build_request(target, mode.request_path(), &api_cfg.auth_header);
+    stream
+        .write_all(request.as_bytes())
+        .map_err(|error| format!("request write failed: {error}"))?;
+    stream
+        .flush()
+        .map_err(|error| format!("request flush failed: {error}"))?;
+
+    let mut raw_response = Vec::new();
+    stream
+        .read_to_end(&mut raw_response)
+        .map_err(|error| format!("response read failed: {error}"))?;
+    let response =
+        String::from_utf8(raw_response).map_err(|_| "response is not valid UTF-8".to_string())?;
+
+    let (status_code, body) = split_response(&response)?;
+    if status_code != 200 {
+        return Err(format!("HTTP status {status_code}"));
+    }
+
+    validate_payload(mode, body)?;
+    Ok(())
+}
+
+fn probe_target(listen: SocketAddr) -> SocketAddr {
+    match listen {
+        SocketAddr::V4(addr) => {
+            let ip = if addr.ip().is_unspecified() {
+                Ipv4Addr::LOCALHOST
+            } else {
+                *addr.ip()
+            };
+            SocketAddr::from((ip, addr.port()))
+        }
+        SocketAddr::V6(addr) => {
+            let ip = if addr.ip().is_unspecified() {
+                Ipv6Addr::LOCALHOST
+            } else {
+                *addr.ip()
+            };
+            SocketAddr::from((ip, addr.port()))
+        }
+    }
+}
+
+fn build_request(target: SocketAddr, path: &str, auth_header: &str) -> String {
+    let mut request = format!("GET {path} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\n", target);
+    if !auth_header.is_empty() {
+        request.push_str("Authorization: ");
+        request.push_str(auth_header);
+        request.push_str("\r\n");
+    }
+    request.push_str("\r\n");
+    request
+}
+
+fn split_response(response: &str) -> Result<(u16, &str), String> {
+    let header_end = response
+        .find("\r\n\r\n")
+        .ok_or_else(|| "invalid HTTP response headers".to_string())?;
+    let header = &response[..header_end];
+    let body = &response[header_end + 4..];
+    let status_line = header
+        .lines()
+        .next()
+        .ok_or_else(|| "missing HTTP status line".to_string())?;
+    let status_code = parse_status_code(status_line)?;
+    Ok((status_code, body))
+}
+
+fn parse_status_code(status_line: &str) -> Result<u16, String> {
+    let mut parts = status_line.split_whitespace();
+    let version = parts
+        .next()
+        .ok_or_else(|| "missing HTTP version".to_string())?;
+    if !version.starts_with("HTTP/") {
+        return Err(format!("invalid HTTP status line: {status_line}"));
+    }
+    let code = parts
+        .next()
+        .ok_or_else(|| "missing HTTP status code".to_string())?;
+    code.parse::<u16>()
+        .map_err(|_| format!("invalid HTTP status code: {code}"))
+}
+
+fn validate_payload(mode: HealthcheckMode, body: &str) -> Result<(), String> {
+    let payload: Value =
+        serde_json::from_str(body).map_err(|_| "response body is not valid JSON".to_string())?;
+    if payload.get("ok").and_then(Value::as_bool) != Some(true) {
+        return Err("response JSON has ok=false".to_string());
+    }
+
+    let data = payload
+        .get("data")
+        .ok_or_else(|| "response JSON has no data field".to_string())?;
+    match mode {
+        HealthcheckMode::Liveness => {
+            if data.get("status").and_then(Value::as_str) != Some("ok") {
+                return Err("liveness status is not ok".to_string());
+            }
+        }
+        HealthcheckMode::Ready => {
+            if data.get("ready").and_then(Value::as_bool) != Some(true) {
+                return Err("readiness flag is false".to_string());
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{HealthcheckMode, parse_status_code, split_response, validate_payload};
+
+    #[test]
+    fn parse_status_code_reads_http_200() {
+        let status = parse_status_code("HTTP/1.1 200 OK").expect("must parse status");
+        assert_eq!(status, 200);
+    }
+
+    #[test]
+    fn split_response_extracts_status_and_body() {
+        let response = "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n{\"ok\":true}";
+        let (status, body) = split_response(response).expect("must split response");
+        assert_eq!(status, 200);
+        assert_eq!(body, "{\"ok\":true}");
+    }
+
+    #[test]
+    fn validate_payload_accepts_liveness_contract() {
+        let body = "{\"ok\":true,\"data\":{\"status\":\"ok\"}}";
+        validate_payload(HealthcheckMode::Liveness, body).expect("liveness payload must pass");
+    }
+
+    #[test]
+    fn validate_payload_rejects_not_ready() {
+        let body = "{\"ok\":true,\"data\":{\"ready\":false}}";
+        let result = validate_payload(HealthcheckMode::Ready, body);
+        assert!(result.is_err());
+    }
+}

src/maestro/runtime_tasks.rs

@@ -190,8 +190,16 @@ pub(crate) async fn spawn_runtime_tasks(
     );
     let mut config_rx_rate_limits = config_rx.clone();
     tokio::spawn(async move {
-        let mut prev_user_limits = config_rx_rate_limits.borrow().access.user_rate_limits.clone();
-        let mut prev_cidr_limits = config_rx_rate_limits.borrow().access.cidr_rate_limits.clone();
+        let mut prev_user_limits = config_rx_rate_limits
+            .borrow()
+            .access
+            .user_rate_limits
+            .clone();
+        let mut prev_cidr_limits = config_rx_rate_limits
+            .borrow()
+            .access
+            .cidr_rate_limits
+            .clone();
         loop {
             if config_rx_rate_limits.changed().await.is_err() {
                 break;

src/main.rs

@@ -8,6 +8,7 @@ mod crypto;
 #[cfg(unix)]
 mod daemon;
 mod error;
+mod healthcheck;
 mod ip_tracker;
 #[cfg(test)]
 #[path = "tests/ip_tracker_encapsulation_adversarial_tests.rs"]

src/proxy/direct_relay.rs

@@ -316,7 +316,9 @@ where
 
     stats.increment_user_connects(user);
     let _direct_connection_lease = stats.acquire_direct_connection_lease();
-    let traffic_lease = shared.traffic_limiter.acquire_lease(user, success.peer.ip());
+    let traffic_lease = shared
+        .traffic_limiter
+        .acquire_lease(user, success.peer.ip());
 
     let buffer_pool_trim = Arc::clone(&buffer_pool);
     let relay_activity_timeout = if shared.conntrack_pressure_active() {

src/proxy/relay.rs

@@ -289,17 +289,9 @@ impl<S> StatsIo<S> {
         let Some(started_at) = wait.started_at.take() else {
             return;
         };
-        let wait_ms = started_at
-            .elapsed()
-            .as_millis()
-            .min(u128::from(u64::MAX)) as u64;
+        let wait_ms = started_at.elapsed().as_millis().min(u128::from(u64::MAX)) as u64;
         if let Some(lease) = lease {
-            lease.observe_wait_ms(
-                direction,
-                wait.blocked_user,
-                wait.blocked_cidr,
-                wait_ms,
-            );
+            lease.observe_wait_ms(direction, wait.blocked_user, wait.blocked_cidr, wait_ms);
         }
         wait.blocked_user = false;
         wait.blocked_cidr = false;
@@ -340,8 +332,7 @@ impl<S> StatsIo<S> {
         while self.c2s_rate_debt_bytes > 0 {
             let consume = lease.try_consume(RateDirection::Up, self.c2s_rate_debt_bytes);
             if consume.granted > 0 {
-                self.c2s_rate_debt_bytes =
-                    self.c2s_rate_debt_bytes.saturating_sub(consume.granted);
+                self.c2s_rate_debt_bytes = self.c2s_rate_debt_bytes.saturating_sub(consume.granted);
                 continue;
             }
             Self::arm_wait(
@@ -647,7 +638,10 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
         match Pin::new(&mut this.inner).poll_write(cx, write_buf) {
             Poll::Ready(Ok(n)) => {
                 if reserved_bytes > n as u64 {
-                    refund_reserved_quota_bytes(this.user_stats.as_ref(), reserved_bytes - n as u64);
+                    refund_reserved_quota_bytes(
+                        this.user_stats.as_ref(),
+                        reserved_bytes - n as u64,
+                    );
                 }
                 if shaper_reserved_bytes > n as u64
                     && let Some(lease) = this.traffic_lease.as_ref()

src/proxy/traffic_limiter.rs

@@ -74,7 +74,8 @@ impl ScopeMetrics {
                 self.wait_up_ms_total.fetch_add(wait_ms, Ordering::Relaxed);
             }
             RateDirection::Down => {
-                self.wait_down_ms_total.fetch_add(wait_ms, Ordering::Relaxed);
+                self.wait_down_ms_total
+                    .fetch_add(wait_ms, Ordering::Relaxed);
             }
         }
     }
@@ -254,9 +255,7 @@ impl CidrDirectionBucket {
             let grant = if guaranteed_remaining > 0 {
                 requested.min(guaranteed_remaining).min(total_remaining)
             } else {
-                requested
-                    .min(total_remaining)
-                    .min(MAX_BORROW_CHUNK_BYTES)
+                requested.min(total_remaining).min(MAX_BORROW_CHUNK_BYTES)
             };
 
             if grant == 0 {
@@ -266,12 +265,7 @@ impl CidrDirectionBucket {
             let next_total = total_used.saturating_add(grant);
             if self
                 .used
-                .compare_exchange_weak(
-                    total_used,
-                    next_total,
-                    Ordering::Relaxed,
-                    Ordering::Relaxed,
-                )
+                .compare_exchange_weak(total_used, next_total, Ordering::Relaxed, Ordering::Relaxed)
                 .is_ok()
             {
                 user_state.used.fetch_add(grant, Ordering::Relaxed);
@@ -430,8 +424,14 @@ struct PolicySnapshot {
 impl PolicySnapshot {
     fn match_cidr(&self, ip: IpAddr) -> Option<&CidrRule> {
         match ip {
-            IpAddr::V4(_) => self.cidr_rules_v4.iter().find(|rule| rule.cidr.contains(ip)),
-            IpAddr::V6(_) => self.cidr_rules_v6.iter().find(|rule| rule.cidr.contains(ip)),
+            IpAddr::V4(_) => self
+                .cidr_rules_v4
+                .iter()
+                .find(|rule| rule.cidr.contains(ip)),
+            IpAddr::V6(_) => self
+                .cidr_rules_v6
+                .iter()
+                .find(|rule| rule.cidr.contains(ip)),
         }
     }
 }
@@ -535,7 +535,8 @@ impl TrafficLease {
         if let (Some(cidr_bucket), Some(cidr_user_share)) =
             (self.cidr_bucket.as_ref(), self.cidr_user_share.as_ref())
         {
-            let cidr_granted = cidr_bucket.try_consume_for_user(direction, cidr_user_share, granted);
+            let cidr_granted =
+                cidr_bucket.try_consume_for_user(direction, cidr_user_share, granted);
             if cidr_granted < granted
                 && let Some(user_bucket) = self.user_bucket.as_ref()
             {
@@ -693,7 +694,9 @@ impl TrafficLimiter {
                 .get_or_insert_with(user, || UserBucket::new(limit));
             bucket.set_rates(limit);
             bucket.active_leases.fetch_add(1, Ordering::Relaxed);
-            self.user_scope.active_leases.fetch_add(1, Ordering::Relaxed);
+            self.user_scope
+                .active_leases
+                .fetch_add(1, Ordering::Relaxed);
             user_bucket = Some(bucket);
         }
 
@@ -706,7 +709,9 @@ impl TrafficLimiter {
                 .get_or_insert_with(rule.key.as_str(), || CidrBucket::new(rule.limits));
             bucket.set_rates(rule.limits);
             bucket.active_leases.fetch_add(1, Ordering::Relaxed);
-            self.cidr_scope.active_leases.fetch_add(1, Ordering::Relaxed);
+            self.cidr_scope
+                .active_leases
+                .fetch_add(1, Ordering::Relaxed);
             let share = bucket.acquire_user_share(user);
             cidr_user_key = Some(user.to_string());
             cidr_user_share = Some(share);
@@ -784,7 +789,8 @@ impl TrafficLimiter {
 
         let policy = self.policy.load_full();
         self.user_buckets.retain(|user, bucket| {
-            bucket.active_leases.load(Ordering::Relaxed) > 0 || policy.user_limits.contains_key(user)
+            bucket.active_leases.load(Ordering::Relaxed) > 0
+                || policy.user_limits.contains_key(user)
         });
         self.cidr_buckets.retain(|cidr_key, bucket| {
             bucket.cleanup_idle_users();

src/tls_front/emulator.rs

@@ -66,9 +66,19 @@ fn ensure_payload_capacity(mut sizes: Vec<usize>, payload_len: usize) -> Vec<usi
 fn emulated_app_data_sizes(cached: &CachedTlsData) -> Vec<usize> {
     match cached.behavior_profile.source {
         TlsProfileSource::Raw | TlsProfileSource::Merged => {
-            if !cached.behavior_profile.app_data_record_sizes.is_empty() {
-                return cached.behavior_profile.app_data_record_sizes.clone();
-            }
+            return cached
+                .app_data_records_sizes
+                .first()
+                .copied()
+                .or_else(|| {
+                    cached
+                        .behavior_profile
+                        .app_data_record_sizes
+                        .first()
+                        .copied()
+                })
+                .map(|size| vec![size])
+                .unwrap_or_else(|| vec![cached.total_app_data_len.max(1024)]);
         }
         TlsProfileSource::Default | TlsProfileSource::Rustls => {}
     }
@@ -80,8 +90,8 @@ fn emulated_app_data_sizes(cached: &CachedTlsData) -> Vec<usize> {
     sizes
 }
 
-fn emulated_change_cipher_spec_count(cached: &CachedTlsData) -> usize {
-    usize::from(cached.behavior_profile.change_cipher_spec_count.max(1))
+fn emulated_change_cipher_spec_count(_cached: &CachedTlsData) -> usize {
+    1
 }
 
 fn emulated_ticket_record_sizes(
@@ -89,17 +99,20 @@ fn emulated_ticket_record_sizes(
     new_session_tickets: u8,
     rng: &SecureRandom,
 ) -> Vec<usize> {
-    let mut sizes = match cached.behavior_profile.source {
+    let target_count = usize::from(new_session_tickets.min(MAX_TICKET_RECORDS as u8));
+    if target_count == 0 {
+        return Vec::new();
+    }
+
+    let profiled_sizes = match cached.behavior_profile.source {
         TlsProfileSource::Raw | TlsProfileSource::Merged => {
-            cached.behavior_profile.ticket_record_sizes.clone()
+            cached.behavior_profile.ticket_record_sizes.as_slice()
         }
-        TlsProfileSource::Default | TlsProfileSource::Rustls => Vec::new(),
+        TlsProfileSource::Default | TlsProfileSource::Rustls => &[],
     };
 
-    let target_count = sizes
-        .len()
-        .max(usize::from(new_session_tickets.min(MAX_TICKET_RECORDS as u8)))
-        .min(MAX_TICKET_RECORDS);
+    let mut sizes = Vec::with_capacity(target_count);
+    sizes.extend(profiled_sizes.iter().copied().take(target_count));
 
     while sizes.len() < target_count {
         sizes.push(rng.range(48) + 48);
@@ -329,11 +342,11 @@ pub fn build_emulated_server_hello(
     let mut tickets = Vec::new();
     for ticket_len in emulated_ticket_record_sizes(cached, new_session_tickets, rng) {
         let mut rec = Vec::with_capacity(5 + ticket_len);
-            rec.push(TLS_RECORD_APPLICATION);
-            rec.extend_from_slice(&TLS_VERSION);
-            rec.extend_from_slice(&(ticket_len as u16).to_be_bytes());
-            rec.extend_from_slice(&rng.bytes(ticket_len));
-            tickets.extend_from_slice(&rec);
+        rec.push(TLS_RECORD_APPLICATION);
+        rec.extend_from_slice(&TLS_VERSION);
+        rec.extend_from_slice(&(ticket_len as u16).to_be_bytes());
+        rec.extend_from_slice(&rng.bytes(ticket_len));
+        tickets.extend_from_slice(&rec);
     }
 
     let mut response = Vec::with_capacity(
@@ -506,7 +519,7 @@ mod tests {
     }
 
     #[test]
-    fn test_build_emulated_server_hello_replays_tail_records_for_profiled_tls() {
+    fn test_build_emulated_server_hello_ignores_tail_records_for_profiled_tls() {
         let mut cached = make_cached(None);
         cached.app_data_records_sizes = vec![27, 3905, 537, 69];
         cached.total_app_data_len = 4538;
@@ -528,19 +541,11 @@ mod tests {
 
         let hello_len = u16::from_be_bytes([response[3], response[4]]) as usize;
         let ccs_start = 5 + hello_len;
-        let mut pos = ccs_start + 6;
-        let mut app_lengths = Vec::new();
-        while pos + 5 <= response.len() {
-            assert_eq!(response[pos], TLS_RECORD_APPLICATION);
-            let record_len = u16::from_be_bytes([response[pos + 3], response[pos + 4]]) as usize;
-            app_lengths.push(record_len);
-            pos += 5 + record_len;
-        }
-
-        assert_eq!(app_lengths.len(), 4);
-        assert_eq!(app_lengths[0], 64);
-        assert_eq!(app_lengths[3], 69);
-        assert!(app_lengths[1] >= 64);
-        assert!(app_lengths[2] >= 64);
+        let app_start = ccs_start + 6;
+        let app_len =
+            u16::from_be_bytes([response[app_start + 3], response[app_start + 4]]) as usize;
+        assert_eq!(response[app_start], TLS_RECORD_APPLICATION);
+        assert_eq!(app_len, 64);
+        assert_eq!(app_start + 5 + app_len, response.len());
     }
 }

src/tls_front/fetcher.rs

@@ -1,6 +1,7 @@
 #![allow(clippy::too_many_arguments)]
 
 use dashmap::DashMap;
+use std::net::SocketAddr;
 use std::sync::Arc;
 use std::sync::OnceLock;
 use std::time::{Duration, Instant};
@@ -793,6 +794,51 @@ async fn connect_tcp_with_upstream(
     ))
 }
 
+fn socket_addrs_from_upstream_stream(
+    stream: &UpstreamStream,
+) -> (Option<SocketAddr>, Option<SocketAddr>) {
+    match stream {
+        UpstreamStream::Tcp(tcp) => (tcp.local_addr().ok(), tcp.peer_addr().ok()),
+        UpstreamStream::Shadowsocks(_) => (None, None),
+    }
+}
+
+fn build_tls_fetch_proxy_header(
+    proxy_protocol: u8,
+    src_addr: Option<SocketAddr>,
+    dst_addr: Option<SocketAddr>,
+) -> Option<Vec<u8>> {
+    match proxy_protocol {
+        0 => None,
+        2 => {
+            let header = match (src_addr, dst_addr) {
+                (Some(src @ SocketAddr::V4(_)), Some(dst @ SocketAddr::V4(_)))
+                | (Some(src @ SocketAddr::V6(_)), Some(dst @ SocketAddr::V6(_))) => {
+                    ProxyProtocolV2Builder::new().with_addrs(src, dst).build()
+                }
+                _ => ProxyProtocolV2Builder::new().build(),
+            };
+            Some(header)
+        }
+        _ => {
+            let header = match (src_addr, dst_addr) {
+                (Some(SocketAddr::V4(src)), Some(SocketAddr::V4(dst))) => {
+                    ProxyProtocolV1Builder::new()
+                        .tcp4(src.into(), dst.into())
+                        .build()
+                }
+                (Some(SocketAddr::V6(src)), Some(SocketAddr::V6(dst))) => {
+                    ProxyProtocolV1Builder::new()
+                        .tcp6(src.into(), dst.into())
+                        .build()
+                }
+                _ => ProxyProtocolV1Builder::new().build(),
+            };
+            Some(header)
+        }
+    }
+}
+
 fn encode_tls13_certificate_message(cert_chain_der: &[Vec<u8>]) -> Option<Vec<u8>> {
     if cert_chain_der.is_empty() {
         return None;
@@ -824,7 +870,7 @@ async fn fetch_via_raw_tls_stream<S>(
     mut stream: S,
     sni: &str,
     connect_timeout: Duration,
-    proxy_protocol: u8,
+    proxy_header: Option<Vec<u8>>,
     profile: TlsFetchProfile,
     grease_enabled: bool,
     deterministic: bool,
@@ -835,11 +881,7 @@ where
     let rng = SecureRandom::new();
     let client_hello = build_client_hello(sni, &rng, profile, grease_enabled, deterministic);
     timeout(connect_timeout, async {
-        if proxy_protocol > 0 {
-            let header = match proxy_protocol {
-                2 => ProxyProtocolV2Builder::new().build(),
-                _ => ProxyProtocolV1Builder::new().build(),
-            };
+        if let Some(header) = proxy_header.as_ref() {
             stream.write_all(&header).await?;
         }
         stream.write_all(&client_hello).await?;
@@ -921,11 +963,12 @@ async fn fetch_via_raw_tls(
                     sock = %sock_path,
                     "Raw TLS fetch using mask unix socket"
                 );
+                let proxy_header = build_tls_fetch_proxy_header(proxy_protocol, None, None);
                 return fetch_via_raw_tls_stream(
                     stream,
                     sni,
                     connect_timeout,
-                    proxy_protocol,
+                    proxy_header,
                     profile,
                     grease_enabled,
                     deterministic,
@@ -956,11 +999,13 @@ async fn fetch_via_raw_tls(
     let stream =
         connect_tcp_with_upstream(host, port, connect_timeout, upstream, scope, strict_route)
             .await?;
+    let (src_addr, dst_addr) = socket_addrs_from_upstream_stream(&stream);
+    let proxy_header = build_tls_fetch_proxy_header(proxy_protocol, src_addr, dst_addr);
     fetch_via_raw_tls_stream(
         stream,
         sni,
         connect_timeout,
-        proxy_protocol,
+        proxy_header,
         profile,
         grease_enabled,
         deterministic,
@@ -972,17 +1017,13 @@ async fn fetch_via_rustls_stream<S>(
     mut stream: S,
     host: &str,
     sni: &str,
-    proxy_protocol: u8,
+    proxy_header: Option<Vec<u8>>,
 ) -> Result<TlsFetchResult>
 where
     S: AsyncRead + AsyncWrite + Unpin,
 {
     // rustls handshake path for certificate and basic negotiated metadata.
-    if proxy_protocol > 0 {
-        let header = match proxy_protocol {
-            2 => ProxyProtocolV2Builder::new().build(),
-            _ => ProxyProtocolV1Builder::new().build(),
-        };
+    if let Some(header) = proxy_header.as_ref() {
         stream.write_all(&header).await?;
         stream.flush().await?;
     }
@@ -1082,7 +1123,8 @@ async fn fetch_via_rustls(
                     sock = %sock_path,
                     "Rustls fetch using mask unix socket"
                 );
-                return fetch_via_rustls_stream(stream, host, sni, proxy_protocol).await;
+                let proxy_header = build_tls_fetch_proxy_header(proxy_protocol, None, None);
+                return fetch_via_rustls_stream(stream, host, sni, proxy_header).await;
             }
             Ok(Err(e)) => {
                 warn!(
@@ -1108,7 +1150,9 @@ async fn fetch_via_rustls(
     let stream =
         connect_tcp_with_upstream(host, port, connect_timeout, upstream, scope, strict_route)
             .await?;
-    fetch_via_rustls_stream(stream, host, sni, proxy_protocol).await
+    let (src_addr, dst_addr) = socket_addrs_from_upstream_stream(&stream);
+    let proxy_header = build_tls_fetch_proxy_header(proxy_protocol, src_addr, dst_addr);
+    fetch_via_rustls_stream(stream, host, sni, proxy_header).await
 }
 
 /// Fetch real TLS metadata with an adaptive multi-profile strategy.
@@ -1278,11 +1322,13 @@ pub async fn fetch_real_tls(
 
 #[cfg(test)]
 mod tests {
+    use std::net::SocketAddr;
     use std::time::{Duration, Instant};
 
     use super::{
-        ProfileCacheValue, TlsFetchStrategy, build_client_hello, derive_behavior_profile,
-        encode_tls13_certificate_message, order_profiles, profile_cache, profile_cache_key,
+        ProfileCacheValue, TlsFetchStrategy, build_client_hello, build_tls_fetch_proxy_header,
+        derive_behavior_profile, encode_tls13_certificate_message, order_profiles, profile_cache,
+        profile_cache_key,
     };
     use crate::config::TlsFetchProfile;
     use crate::crypto::SecureRandom;
@@ -1423,4 +1469,48 @@ mod tests {
 
         assert_eq!(first, second);
     }
+
+    #[test]
+    fn test_build_tls_fetch_proxy_header_v2_with_tcp_addrs() {
+        let src: SocketAddr = "198.51.100.10:42000".parse().expect("valid src");
+        let dst: SocketAddr = "203.0.113.20:443".parse().expect("valid dst");
+        let header = build_tls_fetch_proxy_header(2, Some(src), Some(dst)).expect("header");
+
+        assert_eq!(
+            &header[..12],
+            &[
+                0x0d, 0x0a, 0x0d, 0x0a, 0x00, 0x0d, 0x0a, 0x51, 0x55, 0x49, 0x54, 0x0a
+            ]
+        );
+        assert_eq!(header[12], 0x21);
+        assert_eq!(header[13], 0x11);
+        assert_eq!(u16::from_be_bytes([header[14], header[15]]), 12);
+        assert_eq!(&header[16..20], &[198, 51, 100, 10]);
+        assert_eq!(&header[20..24], &[203, 0, 113, 20]);
+        assert_eq!(u16::from_be_bytes([header[24], header[25]]), 42000);
+        assert_eq!(u16::from_be_bytes([header[26], header[27]]), 443);
+    }
+
+    #[test]
+    fn test_build_tls_fetch_proxy_header_v2_mixed_family_falls_back_to_local_command() {
+        let src: SocketAddr = "198.51.100.10:42000".parse().expect("valid src");
+        let dst: SocketAddr = "[2001:db8::20]:443".parse().expect("valid dst");
+        let header = build_tls_fetch_proxy_header(2, Some(src), Some(dst)).expect("header");
+
+        assert_eq!(header[12], 0x20);
+        assert_eq!(header[13], 0x00);
+        assert_eq!(u16::from_be_bytes([header[14], header[15]]), 0);
+    }
+
+    #[test]
+    fn test_build_tls_fetch_proxy_header_v1_with_tcp_addrs() {
+        let src: SocketAddr = "198.51.100.10:42000".parse().expect("valid src");
+        let dst: SocketAddr = "203.0.113.20:443".parse().expect("valid dst");
+        let header = build_tls_fetch_proxy_header(1, Some(src), Some(dst)).expect("header");
+
+        assert_eq!(
+            header,
+            b"PROXY TCP4 198.51.100.10 203.0.113.20 42000 443\r\n"
+        );
+    }
 }

src/tls_front/tests/emulator_profile_fidelity_security_tests.rs

@@ -52,7 +52,7 @@ fn record_lengths_by_type(response: &[u8], wanted_type: u8) -> Vec<usize> {
 }
 
 #[test]
-fn emulated_server_hello_replays_profile_change_cipher_spec_count() {
+fn emulated_server_hello_keeps_single_change_cipher_spec_for_client_compatibility() {
     let cached = make_cached();
     let rng = SecureRandom::new();
 
@@ -69,12 +69,12 @@ fn emulated_server_hello_replays_profile_change_cipher_spec_count() {
 
     assert_eq!(response[0], TLS_RECORD_HANDSHAKE);
     let ccs_records = record_lengths_by_type(&response, TLS_RECORD_CHANGE_CIPHER);
-    assert_eq!(ccs_records.len(), 2);
+    assert_eq!(ccs_records.len(), 1);
     assert!(ccs_records.iter().all(|len| *len == 1));
 }
 
 #[test]
-fn emulated_server_hello_replays_profile_ticket_tail_lengths() {
+fn emulated_server_hello_does_not_emit_profile_ticket_tail_when_disabled() {
     let cached = make_cached();
     let rng = SecureRandom::new();
 
@@ -90,6 +90,25 @@ fn emulated_server_hello_replays_profile_ticket_tail_lengths() {
     );
 
     let app_records = record_lengths_by_type(&response, TLS_RECORD_APPLICATION);
-    assert!(app_records.len() >= 4);
-    assert_eq!(&app_records[app_records.len() - 2..], &[220, 180]);
+    assert_eq!(app_records, vec![1200]);
+}
+
+#[test]
+fn emulated_server_hello_uses_profile_ticket_lengths_when_enabled() {
+    let cached = make_cached();
+    let rng = SecureRandom::new();
+
+    let response = build_emulated_server_hello(
+        b"secret",
+        &[0x91; 32],
+        &[0x92; 16],
+        &cached,
+        false,
+        &rng,
+        None,
+        2,
+    );
+
+    let app_records = record_lengths_by_type(&response, TLS_RECORD_APPLICATION);
+    assert_eq!(app_records, vec![1200, 220, 180]);
 }

src/transport/middle_proxy/fairness/mod.rs

@@ -7,7 +7,7 @@ mod model;
 mod pressure;
 mod scheduler;
 
-pub(crate) use model::{
-    AdmissionDecision, DispatchAction, DispatchFeedback, PressureState, SchedulerDecision,
-};
+#[cfg(test)]
+pub(crate) use model::PressureState;
+pub(crate) use model::{AdmissionDecision, DispatchAction, DispatchFeedback, SchedulerDecision};
 pub(crate) use scheduler::{WorkerFairnessConfig, WorkerFairnessSnapshot, WorkerFairnessState};

src/transport/middle_proxy/fairness/pressure.rs

@@ -136,8 +136,8 @@ impl PressureEvaluator {
         let queue_ratio_pct = if max_total_queued_bytes == 0 {
             100
         } else {
-            ((signals.total_queued_bytes.saturating_mul(100)) / max_total_queued_bytes)
-                .min(100) as u8
+            ((signals.total_queued_bytes.saturating_mul(100)) / max_total_queued_bytes).min(100)
+                as u8
         };
 
         let standing_ratio_pct = if signals.active_flows == 0 {

src/transport/middle_proxy/fairness/scheduler.rs

@@ -166,7 +166,8 @@ impl WorkerFairnessState {
             return AdmissionDecision::RejectSaturated;
         }
 
-        if self.total_queued_bytes.saturating_add(frame_bytes) > self.config.max_total_queued_bytes {
+        if self.total_queued_bytes.saturating_add(frame_bytes) > self.config.max_total_queued_bytes
+        {
             self.pressure
                 .note_admission_reject(now, &self.config.pressure);
             self.enqueue_rejects = self.enqueue_rejects.saturating_add(1);
@@ -211,7 +212,8 @@ impl WorkerFairnessState {
                 .expect("flow inserted must be retrievable")
         };
 
-        if entry.fairness.pending_bytes.saturating_add(frame_bytes) > self.config.max_flow_queued_bytes
+        if entry.fairness.pending_bytes.saturating_add(frame_bytes)
+            > self.config.max_flow_queued_bytes
         {
             self.pressure
                 .note_admission_reject(now, &self.config.pressure);
@@ -237,7 +239,8 @@ impl WorkerFairnessState {
         entry.queue.push_back(frame);
 
         self.total_queued_bytes = self.total_queued_bytes.saturating_add(frame_bytes);
-        self.bucket_queued_bytes[bucket_id] = self.bucket_queued_bytes[bucket_id].saturating_add(frame_bytes);
+        self.bucket_queued_bytes[bucket_id] =
+            self.bucket_queued_bytes[bucket_id].saturating_add(frame_bytes);
 
         if !entry.fairness.in_active_ring {
             entry.fairness.in_active_ring = true;
@@ -277,23 +280,31 @@ impl WorkerFairnessState {
 
                 Self::classify_flow(&self.config, pressure_state, now, &mut flow.fairness);
 
-                let quantum = Self::effective_quantum_bytes(&self.config, pressure_state, &flow.fairness);
-                flow.fairness.deficit_bytes =
-                    flow.fairness.deficit_bytes.saturating_add(i64::from(quantum));
+                let quantum =
+                    Self::effective_quantum_bytes(&self.config, pressure_state, &flow.fairness);
+                flow.fairness.deficit_bytes = flow
+                    .fairness
+                    .deficit_bytes
+                    .saturating_add(i64::from(quantum));
                 self.deficit_grants = self.deficit_grants.saturating_add(1);
 
                 let front_len = flow.queue.front().map_or(0, |front| front.queued_bytes());
                 if flow.fairness.deficit_bytes < front_len as i64 {
-                    flow.fairness.consecutive_skips = flow.fairness.consecutive_skips.saturating_add(1);
+                    flow.fairness.consecutive_skips =
+                        flow.fairness.consecutive_skips.saturating_add(1);
                     self.deficit_skips = self.deficit_skips.saturating_add(1);
                     requeue_active = true;
                 } else if let Some(frame) = flow.queue.pop_front() {
                     drained_bytes = frame.queued_bytes();
-                    flow.fairness.pending_bytes = flow.fairness.pending_bytes.saturating_sub(drained_bytes);
-                    flow.fairness.deficit_bytes =
-                        flow.fairness.deficit_bytes.saturating_sub(drained_bytes as i64);
+                    flow.fairness.pending_bytes =
+                        flow.fairness.pending_bytes.saturating_sub(drained_bytes);
+                    flow.fairness.deficit_bytes = flow
+                        .fairness
+                        .deficit_bytes
+                        .saturating_sub(drained_bytes as i64);
                     flow.fairness.consecutive_skips = 0;
-                    flow.fairness.queue_started_at = flow.queue.front().map(|front| front.enqueued_at);
+                    flow.fairness.queue_started_at =
+                        flow.queue.front().map(|front| front.enqueued_at);
                     requeue_active = !flow.queue.is_empty();
                     if !requeue_active {
                         flow.fairness.scheduler_state = FlowSchedulerState::Idle;
@@ -359,7 +370,8 @@ impl WorkerFairnessState {
                     return DispatchAction::Continue;
                 };
 
-                flow.fairness.consecutive_stalls = flow.fairness.consecutive_stalls.saturating_add(1);
+                flow.fairness.consecutive_stalls =
+                    flow.fairness.consecutive_stalls.saturating_add(1);
                 flow.fairness.scheduler_state = FlowSchedulerState::Backpressured;
                 flow.fairness.pressure_class = FlowPressureClass::Backpressured;
 
@@ -376,10 +388,10 @@ impl WorkerFairnessState {
                 } else {
                     let frame_bytes = candidate.frame.queued_bytes();
                     flow.queue.push_front(candidate.frame);
-                    flow.fairness.pending_bytes = flow.fairness.pending_bytes.saturating_add(frame_bytes);
-                    if flow.fairness.queue_started_at.is_none() {
-                        flow.fairness.queue_started_at = Some(now);
-                    }
+                    flow.fairness.pending_bytes =
+                        flow.fairness.pending_bytes.saturating_add(frame_bytes);
+                    flow.fairness.queue_started_at =
+                        flow.queue.front().map(|front| front.enqueued_at);
                     self.total_queued_bytes = self.total_queued_bytes.saturating_add(frame_bytes);
                     self.bucket_queued_bytes[flow.fairness.bucket_id] = self.bucket_queued_bytes
                         [flow.fairness.bucket_id]
@@ -390,7 +402,8 @@ impl WorkerFairnessState {
                     }
                 }
 
-                if flow.fairness.consecutive_stalls >= self.config.max_consecutive_stalls_before_close
+                if flow.fairness.consecutive_stalls
+                    >= self.config.max_consecutive_stalls_before_close
                     && self.pressure.state() == PressureState::Saturated
                 {
                     self.remove_flow(conn_id);
@@ -414,18 +427,16 @@ impl WorkerFairnessState {
             return;
         };
 
-        self.bucket_active_flows[entry.fairness.bucket_id] = self.bucket_active_flows
-            [entry.fairness.bucket_id]
-            .saturating_sub(1);
+        self.bucket_active_flows[entry.fairness.bucket_id] =
+            self.bucket_active_flows[entry.fairness.bucket_id].saturating_sub(1);
 
         let mut reclaimed = 0u64;
         for frame in entry.queue {
             reclaimed = reclaimed.saturating_add(frame.queued_bytes());
         }
         self.total_queued_bytes = self.total_queued_bytes.saturating_sub(reclaimed);
-        self.bucket_queued_bytes[entry.fairness.bucket_id] = self.bucket_queued_bytes
-            [entry.fairness.bucket_id]
-            .saturating_sub(reclaimed);
+        self.bucket_queued_bytes[entry.fairness.bucket_id] =
+            self.bucket_queued_bytes[entry.fairness.bucket_id].saturating_sub(reclaimed);
     }
 
     fn evaluate_pressure(&mut self, now: Instant, force: bool) {

src/transport/middle_proxy/mod.rs

@@ -3,6 +3,9 @@
 mod codec;
 mod config_updater;
 mod fairness;
+#[cfg(test)]
+#[path = "tests/fairness_security_tests.rs"]
+mod fairness_security_tests;
 mod handshake;
 mod health;
 #[cfg(test)]
@@ -31,9 +34,6 @@ mod pool_writer;
 #[cfg(test)]
 #[path = "tests/pool_writer_security_tests.rs"]
 mod pool_writer_security_tests;
-#[cfg(test)]
-#[path = "tests/fairness_security_tests.rs"]
-mod fairness_security_tests;
 mod reader;
 mod registry;
 mod rotation;

src/transport/middle_proxy/reader.rs

@@ -118,20 +118,22 @@ fn apply_fairness_metrics_delta(
     stats.set_me_fair_backpressured_flows_gauge(current.backpressured_flows as u64);
     stats.set_me_fair_pressure_state_gauge(current.pressure_state.as_u8() as u64);
     stats.add_me_fair_scheduler_rounds_total(
-        current.scheduler_rounds.saturating_sub(prev.scheduler_rounds),
+        current
+            .scheduler_rounds
+            .saturating_sub(prev.scheduler_rounds),
     );
     stats.add_me_fair_deficit_grants_total(
         current.deficit_grants.saturating_sub(prev.deficit_grants),
     );
-    stats.add_me_fair_deficit_skips_total(
-        current.deficit_skips.saturating_sub(prev.deficit_skips),
-    );
+    stats.add_me_fair_deficit_skips_total(current.deficit_skips.saturating_sub(prev.deficit_skips));
     stats.add_me_fair_enqueue_rejects_total(
         current.enqueue_rejects.saturating_sub(prev.enqueue_rejects),
     );
     stats.add_me_fair_shed_drops_total(current.shed_drops.saturating_sub(prev.shed_drops));
     stats.add_me_fair_penalties_total(
-        current.fairness_penalties.saturating_sub(prev.fairness_penalties),
+        current
+            .fairness_penalties
+            .saturating_sub(prev.fairness_penalties),
     );
     stats.add_me_fair_downstream_stalls_total(
         current

src/transport/middle_proxy/tests/fairness_security_tests.rs

@@ -175,7 +175,8 @@ fn fairness_randomized_sequence_preserves_memory_bounds() {
             } else {
                 DispatchFeedback::QueueFull
             };
-            let _ = fairness.apply_dispatch_feedback(candidate.frame.conn_id, candidate, feedback, now);
+            let _ =
+                fairness.apply_dispatch_feedback(candidate.frame.conn_id, candidate, feedback, now);
         }
 
         let snapshot = fairness.snapshot();

src/transport/upstream.rs

@@ -279,6 +279,12 @@ pub struct UpstreamApiSummarySnapshot {
     pub shadowsocks_total: usize,
 }
 
+#[derive(Debug, Clone, Copy, Default)]
+pub struct UpstreamApiHealthSummary {
+    pub configured_total: usize,
+    pub healthy_total: usize,
+}
+
 #[derive(Debug, Clone)]
 pub struct UpstreamApiSnapshot {
     pub summary: UpstreamApiSummarySnapshot,
@@ -444,6 +450,20 @@ impl UpstreamManager {
         Some(UpstreamApiSnapshot { summary, upstreams })
     }
 
+    pub async fn api_health_summary(&self) -> UpstreamApiHealthSummary {
+        let guard = self.upstreams.read().await;
+        let mut summary = UpstreamApiHealthSummary {
+            configured_total: guard.len(),
+            healthy_total: 0,
+        };
+        for upstream in guard.iter() {
+            if upstream.healthy {
+                summary.healthy_total += 1;
+            }
+        }
+        summary
+    }
+
     fn describe_upstream(upstream_type: &UpstreamType) -> (UpstreamRouteKind, String) {
         match upstream_type {
             UpstreamType::Direct { .. } => (UpstreamRouteKind::Direct, "direct".to_string()),