ME Writers Anti-stuck + Quarantine fixes + ME Writers Advanced Cleanup + Authoritative Teardown + Orphan Watchdog + Force-Close Safery Policy: merge pull request #504 from telemt/flow-me

ME Writers Anti-stuck + Quarantine fixes + ME Writers Advanced Cleanup + Authoritative Teardown + Orphan Watchdog + Force-Close Safery Policy

.dockerignore

@@ -1,8 +0,0 @@
-.git
-.github
-target
-.kilocode
-cache
-tlsfront
-*.tar
-*.tar.gz

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "telemt"
-version = "3.3.25"
+version = "3.3.26"
 edition = "2024"
 
 [dependencies]
@@ -26,7 +26,6 @@ zeroize = { version = "1.8", features = ["derive"] }
 # Network
 socket2 = { version = "0.5", features = ["all"] }
 nix = { version = "0.28", default-features = false, features = ["net"] }
-shadowsocks = { version = "1.24", features = ["aead-cipher-2022"] }
 
 # Serialization
 serde = { version = "1.0", features = ["derive"] }
@@ -41,6 +40,7 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 parking_lot = "0.12"
 dashmap = "5.5"
+arc-swap = "1.7"
 lru = "0.16"
 rand = "0.9"
 chrono = { version = "0.4", features = ["serde"] }

docs/API.md

@@ -497,14 +497,13 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `direct_total` | `usize` | Direct-route upstream entries. |
 | `socks4_total` | `usize` | SOCKS4 upstream entries. |
 | `socks5_total` | `usize` | SOCKS5 upstream entries. |
-| `shadowsocks_total` | `usize` | Shadowsocks upstream entries. |
 
 #### `RuntimeUpstreamQualityUpstreamData`
 | Field | Type | Description |
 | --- | --- | --- |
 | `upstream_id` | `usize` | Runtime upstream index. |
-| `route_kind` | `string` | `direct`, `socks4`, `socks5`, `shadowsocks`. |
-| `address` | `string` | Upstream address (`direct` literal for direct route kind, `host:port` only for proxied upstreams). |
+| `route_kind` | `string` | `direct`, `socks4`, `socks5`. |
+| `address` | `string` | Upstream address (`direct` literal for direct route kind). |
 | `weight` | `u16` | Selection weight. |
 | `scopes` | `string` | Configured scope selector. |
 | `healthy` | `bool` | Current health flag. |
@@ -758,14 +757,13 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `direct_total` | `usize` | Number of direct upstream entries. |
 | `socks4_total` | `usize` | Number of SOCKS4 upstream entries. |
 | `socks5_total` | `usize` | Number of SOCKS5 upstream entries. |
-| `shadowsocks_total` | `usize` | Number of Shadowsocks upstream entries. |
 
 #### `UpstreamStatus`
 | Field | Type | Description |
 | --- | --- | --- |
 | `upstream_id` | `usize` | Runtime upstream index. |
-| `route_kind` | `string` | Upstream route kind: `direct`, `socks4`, `socks5`, `shadowsocks`. |
-| `address` | `string` | Upstream address (`direct` for direct route kind, `host:port` for Shadowsocks). Authentication fields are intentionally omitted. |
+| `route_kind` | `string` | Upstream route kind: `direct`, `socks4`, `socks5`. |
+| `address` | `string` | Upstream address (`direct` for direct route kind). Authentication fields are intentionally omitted. |
 | `weight` | `u16` | Selection weight. |
 | `scopes` | `string` | Configured scope selector string. |
 | `healthy` | `bool` | Current health flag. |

docs/FAQ.en.md

@@ -120,17 +120,3 @@ password = "pass"          # Password for Auth on SOCKS-server
 weight = 1                 # Set Weight for Scenarios
 enabled = true
 ```
-
-#### Shadowsocks as Upstream
-Requires `use_middle_proxy = false`.
-
-```toml
-[general]
-use_middle_proxy = false
-
-[[upstreams]]
-type = "shadowsocks"
-url = "ss://2022-blake3-aes-256-gcm:BASE64_KEY@1.2.3.4:8388"
-weight = 1
-enabled = true
-```

docs/FAQ.ru.md

@@ -121,16 +121,3 @@ weight = 1                 # Set Weight for Scenarios
 enabled = true
 ```
 
-#### Shadowsocks как Upstream
-Требует `use_middle_proxy = false`.
-
-```toml
-[general]
-use_middle_proxy = false
-
-[[upstreams]]
-type = "shadowsocks"
-url = "ss://2022-blake3-aes-256-gcm:BASE64_KEY@1.2.3.4:8388"
-weight = 1
-enabled = true
-```

docs/TUNING.de.md

@@ -82,7 +82,7 @@ Die unten angegebenen `Default`-Werte sind Code-Defaults (bei fehlendem Schlüss
 
 | Feld | Gilt für | Typ | Pflicht | Default | Bedeutung |
 |---|---|---|---|---|---|
-| `[[upstreams]].type` | alle Upstreams | `"direct" \| "socks4" \| "socks5" \| "shadowsocks"` | ja | n/a | Upstream-Transporttyp. |
+| `[[upstreams]].type` | alle Upstreams | `"direct" \| "socks4" \| "socks5"` | ja | n/a | Upstream-Transporttyp. |
 | `[[upstreams]].weight` | alle Upstreams | `u16` | nein | `1` | Basisgewicht für weighted-random Auswahl. |
 | `[[upstreams]].enabled` | alle Upstreams | `bool` | nein | `true` | Deaktivierte Einträge werden beim Start ignoriert. |
 | `[[upstreams]].scopes` | alle Upstreams | `String` | nein | `""` | Komma-separierte Scope-Tags für Request-Routing. |
@@ -95,8 +95,6 @@ Die unten angegebenen `Default`-Werte sind Code-Defaults (bei fehlendem Schlüss
 | `interface` | `socks5` | `Option<String>` | nein | `null` | Wird nur genutzt, wenn `address` als `ip:port` angegeben ist. |
 | `username` | `socks5` | `Option<String>` | nein | `null` | SOCKS5 Benutzername. |
 | `password` | `socks5` | `Option<String>` | nein | `null` | SOCKS5 Passwort. |
-| `url` | `shadowsocks` | `String` | ja | n/a | Shadowsocks-SIP002-URL (`ss://...`). In Runtime-APIs wird nur `host:port` offengelegt. |
-| `interface` | `shadowsocks` | `Option<String>` | nein | `null` | Optionales ausgehendes Bind-Interface oder lokale Literal-IP. |
 
 ### Runtime-Regeln (wichtig)
 
@@ -117,7 +115,6 @@ Die unten angegebenen `Default`-Werte sind Code-Defaults (bei fehlendem Schlüss
 8. Im ME-Modus wird der gewählte Upstream auch für den ME-TCP-Dial-Pfad verwendet.
 9. Im ME-Modus ist bei `direct` mit bind/interface die STUN-Reflection bind-aware für KDF-Adressmaterial.
 10. Im ME-Modus werden bei SOCKS-Upstream `BND.ADDR/BND.PORT` für KDF verwendet, wenn gültig/öffentlich und gleiche IP-Familie.
-11. `shadowsocks`-Upstreams erfordern `general.use_middle_proxy = false`. Mit aktiviertem ME-Modus schlägt das Laden der Config sofort fehl.
 
 ## Upstream-Konfigurationsbeispiele
 
@@ -153,20 +150,7 @@ weight = 2
 enabled = true
 ```
 
-### Beispiel 4: Shadowsocks-Upstream
-
-```toml
-[general]
-use_middle_proxy = false
-
-[[upstreams]]
-type = "shadowsocks"
-url = "ss://2022-blake3-aes-256-gcm:BASE64_KEY@198.51.100.50:8388"
-weight = 2
-enabled = true
-```
-
-### Beispiel 5: Gemischte Upstreams mit Scopes
+### Beispiel 4: Gemischte Upstreams mit Scopes
 
 ```toml
 [[upstreams]]

docs/TUNING.en.md

@@ -82,7 +82,7 @@ Defaults below are code defaults (used when a key is omitted), not necessarily v
 
 | Field | Applies to | Type | Required | Default | Meaning |
 |---|---|---|---|---|---|
-| `[[upstreams]].type` | all upstreams | `"direct" \| "socks4" \| "socks5" \| "shadowsocks"` | yes | n/a | Upstream transport type. |
+| `[[upstreams]].type` | all upstreams | `"direct" \| "socks4" \| "socks5"` | yes | n/a | Upstream transport type. |
 | `[[upstreams]].weight` | all upstreams | `u16` | no | `1` | Base weight for weighted-random selection. |
 | `[[upstreams]].enabled` | all upstreams | `bool` | no | `true` | Disabled entries are ignored at startup. |
 | `[[upstreams]].scopes` | all upstreams | `String` | no | `""` | Comma-separated scope tags for request-level routing. |
@@ -95,8 +95,6 @@ Defaults below are code defaults (used when a key is omitted), not necessarily v
 | `interface` | `socks5` | `Option<String>` | no | `null` | Used only for SOCKS server `ip:port` dial path. |
 | `username` | `socks5` | `Option<String>` | no | `null` | SOCKS5 username auth. |
 | `password` | `socks5` | `Option<String>` | no | `null` | SOCKS5 password auth. |
-| `url` | `shadowsocks` | `String` | yes | n/a | Shadowsocks SIP002 URL (`ss://...`). Only `host:port` is exposed in runtime APIs. |
-| `interface` | `shadowsocks` | `Option<String>` | no | `null` | Optional outgoing bind interface or literal local IP. |
 
 ### Runtime rules (important)
 
@@ -117,7 +115,6 @@ Defaults below are code defaults (used when a key is omitted), not necessarily v
 8. In ME mode, the selected upstream is also used for ME TCP dial path.
 9. In ME mode for `direct` upstream with bind/interface, STUN reflection logic is bind-aware for KDF source material.
 10. In ME mode for SOCKS upstream, SOCKS `BND.ADDR/BND.PORT` is used for KDF when it is valid/public for the same family.
-11. `shadowsocks` upstreams require `general.use_middle_proxy = false`. Config load fails fast if ME mode is enabled.
 
 ## Upstream Configuration Examples
 
@@ -153,20 +150,7 @@ weight = 2
 enabled = true
 ```
 
-### Example 4: Shadowsocks upstream
-
-```toml
-[general]
-use_middle_proxy = false
-
-[[upstreams]]
-type = "shadowsocks"
-url = "ss://2022-blake3-aes-256-gcm:BASE64_KEY@198.51.100.50:8388"
-weight = 2
-enabled = true
-```
-
-### Example 5: Mixed upstreams with scopes
+### Example 4: Mixed upstreams with scopes
 
 ```toml
 [[upstreams]]

docs/TUNING.ru.md

@@ -82,7 +82,7 @@
 
 | Поле | Применимость | Тип | Обязательно | Default | Назначение |
 |---|---|---|---|---|---|
-| `[[upstreams]].type` | все upstream | `"direct" \| "socks4" \| "socks5" \| "shadowsocks"` | да | n/a | Тип upstream транспорта. |
+| `[[upstreams]].type` | все upstream | `"direct" \| "socks4" \| "socks5"` | да | n/a | Тип upstream транспорта. |
 | `[[upstreams]].weight` | все upstream | `u16` | нет | `1` | Базовый вес в weighted-random выборе. |
 | `[[upstreams]].enabled` | все upstream | `bool` | нет | `true` | Выключенные записи игнорируются на старте. |
 | `[[upstreams]].scopes` | все upstream | `String` | нет | `""` | Список scope-токенов через запятую для маршрутизации. |
@@ -95,8 +95,6 @@
 | `interface` | `socks5` | `Option<String>` | нет | `null` | Используется только если `address` задан как `ip:port`. |
 | `username` | `socks5` | `Option<String>` | нет | `null` | Логин SOCKS5 auth. |
 | `password` | `socks5` | `Option<String>` | нет | `null` | Пароль SOCKS5 auth. |
-| `url` | `shadowsocks` | `String` | да | n/a | Shadowsocks SIP002 URL (`ss://...`). В runtime API раскрывается только `host:port`. |
-| `interface` | `shadowsocks` | `Option<String>` | нет | `null` | Необязательный исходящий bind-интерфейс или literal локальный IP. |
 
 ### Runtime-правила
 
@@ -117,7 +115,6 @@
 8. В ME-режиме выбранный upstream также используется для ME TCP dial path.
 9. В ME-режиме для `direct` upstream с bind/interface STUN-рефлексия выполняется bind-aware для KDF материала.
 10. В ME-режиме для SOCKS upstream используются `BND.ADDR/BND.PORT` для KDF, если адрес валиден/публичен и соответствует IP family.
-11. `shadowsocks` upstream требует `general.use_middle_proxy = false`. При включенном ME-режиме конфиг отклоняется при загрузке.
 
 ## Примеры конфигурации Upstreams
 
@@ -153,20 +150,7 @@ weight = 2
 enabled = true
 ```
 
-### Пример 4: Shadowsocks upstream
-
-```toml
-[general]
-use_middle_proxy = false
-
-[[upstreams]]
-type = "shadowsocks"
-url = "ss://2022-blake3-aes-256-gcm:BASE64_KEY@198.51.100.50:8388"
-weight = 2
-enabled = true
-```
-
-### Пример 5: смешанные upstream с scopes
+### Пример 4: смешанные upstream с scopes
 
 ```toml
 [[upstreams]]

src/api/model.rs

@@ -134,7 +134,6 @@ pub(super) struct UpstreamSummaryData {
     pub(super) direct_total: usize,
     pub(super) socks4_total: usize,
     pub(super) socks5_total: usize,
-    pub(super) shadowsocks_total: usize,
 }
 
 #[derive(Serialize, Clone)]

src/api/runtime_min.rs

@@ -159,7 +159,6 @@ pub(super) struct RuntimeUpstreamQualitySummaryData {
     pub(super) direct_total: usize,
     pub(super) socks4_total: usize,
     pub(super) socks5_total: usize,
-    pub(super) shadowsocks_total: usize,
 }
 
 #[derive(Serialize)]
@@ -407,9 +406,7 @@ pub(super) async fn build_runtime_upstream_quality_data(
         connect_attempt_total: shared.stats.get_upstream_connect_attempt_total(),
         connect_success_total: shared.stats.get_upstream_connect_success_total(),
         connect_fail_total: shared.stats.get_upstream_connect_fail_total(),
-        connect_failfast_hard_error_total: shared
-            .stats
-            .get_upstream_connect_failfast_hard_error_total(),
+        connect_failfast_hard_error_total: shared.stats.get_upstream_connect_failfast_hard_error_total(),
     };
 
     let Some(snapshot) = shared.upstream_manager.try_api_snapshot() else {
@@ -449,7 +446,6 @@ pub(super) async fn build_runtime_upstream_quality_data(
             direct_total: snapshot.summary.direct_total,
             socks4_total: snapshot.summary.socks4_total,
             socks5_total: snapshot.summary.socks5_total,
-            shadowsocks_total: snapshot.summary.shadowsocks_total,
         }),
         upstreams: Some(
             snapshot
@@ -461,7 +457,6 @@ pub(super) async fn build_runtime_upstream_quality_data(
                         crate::transport::UpstreamRouteKind::Direct => "direct",
                         crate::transport::UpstreamRouteKind::Socks4 => "socks4",
                         crate::transport::UpstreamRouteKind::Socks5 => "socks5",
-                        crate::transport::UpstreamRouteKind::Shadowsocks => "shadowsocks",
                     },
                     address: upstream.address,
                     weight: upstream.weight,
@@ -481,9 +476,7 @@ pub(super) async fn build_runtime_upstream_quality_data(
                                 crate::transport::upstream::IpPreference::PreferV6 => "prefer_v6",
                                 crate::transport::upstream::IpPreference::PreferV4 => "prefer_v4",
                                 crate::transport::upstream::IpPreference::BothWork => "both_work",
-                                crate::transport::upstream::IpPreference::Unavailable => {
-                                    "unavailable"
-                                }
+                                crate::transport::upstream::IpPreference::Unavailable => "unavailable",
                             },
                         })
                         .collect(),
@@ -521,18 +514,14 @@ pub(super) async fn build_runtime_nat_stun_data(shared: &ApiShared) -> RuntimeNa
                 live_total: snapshot.live_servers.len(),
             },
             reflection: RuntimeNatStunReflectionBlockData {
-                v4: snapshot
-                    .reflection_v4
-                    .map(|entry| RuntimeNatStunReflectionData {
-                        addr: entry.addr.to_string(),
-                        age_secs: entry.age_secs,
-                    }),
-                v6: snapshot
-                    .reflection_v6
-                    .map(|entry| RuntimeNatStunReflectionData {
-                        addr: entry.addr.to_string(),
-                        age_secs: entry.age_secs,
-                    }),
+                v4: snapshot.reflection_v4.map(|entry| RuntimeNatStunReflectionData {
+                    addr: entry.addr.to_string(),
+                    age_secs: entry.age_secs,
+                }),
+                v6: snapshot.reflection_v6.map(|entry| RuntimeNatStunReflectionData {
+                    addr: entry.addr.to_string(),
+                    age_secs: entry.age_secs,
+                }),
             },
             stun_backoff_remaining_ms: snapshot.stun_backoff_remaining_ms,
         }),

src/api/runtime_selftest.rs

@@ -1,5 +1,5 @@
-use std::collections::HashMap;
 use std::net::IpAddr;
+use std::collections::HashMap;
 use std::sync::{Mutex, OnceLock};
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -7,8 +7,8 @@ use serde::Serialize;
 
 use crate::config::{ProxyConfig, UpstreamType};
 use crate::network::probe::{detect_interface_ipv4, detect_interface_ipv6, is_bogon};
-use crate::transport::UpstreamRouteKind;
 use crate::transport::middle_proxy::{bnd_snapshot, timeskew_snapshot, upstream_bnd_snapshots};
+use crate::transport::UpstreamRouteKind;
 
 use super::ApiShared;
 
@@ -262,8 +262,8 @@ fn update_kdf_ewma(now_epoch_secs: u64, total_errors: u64) -> f64 {
     let delta_errors = total_errors.saturating_sub(guard.last_total_errors);
     let instant_rate_per_min = (delta_errors as f64) * 60.0 / (dt_secs as f64);
     let alpha = 1.0 - f64::exp(-(dt_secs as f64) / KDF_EWMA_TAU_SECS);
-    guard.ewma_errors_per_min =
-        guard.ewma_errors_per_min + alpha * (instant_rate_per_min - guard.ewma_errors_per_min);
+    guard.ewma_errors_per_min = guard.ewma_errors_per_min
+        + alpha * (instant_rate_per_min - guard.ewma_errors_per_min);
     guard.last_epoch_secs = now_epoch_secs;
     guard.last_total_errors = total_errors;
     guard.ewma_errors_per_min
@@ -284,7 +284,6 @@ fn map_route_kind(value: UpstreamRouteKind) -> &'static str {
         UpstreamRouteKind::Direct => "direct",
         UpstreamRouteKind::Socks4 => "socks4",
         UpstreamRouteKind::Socks5 => "socks5",
-        UpstreamRouteKind::Shadowsocks => "shadowsocks",
     }
 }
 

src/api/runtime_stats.rs

@@ -2,8 +2,8 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use crate::config::ApiConfig;
 use crate::stats::Stats;
-use crate::transport::UpstreamRouteKind;
 use crate::transport::upstream::IpPreference;
+use crate::transport::UpstreamRouteKind;
 
 use super::ApiShared;
 use super::model::{
@@ -138,8 +138,7 @@ fn build_zero_upstream_data(stats: &Stats) -> ZeroUpstreamData {
             .get_upstream_connect_duration_success_bucket_501_1000ms(),
         connect_duration_success_bucket_gt_1000ms: stats
             .get_upstream_connect_duration_success_bucket_gt_1000ms(),
-        connect_duration_fail_bucket_le_100ms: stats
-            .get_upstream_connect_duration_fail_bucket_le_100ms(),
+        connect_duration_fail_bucket_le_100ms: stats.get_upstream_connect_duration_fail_bucket_le_100ms(),
         connect_duration_fail_bucket_101_500ms: stats
             .get_upstream_connect_duration_fail_bucket_101_500ms(),
         connect_duration_fail_bucket_501_1000ms: stats
@@ -181,7 +180,6 @@ pub(super) fn build_upstreams_data(shared: &ApiShared, api_cfg: &ApiConfig) -> U
         direct_total: snapshot.summary.direct_total,
         socks4_total: snapshot.summary.socks4_total,
         socks5_total: snapshot.summary.socks5_total,
-        shadowsocks_total: snapshot.summary.shadowsocks_total,
     };
     let upstreams = snapshot
         .upstreams
@@ -397,7 +395,8 @@ async fn get_minimal_payload_cached(
         adaptive_floor_min_writers_multi_endpoint: runtime
             .adaptive_floor_min_writers_multi_endpoint,
         adaptive_floor_recover_grace_secs: runtime.adaptive_floor_recover_grace_secs,
-        adaptive_floor_writers_per_core_total: runtime.adaptive_floor_writers_per_core_total,
+        adaptive_floor_writers_per_core_total: runtime
+            .adaptive_floor_writers_per_core_total,
         adaptive_floor_cpu_cores_override: runtime.adaptive_floor_cpu_cores_override,
         adaptive_floor_max_extra_writers_single_per_core: runtime
             .adaptive_floor_max_extra_writers_single_per_core,
@@ -405,9 +404,12 @@ async fn get_minimal_payload_cached(
             .adaptive_floor_max_extra_writers_multi_per_core,
         adaptive_floor_max_active_writers_per_core: runtime
             .adaptive_floor_max_active_writers_per_core,
-        adaptive_floor_max_warm_writers_per_core: runtime.adaptive_floor_max_warm_writers_per_core,
-        adaptive_floor_max_active_writers_global: runtime.adaptive_floor_max_active_writers_global,
-        adaptive_floor_max_warm_writers_global: runtime.adaptive_floor_max_warm_writers_global,
+        adaptive_floor_max_warm_writers_per_core: runtime
+            .adaptive_floor_max_warm_writers_per_core,
+        adaptive_floor_max_active_writers_global: runtime
+            .adaptive_floor_max_active_writers_global,
+        adaptive_floor_max_warm_writers_global: runtime
+            .adaptive_floor_max_warm_writers_global,
         adaptive_floor_cpu_cores_detected: runtime.adaptive_floor_cpu_cores_detected,
         adaptive_floor_cpu_cores_effective: runtime.adaptive_floor_cpu_cores_effective,
         adaptive_floor_global_cap_raw: runtime.adaptive_floor_global_cap_raw,
@@ -526,7 +528,6 @@ fn map_route_kind(value: UpstreamRouteKind) -> &'static str {
         UpstreamRouteKind::Direct => "direct",
         UpstreamRouteKind::Socks4 => "socks4",
         UpstreamRouteKind::Socks5 => "socks5",
-        UpstreamRouteKind::Shadowsocks => "shadowsocks",
     }
 }
 

src/cli.rs

@@ -199,8 +199,14 @@ update_every = 43200
 hardswap = false
 me_pool_drain_ttl_secs = 90
 me_instadrain = false
+me_pool_drain_threshold = 32
+me_pool_drain_soft_evict_grace_secs = 10
+me_pool_drain_soft_evict_per_writer = 2
+me_pool_drain_soft_evict_budget_per_core = 16
+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 = 120
+me_reinit_drain_timeout_secs = 90
 
 [network]
 ipv4 = true

src/config/defaults.rs

@@ -40,10 +40,10 @@ const DEFAULT_ME_ROUTE_HYBRID_MAX_WAIT_MS: u64 = 3000;
 const DEFAULT_ME_ROUTE_BLOCKING_SEND_TIMEOUT_MS: u64 = 250;
 const DEFAULT_ME_C2ME_SEND_TIMEOUT_MS: u64 = 4000;
 const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_ENABLED: bool = true;
-const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_GRACE_SECS: u64 = 30;
-const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_PER_WRITER: u8 = 1;
-const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_BUDGET_PER_CORE: u16 = 8;
-const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_COOLDOWN_MS: u64 = 5000;
+const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_GRACE_SECS: u64 = 10;
+const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_PER_WRITER: u8 = 2;
+const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_BUDGET_PER_CORE: u16 = 16;
+const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_COOLDOWN_MS: u64 = 1000;
 const DEFAULT_USER_MAX_UNIQUE_IPS_WINDOW_SECS: u64 = 30;
 const DEFAULT_ACCEPT_PERMIT_TIMEOUT_MS: u64 = 250;
 const DEFAULT_UPSTREAM_CONNECT_RETRY_ATTEMPTS: u32 = 2;
@@ -606,7 +606,7 @@ pub(crate) fn default_proxy_secret_len_max() -> usize {
 }
 
 pub(crate) fn default_me_reinit_drain_timeout_secs() -> u64 {
-    120
+    90
 }
 
 pub(crate) fn default_me_pool_drain_ttl_secs() -> u64 {
@@ -618,7 +618,7 @@ pub(crate) fn default_me_instadrain() -> bool {
 }
 
 pub(crate) fn default_me_pool_drain_threshold() -> u64 {
-    128
+    32
 }
 
 pub(crate) fn default_me_pool_drain_soft_evict_enabled() -> bool {

src/config/load.rs

@@ -6,9 +6,8 @@ use std::net::{IpAddr, SocketAddr};
 use std::path::{Path, PathBuf};
 
 use rand::Rng;
-use serde::{Deserialize, Serialize};
-use shadowsocks::config::ServerConfig as ShadowsocksServerConfig;
 use tracing::warn;
+use serde::{Serialize, Deserialize};
 
 use crate::error::{ProxyError, Result};
 
@@ -123,37 +122,13 @@ fn sanitize_ad_tag(ad_tag: &mut Option<String>) {
     };
 
     if !is_valid_ad_tag(tag) {
-        warn!("Invalid general.ad_tag value, expected exactly 32 hex chars; ad_tag is disabled");
+        warn!(
+            "Invalid general.ad_tag value, expected exactly 32 hex chars; ad_tag is disabled"
+        );
         *ad_tag = None;
     }
 }
 
-fn validate_upstreams(config: &ProxyConfig) -> Result<()> {
-    let has_enabled_shadowsocks = config.upstreams.iter().any(|upstream| {
-        upstream.enabled && matches!(upstream.upstream_type, UpstreamType::Shadowsocks { .. })
-    });
-
-    if has_enabled_shadowsocks && config.general.use_middle_proxy {
-        return Err(ProxyError::Config(
-            "shadowsocks upstreams require general.use_middle_proxy = false".to_string(),
-        ));
-    }
-
-    for upstream in &config.upstreams {
-        if let UpstreamType::Shadowsocks { url, .. } = &upstream.upstream_type {
-            let parsed = ShadowsocksServerConfig::from_url(url)
-                .map_err(|error| ProxyError::Config(format!("invalid shadowsocks url: {error}")))?;
-            if parsed.plugin().is_some() {
-                return Err(ProxyError::Config(
-                    "shadowsocks plugins are not supported".to_string(),
-                ));
-            }
-        }
-    }
-
-    Ok(())
-}
-
 // ============= Main Config =============
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -205,8 +180,7 @@ impl ProxyConfig {
 
     pub(crate) fn load_with_metadata<P: AsRef<Path>>(path: P) -> Result<LoadedConfig> {
         let path = path.as_ref();
-        let content =
-            std::fs::read_to_string(path).map_err(|e| ProxyError::Config(e.to_string()))?;
+        let content = std::fs::read_to_string(path).map_err(|e| ProxyError::Config(e.to_string()))?;
         let base_dir = path.parent().unwrap_or(Path::new("."));
         let mut source_files = BTreeSet::new();
         source_files.insert(normalize_config_path(path));
@@ -233,17 +207,15 @@ impl ProxyConfig {
             .map(|table| table.contains_key("stun_servers"))
             .unwrap_or(false);
 
-        let mut config: ProxyConfig = parsed_toml
-            .try_into()
-            .map_err(|e| ProxyError::Config(e.to_string()))?;
+        let mut config: ProxyConfig =
+            parsed_toml.try_into().map_err(|e| ProxyError::Config(e.to_string()))?;
 
         if !update_every_is_explicit && (legacy_secret_is_explicit || legacy_config_is_explicit) {
             config.general.update_every = None;
         }
 
         let legacy_nat_stun = config.general.middle_proxy_nat_stun.take();
-        let legacy_nat_stun_servers =
-            std::mem::take(&mut config.general.middle_proxy_nat_stun_servers);
+        let legacy_nat_stun_servers = std::mem::take(&mut config.general.middle_proxy_nat_stun_servers);
         let legacy_nat_stun_used = legacy_nat_stun.is_some() || !legacy_nat_stun_servers.is_empty();
         if stun_servers_is_explicit {
             let mut explicit_stun_servers = Vec::new();
@@ -253,9 +225,7 @@ impl ProxyConfig {
             config.network.stun_servers = explicit_stun_servers;
 
             if legacy_nat_stun_used {
-                warn!(
-                    "general.middle_proxy_nat_stun and general.middle_proxy_nat_stun_servers are ignored because network.stun_servers is explicitly set"
-                );
+                warn!("general.middle_proxy_nat_stun and general.middle_proxy_nat_stun_servers are ignored because network.stun_servers is explicitly set");
             }
         } else {
             // Keep the default STUN pool unless network.stun_servers is explicitly overridden.
@@ -270,9 +240,7 @@ impl ProxyConfig {
             config.network.stun_servers = unified_stun_servers;
 
             if legacy_nat_stun_used {
-                warn!(
-                    "general.middle_proxy_nat_stun and general.middle_proxy_nat_stun_servers are deprecated; use network.stun_servers"
-                );
+                warn!("general.middle_proxy_nat_stun and general.middle_proxy_nat_stun_servers are deprecated; use network.stun_servers");
             }
         }
 
@@ -410,15 +378,13 @@ impl ProxyConfig {
 
         if !(4096..=1024 * 1024).contains(&config.general.direct_relay_copy_buf_c2s_bytes) {
             return Err(ProxyError::Config(
-                "general.direct_relay_copy_buf_c2s_bytes must be within [4096, 1048576]"
-                    .to_string(),
+                "general.direct_relay_copy_buf_c2s_bytes must be within [4096, 1048576]".to_string(),
             ));
         }
 
         if !(8192..=2 * 1024 * 1024).contains(&config.general.direct_relay_copy_buf_s2c_bytes) {
             return Err(ProxyError::Config(
-                "general.direct_relay_copy_buf_s2c_bytes must be within [8192, 2097152]"
-                    .to_string(),
+                "general.direct_relay_copy_buf_s2c_bytes must be within [8192, 2097152]".to_string(),
             ));
         }
 
@@ -667,8 +633,7 @@ impl ProxyConfig {
 
         if !(1..=100).contains(&config.general.me_route_backpressure_high_watermark_pct) {
             return Err(ProxyError::Config(
-                "general.me_route_backpressure_high_watermark_pct must be within [1, 100]"
-                    .to_string(),
+                "general.me_route_backpressure_high_watermark_pct must be within [1, 100]".to_string(),
             ));
         }
 
@@ -848,15 +813,11 @@ impl ProxyConfig {
         crate::network::dns_overrides::validate_entries(&config.network.dns_overrides)?;
 
         if config.general.use_middle_proxy && config.network.ipv6 == Some(true) {
-            warn!(
-                "IPv6 with Middle Proxy is experimental and may cause KDF address mismatch; consider disabling IPv6 or ME"
-            );
+            warn!("IPv6 with Middle Proxy is experimental and may cause KDF address mismatch; consider disabling IPv6 or ME");
         }
 
         // Random fake_cert_len only when default is in use.
-        if !config.censorship.tls_emulation
-            && config.censorship.fake_cert_len == default_fake_cert_len()
-        {
+        if !config.censorship.tls_emulation && config.censorship.fake_cert_len == default_fake_cert_len() {
             config.censorship.fake_cert_len = rand::rng().gen_range(1024..4096);
         }
 
@@ -866,7 +827,8 @@ impl ProxyConfig {
         let listen_tcp = config.server.listen_tcp.unwrap_or_else(|| {
             if config.server.listen_unix_sock.is_some() {
                 // Unix socket present: TCP only if user explicitly set addresses or listeners.
-                config.server.listen_addr_ipv4.is_some() || !config.server.listeners.is_empty()
+                config.server.listen_addr_ipv4.is_some()
+                    || !config.server.listeners.is_empty()
             } else {
                 true
             }
@@ -874,9 +836,7 @@ impl ProxyConfig {
 
         // Migration: Populate listeners if empty (skip when listen_tcp = false).
         if config.server.listeners.is_empty() && listen_tcp {
-            let ipv4_str = config
-                .server
-                .listen_addr_ipv4
+            let ipv4_str = config.server.listen_addr_ipv4
                 .as_deref()
                 .unwrap_or("0.0.0.0");
             if let Ok(ipv4) = ipv4_str.parse::<IpAddr>() {
@@ -918,10 +878,7 @@ impl ProxyConfig {
         // Migration: Populate upstreams if empty (Default Direct).
         if config.upstreams.is_empty() {
             config.upstreams.push(UpstreamConfig {
-                upstream_type: UpstreamType::Direct {
-                    interface: None,
-                    bind_addresses: None,
-                },
+                upstream_type: UpstreamType::Direct { interface: None, bind_addresses: None },
                 weight: 1,
                 enabled: true,
                 scopes: String::new(),
@@ -935,8 +892,6 @@ impl ProxyConfig {
             .entry("203".to_string())
             .or_insert_with(|| vec!["91.105.192.100:443".to_string()]);
 
-        validate_upstreams(&config)?;
-
         Ok(LoadedConfig {
             config,
             source_files: source_files.into_iter().collect(),
@@ -983,9 +938,6 @@ impl ProxyConfig {
 mod tests {
     use super::*;
 
-    const TEST_SHADOWSOCKS_URL: &str =
-        "ss://2022-blake3-aes-256-gcm:MDEyMzQ1Njc4OTAxMjM0NTY3ODkwMTIzNDU2Nzg5MDE=@127.0.0.1:8388";
-
     #[test]
     fn serde_defaults_remain_unchanged_for_present_sections() {
         let toml = r#"
@@ -1015,7 +967,10 @@ mod tests {
             cfg.general.me_init_retry_attempts,
             default_me_init_retry_attempts()
         );
-        assert_eq!(cfg.general.me2dc_fallback, default_me2dc_fallback());
+        assert_eq!(
+            cfg.general.me2dc_fallback,
+            default_me2dc_fallback()
+        );
         assert_eq!(
             cfg.general.proxy_config_v4_cache_path,
             default_proxy_config_v4_cache_path()
@@ -1324,12 +1279,11 @@ mod tests {
         let path = dir.join("telemt_dc_override_test.toml");
         std::fs::write(&path, toml).unwrap();
         let cfg = ProxyConfig::load(&path).unwrap();
-        assert!(
-            cfg.dc_overrides
-                .get("203")
-                .map(|v| v.contains(&"91.105.192.100:443".to_string()))
-                .unwrap_or(false)
-        );
+        assert!(cfg
+            .dc_overrides
+            .get("203")
+            .map(|v| v.contains(&"91.105.192.100:443".to_string()))
+            .unwrap_or(false));
         let _ = std::fs::remove_file(path);
     }
 
@@ -1516,9 +1470,11 @@ mod tests {
         let path = dir.join("telemt_me_adaptive_floor_min_writers_out_of_range_test.toml");
         std::fs::write(&path, toml).unwrap();
         let err = ProxyConfig::load(&path).unwrap_err().to_string();
-        assert!(err.contains(
-            "general.me_adaptive_floor_min_writers_single_endpoint must be within [1, 32]"
-        ));
+        assert!(
+            err.contains(
+                "general.me_adaptive_floor_min_writers_single_endpoint must be within [1, 32]"
+            )
+        );
         let _ = std::fs::remove_file(path);
     }
 
@@ -2081,6 +2037,45 @@ mod tests {
         let _ = std::fs::remove_file(path);
     }
 
+    #[test]
+    fn force_close_default_matches_drain_ttl() {
+        let toml = r#"
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+        "#;
+        let dir = std::env::temp_dir();
+        let path = dir.join("telemt_force_close_default_test.toml");
+        std::fs::write(&path, toml).unwrap();
+        let cfg = ProxyConfig::load(&path).unwrap();
+        assert_eq!(cfg.general.me_reinit_drain_timeout_secs, 90);
+        assert_eq!(cfg.general.effective_me_pool_force_close_secs(), 90);
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[test]
+    fn force_close_zero_uses_runtime_safety_fallback() {
+        let toml = r#"
+            [general]
+            me_reinit_drain_timeout_secs = 0
+
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+        "#;
+        let dir = std::env::temp_dir();
+        let path = dir.join("telemt_force_close_zero_fallback_test.toml");
+        std::fs::write(&path, toml).unwrap();
+        let cfg = ProxyConfig::load(&path).unwrap();
+        assert_eq!(cfg.general.me_reinit_drain_timeout_secs, 0);
+        assert_eq!(cfg.general.effective_me_pool_force_close_secs(), 300);
+        let _ = std::fs::remove_file(path);
+    }
+
     #[test]
     fn force_close_bumped_when_below_drain_ttl() {
         let toml = r#"
@@ -2145,124 +2140,6 @@ mod tests {
         let _ = std::fs::remove_file(path);
     }
 
-    #[test]
-    fn shadowsocks_upstream_url_loads_successfully() {
-        let toml = format!(
-            r#"
-            [general]
-            use_middle_proxy = false
-
-            [censorship]
-            tls_domain = "example.com"
-
-            [access.users]
-            user = "00000000000000000000000000000000"
-
-            [[upstreams]]
-            type = "shadowsocks"
-            url = "{url}"
-            interface = "127.0.0.2"
-            "#,
-            url = TEST_SHADOWSOCKS_URL,
-        );
-        let dir = std::env::temp_dir();
-        let path = dir.join("telemt_shadowsocks_valid_test.toml");
-        std::fs::write(&path, toml).unwrap();
-        let cfg = ProxyConfig::load(&path).unwrap();
-
-        assert!(matches!(
-            &cfg.upstreams[0].upstream_type,
-            UpstreamType::Shadowsocks { url, interface }
-                if url == TEST_SHADOWSOCKS_URL && interface.as_deref() == Some("127.0.0.2")
-        ));
-
-        let _ = std::fs::remove_file(path);
-    }
-
-    #[test]
-    fn shadowsocks_requires_direct_mode() {
-        let toml = format!(
-            r#"
-            [general]
-            use_middle_proxy = true
-
-            [censorship]
-            tls_domain = "example.com"
-
-            [access.users]
-            user = "00000000000000000000000000000000"
-
-            [[upstreams]]
-            type = "shadowsocks"
-            url = "{url}"
-            "#,
-            url = TEST_SHADOWSOCKS_URL,
-        );
-        let dir = std::env::temp_dir();
-        let path = dir.join("telemt_shadowsocks_me_reject_test.toml");
-        std::fs::write(&path, toml).unwrap();
-        let err = ProxyConfig::load(&path).unwrap_err().to_string();
-
-        assert!(err.contains("shadowsocks upstreams require general.use_middle_proxy = false"));
-
-        let _ = std::fs::remove_file(path);
-    }
-
-    #[test]
-    fn invalid_shadowsocks_url_is_rejected() {
-        let toml = r#"
-            [general]
-            use_middle_proxy = false
-
-            [censorship]
-            tls_domain = "example.com"
-
-            [access.users]
-            user = "00000000000000000000000000000000"
-
-            [[upstreams]]
-            type = "shadowsocks"
-            url = "not-a-valid-ss-url"
-        "#;
-        let dir = std::env::temp_dir();
-        let path = dir.join("telemt_shadowsocks_invalid_url_test.toml");
-        std::fs::write(&path, toml).unwrap();
-        let err = ProxyConfig::load(&path).unwrap_err().to_string();
-
-        assert!(err.contains("invalid shadowsocks url"));
-
-        let _ = std::fs::remove_file(path);
-    }
-
-    #[test]
-    fn shadowsocks_plugins_are_rejected() {
-        let toml = format!(
-            r#"
-            [general]
-            use_middle_proxy = false
-
-            [censorship]
-            tls_domain = "example.com"
-
-            [access.users]
-            user = "00000000000000000000000000000000"
-
-            [[upstreams]]
-            type = "shadowsocks"
-            url = "{url}?plugin=obfs-local%3Bobfs%3Dhttp"
-            "#,
-            url = TEST_SHADOWSOCKS_URL,
-        );
-        let dir = std::env::temp_dir();
-        let path = dir.join("telemt_shadowsocks_plugin_reject_test.toml");
-        std::fs::write(&path, toml).unwrap();
-        let err = ProxyConfig::load(&path).unwrap_err().to_string();
-
-        assert!(err.contains("shadowsocks plugins are not supported"));
-
-        let _ = std::fs::remove_file(path);
-    }
-
     #[test]
     fn invalid_user_ad_tag_reports_access_user_ad_tags_key() {
         let toml = r#"

src/config/types.rs

@@ -135,8 +135,8 @@ impl MeSocksKdfPolicy {
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
 #[serde(rename_all = "lowercase")]
 pub enum MeBindStaleMode {
-    Never,
     #[default]
+    Never,
     Ttl,
     Always,
 }
@@ -855,7 +855,7 @@ pub struct GeneralConfig {
     pub me_pool_min_fresh_ratio: f32,
 
     /// Drain timeout in seconds for stale ME writers after endpoint map changes.
-    /// Set to 0 to keep stale writers draining indefinitely (no force-close).
+    /// Set to 0 to use the runtime safety fallback timeout.
     #[serde(default = "default_me_reinit_drain_timeout_secs")]
     pub me_reinit_drain_timeout_secs: u64,
 
@@ -955,38 +955,24 @@ impl Default for GeneralConfig {
             me_reconnect_backoff_cap_ms: default_reconnect_backoff_cap_ms(),
             me_reconnect_fast_retry_count: default_me_reconnect_fast_retry_count(),
             me_single_endpoint_shadow_writers: default_me_single_endpoint_shadow_writers(),
-            me_single_endpoint_outage_mode_enabled: default_me_single_endpoint_outage_mode_enabled(
-            ),
-            me_single_endpoint_outage_disable_quarantine:
-                default_me_single_endpoint_outage_disable_quarantine(),
-            me_single_endpoint_outage_backoff_min_ms:
-                default_me_single_endpoint_outage_backoff_min_ms(),
-            me_single_endpoint_outage_backoff_max_ms:
-                default_me_single_endpoint_outage_backoff_max_ms(),
-            me_single_endpoint_shadow_rotate_every_secs:
-                default_me_single_endpoint_shadow_rotate_every_secs(),
+            me_single_endpoint_outage_mode_enabled: default_me_single_endpoint_outage_mode_enabled(),
+            me_single_endpoint_outage_disable_quarantine: default_me_single_endpoint_outage_disable_quarantine(),
+            me_single_endpoint_outage_backoff_min_ms: default_me_single_endpoint_outage_backoff_min_ms(),
+            me_single_endpoint_outage_backoff_max_ms: default_me_single_endpoint_outage_backoff_max_ms(),
+            me_single_endpoint_shadow_rotate_every_secs: default_me_single_endpoint_shadow_rotate_every_secs(),
             me_floor_mode: MeFloorMode::default(),
             me_adaptive_floor_idle_secs: default_me_adaptive_floor_idle_secs(),
-            me_adaptive_floor_min_writers_single_endpoint:
-                default_me_adaptive_floor_min_writers_single_endpoint(),
-            me_adaptive_floor_min_writers_multi_endpoint:
-                default_me_adaptive_floor_min_writers_multi_endpoint(),
+            me_adaptive_floor_min_writers_single_endpoint: default_me_adaptive_floor_min_writers_single_endpoint(),
+            me_adaptive_floor_min_writers_multi_endpoint: default_me_adaptive_floor_min_writers_multi_endpoint(),
             me_adaptive_floor_recover_grace_secs: default_me_adaptive_floor_recover_grace_secs(),
-            me_adaptive_floor_writers_per_core_total:
-                default_me_adaptive_floor_writers_per_core_total(),
+            me_adaptive_floor_writers_per_core_total: default_me_adaptive_floor_writers_per_core_total(),
             me_adaptive_floor_cpu_cores_override: default_me_adaptive_floor_cpu_cores_override(),
-            me_adaptive_floor_max_extra_writers_single_per_core:
-                default_me_adaptive_floor_max_extra_writers_single_per_core(),
-            me_adaptive_floor_max_extra_writers_multi_per_core:
-                default_me_adaptive_floor_max_extra_writers_multi_per_core(),
-            me_adaptive_floor_max_active_writers_per_core:
-                default_me_adaptive_floor_max_active_writers_per_core(),
-            me_adaptive_floor_max_warm_writers_per_core:
-                default_me_adaptive_floor_max_warm_writers_per_core(),
-            me_adaptive_floor_max_active_writers_global:
-                default_me_adaptive_floor_max_active_writers_global(),
-            me_adaptive_floor_max_warm_writers_global:
-                default_me_adaptive_floor_max_warm_writers_global(),
+            me_adaptive_floor_max_extra_writers_single_per_core: default_me_adaptive_floor_max_extra_writers_single_per_core(),
+            me_adaptive_floor_max_extra_writers_multi_per_core: default_me_adaptive_floor_max_extra_writers_multi_per_core(),
+            me_adaptive_floor_max_active_writers_per_core: default_me_adaptive_floor_max_active_writers_per_core(),
+            me_adaptive_floor_max_warm_writers_per_core: default_me_adaptive_floor_max_warm_writers_per_core(),
+            me_adaptive_floor_max_active_writers_global: default_me_adaptive_floor_max_active_writers_global(),
+            me_adaptive_floor_max_warm_writers_global: default_me_adaptive_floor_max_warm_writers_global(),
             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(),
@@ -1001,8 +987,7 @@ impl Default for GeneralConfig {
             me_socks_kdf_policy: MeSocksKdfPolicy::Strict,
             me_route_backpressure_base_timeout_ms: default_me_route_backpressure_base_timeout_ms(),
             me_route_backpressure_high_timeout_ms: default_me_route_backpressure_high_timeout_ms(),
-            me_route_backpressure_high_watermark_pct:
-                default_me_route_backpressure_high_watermark_pct(),
+            me_route_backpressure_high_watermark_pct: default_me_route_backpressure_high_watermark_pct(),
             me_health_interval_ms_unhealthy: default_me_health_interval_ms_unhealthy(),
             me_health_interval_ms_healthy: default_me_health_interval_ms_healthy(),
             me_admission_poll_ms: default_me_admission_poll_ms(),
@@ -1028,8 +1013,7 @@ impl Default for GeneralConfig {
             me_hardswap_warmup_delay_min_ms: default_me_hardswap_warmup_delay_min_ms(),
             me_hardswap_warmup_delay_max_ms: default_me_hardswap_warmup_delay_max_ms(),
             me_hardswap_warmup_extra_passes: default_me_hardswap_warmup_extra_passes(),
-            me_hardswap_warmup_pass_backoff_base_ms:
-                default_me_hardswap_warmup_pass_backoff_base_ms(),
+            me_hardswap_warmup_pass_backoff_base_ms: default_me_hardswap_warmup_pass_backoff_base_ms(),
             me_config_stable_snapshots: default_me_config_stable_snapshots(),
             me_config_apply_cooldown_secs: default_me_config_apply_cooldown_secs(),
             me_snapshot_require_http_2xx: default_me_snapshot_require_http_2xx(),
@@ -1073,10 +1057,8 @@ impl GeneralConfig {
     /// Resolve the active updater interval for ME infrastructure refresh tasks.
     /// `update_every` has priority, otherwise legacy proxy_*_auto_reload_secs are used.
     pub fn effective_update_every_secs(&self) -> u64 {
-        self.update_every.unwrap_or_else(|| {
-            self.proxy_secret_auto_reload_secs
-                .min(self.proxy_config_auto_reload_secs)
-        })
+        self.update_every
+            .unwrap_or_else(|| self.proxy_secret_auto_reload_secs.min(self.proxy_config_auto_reload_secs))
     }
 
     /// Resolve periodic zero-downtime reinit interval for ME writers.
@@ -1086,8 +1068,13 @@ impl GeneralConfig {
 
     /// Resolve force-close timeout for stale writers.
     /// `me_reinit_drain_timeout_secs` remains backward-compatible alias.
+    /// A configured `0` uses the runtime safety fallback (300s).
     pub fn effective_me_pool_force_close_secs(&self) -> u64 {
-        self.me_reinit_drain_timeout_secs
+        if self.me_reinit_drain_timeout_secs == 0 {
+            300
+        } else {
+            self.me_reinit_drain_timeout_secs
+        }
     }
 }
 
@@ -1483,11 +1470,6 @@ pub enum UpstreamType {
         #[serde(default)]
         password: Option<String>,
     },
-    Shadowsocks {
-        url: String,
-        #[serde(default)]
-        interface: Option<String>,
-    },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -1568,10 +1550,7 @@ impl ShowLink {
 }
 
 impl Serialize for ShowLink {
-    fn serialize<S: serde::Serializer>(
-        &self,
-        serializer: S,
-    ) -> std::result::Result<S::Ok, S::Error> {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
         match self {
             ShowLink::None => Vec::<String>::new().serialize(serializer),
             ShowLink::All => serializer.serialize_str("*"),
@@ -1581,9 +1560,7 @@ impl Serialize for ShowLink {
 }
 
 impl<'de> Deserialize<'de> for ShowLink {
-    fn deserialize<D: serde::Deserializer<'de>>(
-        deserializer: D,
-    ) -> std::result::Result<Self, D::Error> {
+    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
         use serde::de;
 
         struct ShowLinkVisitor;
@@ -1599,14 +1576,14 @@ impl<'de> Deserialize<'de> for ShowLink {
                 if v == "*" {
                     Ok(ShowLink::All)
                 } else {
-                    Err(de::Error::invalid_value(de::Unexpected::Str(v), &r#""*""#))
+                    Err(de::Error::invalid_value(
+                        de::Unexpected::Str(v),
+                        &r#""*""#,
+                    ))
                 }
             }
 
-            fn visit_seq<A: de::SeqAccess<'de>>(
-                self,
-                mut seq: A,
-            ) -> std::result::Result<ShowLink, A::Error> {
+            fn visit_seq<A: de::SeqAccess<'de>>(self, mut seq: A) -> std::result::Result<ShowLink, A::Error> {
                 let mut names = Vec::new();
                 while let Some(name) = seq.next_element::<String>()? {
                     names.push(name);

src/maestro/me_startup.rs

@@ -332,25 +332,76 @@ pub(crate) async fn initialize_me_pool(
                                             "Middle-End pool initialized successfully"
                                         );
 
-                                        let pool_health = pool_bg.clone();
-                                        let rng_health = rng_bg.clone();
-                                        let min_conns = pool_size;
-                                        tokio::spawn(async move {
-                                            crate::transport::middle_proxy::me_health_monitor(
-                                                pool_health,
-                                                rng_health,
-                                                min_conns,
-                                            )
-                                            .await;
-                                        });
-                                        let pool_drain_enforcer = pool_bg.clone();
-                                        tokio::spawn(async move {
-                                            crate::transport::middle_proxy::me_drain_timeout_enforcer(
-                                                pool_drain_enforcer,
-                                            )
-                                            .await;
-                                        });
-                                        break;
+                                            // ── Supervised background tasks ──────────────────
+                                            // Each task runs inside a nested tokio::spawn so
+                                            // that a panic is caught via JoinHandle and the
+                                            // outer loop restarts the task automatically.
+                                            let pool_health = pool_bg.clone();
+                                            let rng_health = rng_bg.clone();
+                                            let min_conns = pool_size;
+                                            tokio::spawn(async move {
+                                                loop {
+                                                    let p = pool_health.clone();
+                                                    let r = rng_health.clone();
+                                                    let res = tokio::spawn(async move {
+                                                        crate::transport::middle_proxy::me_health_monitor(
+                                                            p, r, min_conns,
+                                                        )
+                                                        .await;
+                                                    })
+                                                    .await;
+                                                    match res {
+                                                        Ok(()) => warn!("me_health_monitor exited unexpectedly, restarting"),
+                                                        Err(e) => {
+                                                            error!(error = %e, "me_health_monitor panicked, restarting in 1s");
+                                                            tokio::time::sleep(Duration::from_secs(1)).await;
+                                                        }
+                                                    }
+                                                }
+                                            });
+                                            let pool_drain_enforcer = pool_bg.clone();
+                                            tokio::spawn(async move {
+                                                loop {
+                                                    let p = pool_drain_enforcer.clone();
+                                                    let res = tokio::spawn(async move {
+                                                        crate::transport::middle_proxy::me_drain_timeout_enforcer(p).await;
+                                                    })
+                                                    .await;
+                                                    match res {
+                                                        Ok(()) => warn!("me_drain_timeout_enforcer exited unexpectedly, restarting"),
+                                                        Err(e) => {
+                                                            error!(error = %e, "me_drain_timeout_enforcer panicked, restarting in 1s");
+                                                            tokio::time::sleep(Duration::from_secs(1)).await;
+                                                        }
+                                                    }
+                                                }
+                                            });
+                                            let pool_watchdog = pool_bg.clone();
+                                            tokio::spawn(async move {
+                                                loop {
+                                                    let p = pool_watchdog.clone();
+                                                    let res = tokio::spawn(async move {
+                                                        crate::transport::middle_proxy::me_zombie_writer_watchdog(p).await;
+                                                    })
+                                                    .await;
+                                                    match res {
+                                                        Ok(()) => warn!("me_zombie_writer_watchdog exited unexpectedly, restarting"),
+                                                        Err(e) => {
+                                                            error!(error = %e, "me_zombie_writer_watchdog panicked, restarting in 1s");
+                                                            tokio::time::sleep(Duration::from_secs(1)).await;
+                                                        }
+                                                    }
+                                                }
+                                            });
+                                            // CRITICAL: keep the current-thread runtime
+                                            // alive. Without this, block_on() returns,
+                                            // the Runtime is dropped, and ALL spawned
+                                            // background tasks (health monitor, drain
+                                            // enforcer, zombie watchdog) are silently
+                                            // cancelled — causing the draining-writer
+                                            // leak that brought us here.
+                                            std::future::pending::<()>().await;
+                                            unreachable!();
                                     }
                                     Err(e) => {
                                         startup_tracker_bg.set_me_last_error(Some(e.to_string())).await;
@@ -408,23 +459,65 @@ pub(crate) async fn initialize_me_pool(
                                     "Middle-End pool initialized successfully"
                                 );
 
-                                let pool_clone = pool.clone();
-                                let rng_clone = rng.clone();
-                                let min_conns = pool_size;
-                                tokio::spawn(async move {
-                                    crate::transport::middle_proxy::me_health_monitor(
-                                        pool_clone, rng_clone, min_conns,
-                                    )
-                                    .await;
-                                });
-                                let pool_drain_enforcer = pool.clone();
-                                tokio::spawn(async move {
-                                    crate::transport::middle_proxy::me_drain_timeout_enforcer(
-                                        pool_drain_enforcer,
-                                    )
-                                    .await;
-                                });
-
+                                    // ── Supervised background tasks ──────────────────
+                                    let pool_clone = pool.clone();
+                                    let rng_clone = rng.clone();
+                                    let min_conns = pool_size;
+                                    tokio::spawn(async move {
+                                        loop {
+                                            let p = pool_clone.clone();
+                                            let r = rng_clone.clone();
+                                            let res = tokio::spawn(async move {
+                                                crate::transport::middle_proxy::me_health_monitor(
+                                                    p, r, min_conns,
+                                                )
+                                                .await;
+                                            })
+                                            .await;
+                                            match res {
+                                                Ok(()) => warn!("me_health_monitor exited unexpectedly, restarting"),
+                                                Err(e) => {
+                                                    error!(error = %e, "me_health_monitor panicked, restarting in 1s");
+                                                    tokio::time::sleep(Duration::from_secs(1)).await;
+                                                }
+                                            }
+                                        }
+                                    });
+                                    let pool_drain_enforcer = pool.clone();
+                                    tokio::spawn(async move {
+                                        loop {
+                                            let p = pool_drain_enforcer.clone();
+                                            let res = tokio::spawn(async move {
+                                                crate::transport::middle_proxy::me_drain_timeout_enforcer(p).await;
+                                            })
+                                            .await;
+                                            match res {
+                                                Ok(()) => warn!("me_drain_timeout_enforcer exited unexpectedly, restarting"),
+                                                Err(e) => {
+                                                    error!(error = %e, "me_drain_timeout_enforcer panicked, restarting in 1s");
+                                                    tokio::time::sleep(Duration::from_secs(1)).await;
+                                                }
+                                            }
+                                        }
+                                    });
+                                    let pool_watchdog = pool.clone();
+                                    tokio::spawn(async move {
+                                        loop {
+                                            let p = pool_watchdog.clone();
+                                            let res = tokio::spawn(async move {
+                                                crate::transport::middle_proxy::me_zombie_writer_watchdog(p).await;
+                                            })
+                                            .await;
+                                            match res {
+                                                Ok(()) => warn!("me_zombie_writer_watchdog exited unexpectedly, restarting"),
+                                                Err(e) => {
+                                                    error!(error = %e, "me_zombie_writer_watchdog panicked, restarting in 1s");
+                                                    tokio::time::sleep(Duration::from_secs(1)).await;
+                                                }
+                                            }
+                                        }
+                                    });
+    
                                 break Some(pool);
                             }
                             Err(e) => {

src/proxy/direct_relay.rs

@@ -3,7 +3,8 @@ use std::io::Write;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf, split};
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
+use tokio::net::TcpStream;
 use tokio::sync::watch;
 use tracing::{debug, info, warn};
 
@@ -14,7 +15,7 @@ use crate::protocol::constants::*;
 use crate::proxy::handshake::{HandshakeSuccess, encrypt_tg_nonce_with_ciphers, generate_tg_nonce};
 use crate::proxy::relay::relay_bidirectional;
 use crate::proxy::route_mode::{
-    ROUTE_SWITCH_ERROR_MSG, RelayRouteMode, RouteCutoverState, affected_cutover_state,
+    RelayRouteMode, RouteCutoverState, ROUTE_SWITCH_ERROR_MSG, affected_cutover_state,
     cutover_stagger_delay,
 };
 use crate::proxy::adaptive_buffers;
@@ -55,11 +56,7 @@ where
     );
 
     let tg_stream = upstream_manager
-        .connect(
-            dc_addr,
-            Some(success.dc_idx),
-            user.strip_prefix("scope_").filter(|s| !s.is_empty()),
-        )
+        .connect(dc_addr, Some(success.dc_idx), user.strip_prefix("scope_").filter(|s| !s.is_empty()))
         .await?;
 
     debug!(peer = %success.peer, dc_addr = %dc_addr, "Connected, performing TG handshake");
@@ -96,9 +93,11 @@ where
     );
     tokio::pin!(relay_result);
     let relay_result = loop {
-        if let Some(cutover) =
-            affected_cutover_state(&route_rx, RelayRouteMode::Direct, route_snapshot.generation)
-        {
+        if let Some(cutover) = affected_cutover_state(
+            &route_rx,
+            RelayRouteMode::Direct,
+            route_snapshot.generation,
+        ) {
             let delay = cutover_stagger_delay(session_id, cutover.generation);
             warn!(
                 user = %user,
@@ -149,9 +148,7 @@ fn get_dc_addr_static(dc_idx: i16, config: &ProxyConfig) -> Result<SocketAddr> {
         for addr_str in addrs {
             match addr_str.parse::<SocketAddr>() {
                 Ok(addr) => parsed.push(addr),
-                Err(_) => {
-                    warn!(dc_idx = dc_idx, addr_str = %addr_str, "Invalid DC override address in config, ignoring")
-                }
+                Err(_) => warn!(dc_idx = dc_idx, addr_str = %addr_str, "Invalid DC override address in config, ignoring"),
             }
         }
 
@@ -173,10 +170,7 @@ fn get_dc_addr_static(dc_idx: i16, config: &ProxyConfig) -> Result<SocketAddr> {
 
     // Unknown DC requested by client without override: log and fall back.
     if !config.dc_overrides.contains_key(&dc_key) {
-        warn!(
-            dc_idx = dc_idx,
-            "Requested non-standard DC with no override; falling back to default cluster"
-        );
+        warn!(dc_idx = dc_idx, "Requested non-standard DC with no override; falling back to default cluster");
         if config.general.unknown_dc_file_log_enabled
             && let Some(path) = &config.general.unknown_dc_log_path
             && let Ok(handle) = tokio::runtime::Handle::try_current()
@@ -210,15 +204,15 @@ fn get_dc_addr_static(dc_idx: i16, config: &ProxyConfig) -> Result<SocketAddr> {
     ))
 }
 
-async fn do_tg_handshake_static<S>(
-    mut stream: S,
+async fn do_tg_handshake_static(
+    mut stream: TcpStream,
     success: &HandshakeSuccess,
     config: &ProxyConfig,
     rng: &SecureRandom,
-) -> Result<(CryptoReader<ReadHalf<S>>, CryptoWriter<WriteHalf<S>>)>
-where
-    S: AsyncRead + AsyncWrite + Unpin,
-{
+) -> Result<(
+    CryptoReader<tokio::net::tcp::OwnedReadHalf>,
+    CryptoWriter<tokio::net::tcp::OwnedWriteHalf>,
+)> {
     let (nonce, _tg_enc_key, _tg_enc_iv, _tg_dec_key, _tg_dec_iv) = generate_tg_nonce(
         success.proto_tag,
         success.dc_idx,
@@ -241,7 +235,7 @@ where
     stream.write_all(&encrypted_nonce).await?;
     stream.flush().await?;
 
-    let (read_half, write_half) = split(stream);
+    let (read_half, write_half) = stream.into_split();
 
     let max_pending = config.general.crypto_pending_buffer;
     Ok((

src/tls_front/fetcher.rs

@@ -7,29 +7,33 @@ use tokio::net::TcpStream;
 #[cfg(unix)]
 use tokio::net::UnixStream;
 use tokio::time::timeout;
-use tokio_rustls::TlsConnector;
 use tokio_rustls::client::TlsStream;
+use tokio_rustls::TlsConnector;
 use tracing::{debug, warn};
 
-use rustls::client::ClientConfig;
 use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
+use rustls::client::ClientConfig;
 use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
 use rustls::{DigitallySignedStruct, Error as RustlsError};
 
-use x509_parser::certificate::X509Certificate;
 use x509_parser::prelude::FromDer;
+use x509_parser::certificate::X509Certificate;
 
 use crate::crypto::SecureRandom;
 use crate::network::dns_overrides::resolve_socket_addr;
 use crate::protocol::constants::{
     TLS_RECORD_APPLICATION, TLS_RECORD_CHANGE_CIPHER, TLS_RECORD_HANDSHAKE,
 };
-use crate::tls_front::types::{
-    ParsedCertificateInfo, ParsedServerHello, TlsBehaviorProfile, TlsCertPayload, TlsExtension,
-    TlsFetchResult, TlsProfileSource,
-};
-use crate::transport::UpstreamStream;
 use crate::transport::proxy_protocol::{ProxyProtocolV1Builder, ProxyProtocolV2Builder};
+use crate::tls_front::types::{
+    ParsedCertificateInfo,
+    ParsedServerHello,
+    TlsBehaviorProfile,
+    TlsCertPayload,
+    TlsExtension,
+    TlsFetchResult,
+    TlsProfileSource,
+};
 
 /// No-op verifier: accept any certificate (we only need lengths and metadata).
 #[derive(Debug)]
@@ -140,27 +144,21 @@ fn build_client_hello(sni: &str, rng: &SecureRandom) -> Vec<u8> {
     exts.extend_from_slice(&0x000au16.to_be_bytes());
     exts.extend_from_slice(&((2 + groups.len() * 2) as u16).to_be_bytes());
     exts.extend_from_slice(&(groups.len() as u16 * 2).to_be_bytes());
-    for g in groups {
-        exts.extend_from_slice(&g.to_be_bytes());
-    }
+    for g in groups { exts.extend_from_slice(&g.to_be_bytes()); }
 
     // signature_algorithms
     let sig_algs: [u16; 4] = [0x0804, 0x0805, 0x0403, 0x0503]; // rsa_pss_rsae_sha256/384, ecdsa_secp256r1_sha256, rsa_pkcs1_sha256
     exts.extend_from_slice(&0x000du16.to_be_bytes());
     exts.extend_from_slice(&((2 + sig_algs.len() * 2) as u16).to_be_bytes());
     exts.extend_from_slice(&(sig_algs.len() as u16 * 2).to_be_bytes());
-    for a in sig_algs {
-        exts.extend_from_slice(&a.to_be_bytes());
-    }
+    for a in sig_algs { exts.extend_from_slice(&a.to_be_bytes()); }
 
     // supported_versions (TLS1.3 + TLS1.2)
     let versions: [u16; 2] = [0x0304, 0x0303];
     exts.extend_from_slice(&0x002bu16.to_be_bytes());
     exts.extend_from_slice(&((1 + versions.len() * 2) as u16).to_be_bytes());
     exts.push((versions.len() * 2) as u8);
-    for v in versions {
-        exts.extend_from_slice(&v.to_be_bytes());
-    }
+    for v in versions { exts.extend_from_slice(&v.to_be_bytes()); }
 
     // key_share (x25519)
     let key = gen_key_share(rng);
@@ -275,10 +273,7 @@ fn parse_server_hello(body: &[u8]) -> Option<ParsedServerHello> {
         pos += 4;
         let data = body.get(pos..pos + elen)?.to_vec();
         pos += elen;
-        extensions.push(TlsExtension {
-            ext_type: etype,
-            data,
-        });
+        extensions.push(TlsExtension { ext_type: etype, data });
     }
 
     Some(ParsedServerHello {
@@ -399,7 +394,7 @@ async fn connect_tcp_with_upstream(
     port: u16,
     connect_timeout: Duration,
     upstream: Option<std::sync::Arc<crate::transport::UpstreamManager>>,
-) -> Result<UpstreamStream> {
+) -> Result<TcpStream> {
     if let Some(manager) = upstream {
         if let Some(addr) = resolve_socket_addr(host, port) {
             match manager.connect(addr, None, None).await {
@@ -413,25 +408,23 @@ async fn connect_tcp_with_upstream(
                     );
                 }
             }
-        } else if let Ok(mut addrs) = tokio::net::lookup_host((host, port)).await
-            && let Some(addr) = addrs.find(|a| a.is_ipv4())
-        {
-            match manager.connect(addr, None, None).await {
-                Ok(stream) => return Ok(stream),
-                Err(e) => {
-                    warn!(
-                        host = %host,
-                        port = port,
-                        error = %e,
-                        "Upstream connect failed, using direct connect"
-                    );
+        } else if let Ok(mut addrs) = tokio::net::lookup_host((host, port)).await {
+            if let Some(addr) = addrs.find(|a| a.is_ipv4()) {
+                match manager.connect(addr, None, None).await {
+                    Ok(stream) => return Ok(stream),
+                    Err(e) => {
+                        warn!(
+                            host = %host,
+                            port = port,
+                            error = %e,
+                            "Upstream connect failed, using direct connect"
+                        );
+                    }
                 }
             }
         }
     }
-    Ok(UpstreamStream::Tcp(
-        connect_with_dns_override(host, port, connect_timeout).await?,
-    ))
+    connect_with_dns_override(host, port, connect_timeout).await
 }
 
 fn encode_tls13_certificate_message(cert_chain_der: &[Vec<u8>]) -> Option<Vec<u8>> {
@@ -450,7 +443,9 @@ fn encode_tls13_certificate_message(cert_chain_der: &[Vec<u8>]) -> Option<Vec<u8
     }
 
     // Certificate = context_len(1) + certificate_list_len(3) + entries
-    let body_len = 1usize.checked_add(3)?.checked_add(certificate_list.len())?;
+    let body_len = 1usize
+        .checked_add(3)?
+        .checked_add(certificate_list.len())?;
 
     let mut message = Vec::with_capacity(4 + body_len);
     message.push(0x0b); // HandshakeType::certificate
@@ -554,8 +549,7 @@ async fn fetch_via_raw_tls(
                     sock = %sock_path,
                     "Raw TLS fetch using mask unix socket"
                 );
-                return fetch_via_raw_tls_stream(stream, sni, connect_timeout, proxy_protocol)
-                    .await;
+                return fetch_via_raw_tls_stream(stream, sni, connect_timeout, proxy_protocol).await;
             }
             Ok(Err(e)) => {
                 warn!(
@@ -622,13 +616,12 @@ where
         .map(|slice| slice.to_vec())
         .unwrap_or_default();
     let cert_chain_der: Vec<Vec<u8>> = certs.iter().map(|c| c.as_ref().to_vec()).collect();
-    let cert_payload =
-        encode_tls13_certificate_message(&cert_chain_der).map(|certificate_message| {
-            TlsCertPayload {
-                cert_chain_der: cert_chain_der.clone(),
-                certificate_message,
-            }
-        });
+    let cert_payload = encode_tls13_certificate_message(&cert_chain_der).map(|certificate_message| {
+        TlsCertPayload {
+            cert_chain_der: cert_chain_der.clone(),
+            certificate_message,
+        }
+    });
 
     let total_cert_len = cert_payload
         .as_ref()

src/transport/middle_proxy/health.rs

@@ -1550,6 +1550,170 @@ async fn maybe_rotate_single_endpoint_shadow(
     );
 }
 
+/// Last-resort safety net for draining writers stuck past their deadline.
+///
+/// Runs every `TICK_SECS` and force-closes any draining writer whose
+/// `drain_deadline_epoch_secs` has been exceeded by more than a threshold.
+///
+/// Two thresholds:
+///   - `SOFT_THRESHOLD_SECS` (60s): writers with no bound clients
+///   - `HARD_THRESHOLD_SECS` (300s): writers WITH bound clients (unconditional)
+///
+/// Intentionally kept trivial and independent of pool config to minimise
+/// the probability of panicking itself. Uses `SystemTime` directly
+/// as a fallback clock source and timeouts on every lock acquisition
+/// and writer removal so one stuck writer cannot block the rest.
+pub async fn me_zombie_writer_watchdog(pool: Arc<MePool>) {
+    use std::time::{SystemTime, UNIX_EPOCH};
+
+    const TICK_SECS: u64 = 30;
+    const SOFT_THRESHOLD_SECS: u64 = 60;
+    const HARD_THRESHOLD_SECS: u64 = 300;
+    const LOCK_TIMEOUT_SECS: u64 = 5;
+    const REMOVE_TIMEOUT_SECS: u64 = 10;
+    const HARD_DETACH_TIMEOUT_STREAK: u8 = 3;
+
+    let mut removal_timeout_streak = HashMap::<u64, u8>::new();
+
+    loop {
+        tokio::time::sleep(Duration::from_secs(TICK_SECS)).await;
+
+        let now = match SystemTime::now().duration_since(UNIX_EPOCH) {
+            Ok(d) => d.as_secs(),
+            Err(_) => continue,
+        };
+
+        // Phase 1: collect zombie IDs under a short read-lock with timeout.
+        let zombie_ids_with_meta: Vec<(u64, bool)> = {
+            let Ok(ws) = tokio::time::timeout(
+                Duration::from_secs(LOCK_TIMEOUT_SECS),
+                pool.writers.read(),
+            )
+            .await
+            else {
+                warn!("zombie_watchdog: writers read-lock timeout, skipping tick");
+                continue;
+            };
+            ws.iter()
+                .filter(|w| w.draining.load(std::sync::atomic::Ordering::Relaxed))
+                .filter_map(|w| {
+                    let deadline = w
+                        .drain_deadline_epoch_secs
+                        .load(std::sync::atomic::Ordering::Relaxed);
+                    if deadline == 0 {
+                        return None;
+                    }
+                    let overdue = now.saturating_sub(deadline);
+                    if overdue == 0 {
+                        return None;
+                    }
+                    let started = w
+                        .draining_started_at_epoch_secs
+                        .load(std::sync::atomic::Ordering::Relaxed);
+                    let drain_age = now.saturating_sub(started);
+                    if drain_age > HARD_THRESHOLD_SECS {
+                        return Some((w.id, true));
+                    }
+                    if overdue > SOFT_THRESHOLD_SECS {
+                        return Some((w.id, false));
+                    }
+                    None
+                })
+                .collect()
+        };
+        // read lock released here
+
+        if zombie_ids_with_meta.is_empty() {
+            removal_timeout_streak.clear();
+            continue;
+        }
+
+        let mut active_zombie_ids = HashSet::<u64>::with_capacity(zombie_ids_with_meta.len());
+        for (writer_id, _) in &zombie_ids_with_meta {
+            active_zombie_ids.insert(*writer_id);
+        }
+        removal_timeout_streak.retain(|writer_id, _| active_zombie_ids.contains(writer_id));
+
+        warn!(
+            zombie_count = zombie_ids_with_meta.len(),
+            soft_threshold_secs = SOFT_THRESHOLD_SECS,
+            hard_threshold_secs = HARD_THRESHOLD_SECS,
+            "Zombie draining writers detected by watchdog, force-closing"
+        );
+
+        // Phase 2: remove each writer individually with a timeout.
+        // One stuck removal cannot block the rest.
+        for (writer_id, had_clients) in &zombie_ids_with_meta {
+            let result = tokio::time::timeout(
+                Duration::from_secs(REMOVE_TIMEOUT_SECS),
+                pool.remove_writer_and_close_clients(*writer_id),
+            )
+            .await;
+            match result {
+                Ok(()) => {
+                    removal_timeout_streak.remove(writer_id);
+                    pool.stats.increment_pool_force_close_total();
+                    pool.stats
+                        .increment_me_draining_writers_reap_progress_total();
+                    info!(
+                        writer_id,
+                        had_clients,
+                        "Zombie writer removed by watchdog"
+                    );
+                }
+                Err(_) => {
+                    let streak = removal_timeout_streak
+                        .entry(*writer_id)
+                        .and_modify(|value| *value = value.saturating_add(1))
+                        .or_insert(1);
+                    warn!(
+                        writer_id,
+                        had_clients,
+                        timeout_streak = *streak,
+                        "Zombie writer removal timed out"
+                    );
+                    if *streak < HARD_DETACH_TIMEOUT_STREAK {
+                        continue;
+                    }
+
+                    let hard_detach = tokio::time::timeout(
+                        Duration::from_secs(REMOVE_TIMEOUT_SECS),
+                        pool.remove_draining_writer_hard_detach(*writer_id),
+                    )
+                    .await;
+                    match hard_detach {
+                        Ok(true) => {
+                            removal_timeout_streak.remove(writer_id);
+                            pool.stats.increment_pool_force_close_total();
+                            pool.stats
+                                .increment_me_draining_writers_reap_progress_total();
+                            info!(
+                                writer_id,
+                                had_clients,
+                                "Zombie writer hard-detached after repeated timeouts"
+                            );
+                        }
+                        Ok(false) => {
+                            removal_timeout_streak.remove(writer_id);
+                            debug!(
+                                writer_id,
+                                had_clients,
+                                "Zombie hard-detach skipped (writer already gone or no longer draining)"
+                            );
+                        }
+                        Err(_) => {
+                            warn!(
+                                writer_id,
+                                had_clients,
+                                "Zombie hard-detach timed out, will retry next tick"
+                            );
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;

src/transport/middle_proxy/health_regression_tests.rs

@@ -12,7 +12,9 @@ use super::codec::WriterCommand;
 use super::health::{health_drain_close_budget, reap_draining_writers};
 use super::pool::{MePool, MeWriter, WriterContour};
 use super::registry::ConnMeta;
-use crate::config::{GeneralConfig, MeRouteNoWriterMode, MeSocksKdfPolicy, MeWriterPickMode};
+use crate::config::{
+    GeneralConfig, MeBindStaleMode, MeRouteNoWriterMode, MeSocksKdfPolicy, MeWriterPickMode,
+};
 use crate::crypto::SecureRandom;
 use crate::network::probe::NetworkDecision;
 use crate::stats::Stats;
@@ -646,10 +648,23 @@ async fn reap_draining_writers_instadrain_removes_non_expired_writers_immediatel
 
 #[test]
 fn general_config_default_drain_threshold_remains_enabled() {
-    assert_eq!(GeneralConfig::default().me_pool_drain_threshold, 128);
+    assert_eq!(GeneralConfig::default().me_pool_drain_threshold, 32);
     assert!(GeneralConfig::default().me_pool_drain_soft_evict_enabled);
     assert_eq!(
-        GeneralConfig::default().me_pool_drain_soft_evict_per_writer,
-        1
+        GeneralConfig::default().me_pool_drain_soft_evict_grace_secs,
+        10
     );
+    assert_eq!(
+        GeneralConfig::default().me_pool_drain_soft_evict_per_writer,
+        2
+    );
+    assert_eq!(
+        GeneralConfig::default().me_pool_drain_soft_evict_budget_per_core,
+        16
+    );
+    assert_eq!(
+        GeneralConfig::default().me_pool_drain_soft_evict_cooldown_ms,
+        1000
+    );
+    assert_eq!(GeneralConfig::default().me_bind_stale_mode, MeBindStaleMode::Never);
 }

src/transport/middle_proxy/mod.rs

@@ -30,7 +30,7 @@ mod health_adversarial_tests;
 
 use bytes::Bytes;
 
-pub use health::{me_drain_timeout_enforcer, me_health_monitor};
+pub use health::{me_drain_timeout_enforcer, me_health_monitor, me_zombie_writer_watchdog};
 #[allow(unused_imports)]
 pub use ping::{run_me_ping, format_sample_line, format_me_route, MePingReport, MePingSample, MePingFamily};
 pub use pool::MePool;

src/transport/middle_proxy/ping.rs

@@ -7,7 +7,6 @@ use tokio::net::UdpSocket;
 use crate::config::{UpstreamConfig, UpstreamType};
 use crate::crypto::SecureRandom;
 use crate::error::ProxyError;
-use crate::transport::shadowsocks::sanitize_shadowsocks_url;
 use crate::transport::{UpstreamEgressInfo, UpstreamRouteKind};
 
 use super::MePool;
@@ -41,11 +40,7 @@ pub fn format_sample_line(sample: &MePingSample) -> String {
     let sign = if sample.dc >= 0 { "+" } else { "-" };
     let addr = format!("{}:{}", sample.addr.ip(), sample.addr.port());
 
-    match (
-        sample.connect_ms,
-        sample.handshake_ms.as_ref(),
-        sample.error.as_ref(),
-    ) {
+    match (sample.connect_ms, sample.handshake_ms.as_ref(), sample.error.as_ref()) {
         (Some(conn), Some(hs), None) => format!(
             "     {sign} {addr}\tPing: {:.0} ms / RPC: {:.0} ms / OK",
             conn, hs
@@ -126,7 +121,6 @@ fn route_from_egress(egress: Option<UpstreamEgressInfo>) -> Option<String> {
                 None => route,
             })
         }
-        UpstreamRouteKind::Shadowsocks => Some("shadowsocks".to_string()),
     }
 }
 
@@ -238,9 +232,6 @@ pub async fn format_me_route(
             }
             UpstreamType::Socks4 { address, .. } => format!("socks4://{address}"),
             UpstreamType::Socks5 { address, .. } => format!("socks5://{address}"),
-            UpstreamType::Shadowsocks { url, .. } => sanitize_shadowsocks_url(url)
-                .map(|address| format!("shadowsocks://{address}"))
-                .unwrap_or_else(|_| "shadowsocks://invalid".to_string()),
         };
     }
 
@@ -263,12 +254,6 @@ pub async fn format_me_route(
     if has_socks5 {
         kinds.push("socks5");
     }
-    if enabled_upstreams
-        .iter()
-        .any(|u| matches!(u.upstream_type, UpstreamType::Shadowsocks { .. }))
-    {
-        kinds.push("shadowsocks");
-    }
     format!("mixed upstreams ({})", kinds.join(", "))
 }
 
@@ -350,10 +335,7 @@ pub async fn run_me_ping(pool: &Arc<MePool>, rng: &SecureRandom) -> Vec<MePingRe
                 Ok((stream, conn_rtt, upstream_egress)) => {
                     connect_ms = Some(conn_rtt);
                     route = route_from_egress(upstream_egress);
-                    match pool
-                        .handshake_only(stream, addr, upstream_egress, rng)
-                        .await
-                    {
+                    match pool.handshake_only(stream, addr, upstream_egress, rng).await {
                         Ok(hs) => {
                             handshake_ms = Some(hs.handshake_ms);
                             // drop halves to close

src/transport/middle_proxy/pool.rs

@@ -18,6 +18,8 @@ use crate::transport::UpstreamManager;
 use super::ConnRegistry;
 use super::codec::WriterCommand;
 
+const ME_FORCE_CLOSE_SAFETY_FALLBACK_SECS: u64 = 300;
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub(super) struct RefillDcKey {
     pub dc: i32,
@@ -229,6 +231,14 @@ impl MePool {
             .as_secs()
     }
 
+    fn normalize_force_close_secs(force_close_secs: u64) -> u64 {
+        if force_close_secs == 0 {
+            ME_FORCE_CLOSE_SAFETY_FALLBACK_SECS
+        } else {
+            force_close_secs
+        }
+    }
+
     pub fn new(
         proxy_tag: Option<Vec<u8>>,
         proxy_secret: Vec<u8>,
@@ -477,7 +487,9 @@ impl MePool {
             me_pool_drain_soft_evict_cooldown_ms: AtomicU64::new(
                 me_pool_drain_soft_evict_cooldown_ms.max(1),
             ),
-            me_pool_force_close_secs: AtomicU64::new(me_pool_force_close_secs),
+            me_pool_force_close_secs: AtomicU64::new(Self::normalize_force_close_secs(
+                me_pool_force_close_secs,
+            )),
             me_pool_min_fresh_ratio_permille: AtomicU32::new(Self::ratio_to_permille(
                 me_pool_min_fresh_ratio,
             )),
@@ -587,8 +599,10 @@ impl MePool {
         );
         self.me_pool_drain_soft_evict_cooldown_ms
             .store(pool_drain_soft_evict_cooldown_ms.max(1), Ordering::Relaxed);
-        self.me_pool_force_close_secs
-            .store(force_close_secs, Ordering::Relaxed);
+        self.me_pool_force_close_secs.store(
+            Self::normalize_force_close_secs(force_close_secs),
+            Ordering::Relaxed,
+        );
         self.me_pool_min_fresh_ratio_permille
             .store(Self::ratio_to_permille(min_fresh_ratio), Ordering::Relaxed);
         self.me_hardswap_warmup_delay_min_ms
@@ -733,12 +747,9 @@ impl MePool {
     }
 
     pub(super) fn force_close_timeout(&self) -> Option<Duration> {
-        let secs = self.me_pool_force_close_secs.load(Ordering::Relaxed);
-        if secs == 0 {
-            None
-        } else {
-            Some(Duration::from_secs(secs))
-        }
+        let secs =
+            Self::normalize_force_close_secs(self.me_pool_force_close_secs.load(Ordering::Relaxed));
+        Some(Duration::from_secs(secs))
     }
 
     pub(super) fn drain_soft_evict_enabled(&self) -> bool {

src/transport/middle_proxy/pool_refill.rs

@@ -74,9 +74,8 @@ impl MePool {
             debug!(
                 %addr,
                 wait_ms = expiry.saturating_duration_since(now).as_millis(),
-                "All ME endpoints are quarantined for the DC group; retrying earliest one"
+                "All ME endpoints are quarantined for the DC group; waiting for quarantine expiry"
             );
-            return vec![addr];
         }
 
         Vec::new()

src/transport/middle_proxy/pool_writer.rs

@@ -20,7 +20,6 @@ use crate::protocol::constants::{RPC_CLOSE_EXT_U32, RPC_PING_U32};
 use super::codec::{RpcWriter, WriterCommand};
 use super::pool::{MePool, MeWriter, WriterContour};
 use super::reader::reader_loop;
-use super::registry::BoundConn;
 use super::wire::build_proxy_req_payload;
 
 const ME_ACTIVE_PING_SECS: u64 = 25;
@@ -28,6 +27,12 @@ const ME_ACTIVE_PING_JITTER_SECS: i64 = 5;
 const ME_IDLE_KEEPALIVE_MAX_SECS: u64 = 5;
 const ME_RPC_PROXY_REQ_RESPONSE_WAIT_MS: u64 = 700;
 
+#[derive(Clone, Copy)]
+enum WriterTeardownMode {
+    Any,
+    DrainingOnly,
+}
+
 fn is_me_peer_closed_error(error: &ProxyError) -> bool {
     matches!(error, ProxyError::Io(ioe) if ioe.kind() == ErrorKind::UnexpectedEof)
 }
@@ -143,6 +148,9 @@ impl MePool {
             crc_mode: hs.crc_mode,
         };
         let cancel_wr = cancel.clone();
+        let cleanup_done = Arc::new(AtomicBool::new(false));
+        let cleanup_for_writer = cleanup_done.clone();
+        let pool_writer_task = Arc::downgrade(self);
         tokio::spawn(async move {
             loop {
                 tokio::select! {
@@ -160,6 +168,16 @@ impl MePool {
                     _ = cancel_wr.cancelled() => break,
                 }
             }
+            if cleanup_for_writer
+                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
+                .is_ok()
+            {
+                if let Some(pool) = pool_writer_task.upgrade() {
+                    pool.remove_writer_and_close_clients(writer_id).await;
+                } else {
+                    cancel_wr.cancel();
+                }
+            }
         });
         let writer = MeWriter {
             id: writer_id,
@@ -196,7 +214,6 @@ impl MePool {
         let cancel_ping = cancel.clone();
         let tx_ping = tx.clone();
         let ping_tracker_ping = ping_tracker.clone();
-        let cleanup_done = Arc::new(AtomicBool::new(false));
         let cleanup_for_reader = cleanup_done.clone();
         let cleanup_for_ping = cleanup_done.clone();
         let keepalive_enabled = self.me_keepalive_enabled;
@@ -242,21 +259,25 @@ impl MePool {
                 stats_reader_close.increment_me_idle_close_by_peer_total();
                 info!(writer_id, "ME socket closed by peer on idle writer");
             }
-            if let Some(pool) = pool.upgrade()
-                && cleanup_for_reader
-                    .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
-                    .is_ok()
+            if cleanup_for_reader
+                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
+                .is_ok()
             {
-                pool.remove_writer_and_close_clients(writer_id).await;
+                if let Some(pool) = pool.upgrade() {
+                    pool.remove_writer_and_close_clients(writer_id).await;
+                } else {
+                    // Fallback for shutdown races: make writer task exit quickly so stale
+                    // channels are observable by periodic prune.
+                    cancel_reader_token.cancel();
+                }
             }
             if let Err(e) = res {
                 if !idle_close_by_peer {
                     warn!(error = %e, "ME reader ended");
                 }
             }
-            let mut ws = writers_arc.write().await;
-            ws.retain(|w| w.id != writer_id);
-            info!(remaining = ws.len(), "Dead ME writer removed from pool");
+            let remaining = writers_arc.read().await.len();
+            debug!(writer_id, remaining, "ME reader task finished");
         });
 
         let pool_ping = Arc::downgrade(self);
@@ -494,18 +515,49 @@ impl MePool {
     pub(crate) async fn remove_writer_and_close_clients(self: &Arc<Self>, writer_id: u64) {
         // Full client cleanup now happens inside `registry.writer_lost` to keep
         // writer reap/remove paths strictly non-blocking per connection.
-        let _ = self.remove_writer_only(writer_id).await;
+        let _ = self
+            .remove_writer_with_mode(writer_id, WriterTeardownMode::Any)
+            .await;
     }
 
-    async fn remove_writer_only(self: &Arc<Self>, writer_id: u64) -> Vec<BoundConn> {
+    pub(super) async fn remove_draining_writer_hard_detach(
+        self: &Arc<Self>,
+        writer_id: u64,
+    ) -> bool {
+        self.remove_writer_with_mode(writer_id, WriterTeardownMode::DrainingOnly)
+            .await
+    }
+
+    async fn remove_writer_only(self: &Arc<Self>, writer_id: u64) -> bool {
+        self.remove_writer_with_mode(writer_id, WriterTeardownMode::Any)
+            .await
+    }
+
+    // Authoritative teardown primitive shared by normal cleanup and watchdog path.
+    // Lock-order invariant:
+    // 1) mutate `writers` under pool write lock,
+    // 2) release pool lock,
+    // 3) run registry/metrics/refill side effects.
+    // `registry.writer_lost` must never run while `writers` lock is held.
+    async fn remove_writer_with_mode(
+        self: &Arc<Self>,
+        writer_id: u64,
+        mode: WriterTeardownMode,
+    ) -> bool {
         let mut close_tx: Option<mpsc::Sender<WriterCommand>> = None;
         let mut removed_addr: Option<SocketAddr> = None;
         let mut removed_dc: Option<i32> = None;
         let mut removed_uptime: Option<Duration> = None;
         let mut trigger_refill = false;
+        let mut removed = false;
         {
             let mut ws = self.writers.write().await;
             if let Some(pos) = ws.iter().position(|w| w.id == writer_id) {
+                if matches!(mode, WriterTeardownMode::DrainingOnly)
+                    && !ws[pos].draining.load(Ordering::Relaxed)
+                {
+                    return false;
+                }
                 let w = ws.remove(pos);
                 let was_draining = w.draining.load(Ordering::Relaxed);
                 if was_draining {
@@ -522,6 +574,7 @@ impl MePool {
                 }
                 close_tx = Some(w.tx.clone());
                 self.conn_count.fetch_sub(1, Ordering::Relaxed);
+                removed = true;
             }
         }
         // State invariant:
@@ -529,7 +582,7 @@ impl MePool {
         // - writer is removed from registry routing/binding maps via `writer_lost`.
         // The close command below is only a best-effort accelerator for task shutdown.
         // Cleanup progress must never depend on command-channel availability.
-        let conns = self.registry.writer_lost(writer_id).await;
+        let _ = self.registry.writer_lost(writer_id).await;
         {
             let mut tracker = self.ping_tracker.lock().await;
             tracker.retain(|_, (_, wid)| *wid != writer_id);
@@ -556,16 +609,17 @@ impl MePool {
                 }
             }
         }
-        if trigger_refill
-            && let Some(addr) = removed_addr
-            && let Some(writer_dc) = removed_dc
-        {
+        if let Some(addr) = removed_addr {
             if let Some(uptime) = removed_uptime {
                 self.maybe_quarantine_flapping_endpoint(addr, uptime).await;
             }
-            self.trigger_immediate_refill_for_dc(addr, writer_dc);
+            if trigger_refill
+                && let Some(writer_dc) = removed_dc
+            {
+                self.trigger_immediate_refill_for_dc(addr, writer_dc);
+            }
         }
-        conns
+        removed
     }
 
     pub(crate) async fn mark_writer_draining_with_timeout(

src/transport/mod.rs

@@ -2,7 +2,6 @@
 
 pub mod pool;
 pub mod proxy_protocol;
-pub mod shadowsocks;
 pub mod socket;
 pub mod socks;
 pub mod upstream;
@@ -15,8 +14,5 @@ pub use socket::*;
 #[allow(unused_imports)]
 pub use socks::*;
 #[allow(unused_imports)]
-pub use upstream::{
-    DcPingResult, StartupPingResult, UpstreamEgressInfo, UpstreamManager, UpstreamRouteKind,
-    UpstreamStream,
-};
+pub use upstream::{DcPingResult, StartupPingResult, UpstreamEgressInfo, UpstreamManager, UpstreamRouteKind};
 pub mod middle_proxy;

src/transport/shadowsocks.rs

@@ -1,60 +0,0 @@
-use std::net::{IpAddr, SocketAddr};
-use std::time::Duration;
-
-use shadowsocks::{
-    ProxyClientStream,
-    config::{ServerConfig, ServerType},
-    context::Context,
-    net::ConnectOpts,
-};
-
-use crate::error::{ProxyError, Result};
-
-pub(crate) type ShadowsocksStream = ProxyClientStream<shadowsocks::net::TcpStream>;
-
-fn parse_server_config(url: &str, connect_timeout: Duration) -> Result<ServerConfig> {
-    let mut config = ServerConfig::from_url(url)
-        .map_err(|error| ProxyError::Config(format!("invalid shadowsocks url: {error}")))?;
-
-    if config.plugin().is_some() {
-        return Err(ProxyError::Config(
-            "shadowsocks plugins are not supported".to_string(),
-        ));
-    }
-
-    config.set_timeout(connect_timeout);
-    Ok(config)
-}
-
-pub(crate) fn sanitize_shadowsocks_url(url: &str) -> Result<String> {
-    Ok(parse_server_config(url, Duration::from_secs(1))?
-        .addr()
-        .to_string())
-}
-
-fn connect_opts_for_interface(interface: &Option<String>) -> ConnectOpts {
-    let mut opts = ConnectOpts::default();
-    if let Some(interface) = interface {
-        if let Ok(ip) = interface.parse::<IpAddr>() {
-            opts.bind_local_addr = Some(SocketAddr::new(ip, 0));
-        } else {
-            opts.bind_interface = Some(interface.clone());
-        }
-    }
-    opts
-}
-
-pub(crate) async fn connect_shadowsocks(
-    url: &str,
-    interface: &Option<String>,
-    target: SocketAddr,
-    connect_timeout: Duration,
-) -> Result<ShadowsocksStream> {
-    let config = parse_server_config(url, connect_timeout)?;
-    let context = Context::new_shared(ServerType::Local);
-    let opts = connect_opts_for_interface(interface);
-
-    ProxyClientStream::connect_with_opts(context, &config, target, &opts)
-        .await
-        .map_err(ProxyError::Io)
-}

src/transport/upstream.rs

@@ -4,28 +4,22 @@
 
 #![allow(deprecated)]
 
-use rand::Rng;
 use std::collections::{BTreeSet, HashMap};
-use std::net::{IpAddr, SocketAddr};
-use std::pin::Pin;
+use std::net::{SocketAddr, IpAddr};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
-use std::task::{Context, Poll};
 use std::time::Duration;
-use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::net::TcpStream;
 use tokio::sync::RwLock;
 use tokio::time::Instant;
-use tracing::{debug, info, trace, warn};
+use rand::Rng;
+use tracing::{debug, warn, info, trace};
 
 use crate::config::{UpstreamConfig, UpstreamType};
-use crate::error::{ProxyError, Result};
+use crate::error::{Result, ProxyError};
 use crate::network::dns_overrides::{resolve_socket_addr, split_host_port};
-use crate::protocol::constants::{TG_DATACENTER_PORT, TG_DATACENTERS_V4, TG_DATACENTERS_V6};
+use crate::protocol::constants::{TG_DATACENTERS_V4, TG_DATACENTERS_V6, TG_DATACENTER_PORT};
 use crate::stats::Stats;
-use crate::transport::shadowsocks::{
-    ShadowsocksStream, connect_shadowsocks, sanitize_shadowsocks_url,
-};
 use crate::transport::socket::{create_outgoing_socket_bound, resolve_interface_ip};
 use crate::transport::socks::{connect_socks4, connect_socks5};
 
@@ -53,10 +47,7 @@ struct LatencyEma {
 
 impl LatencyEma {
     const fn new(alpha: f64) -> Self {
-        Self {
-            value_ms: None,
-            alpha,
-        }
+        Self { value_ms: None, alpha }
     }
 
     fn update(&mut self, sample_ms: f64) {
@@ -140,17 +131,11 @@ impl UpstreamState {
             return Some(ms);
         }
 
-        let (sum, count) = self
-            .dc_latency
-            .iter()
+        let (sum, count) = self.dc_latency.iter()
             .filter_map(|l| l.get())
             .fold((0.0, 0u32), |(s, c), v| (s + v, c + 1));
 
-        if count > 0 {
-            Some(sum / count as f64)
-        } else {
-            None
-        }
+        if count > 0 { Some(sum / count as f64) } else { None }
     }
 }
 
@@ -173,78 +158,11 @@ pub struct StartupPingResult {
     pub both_available: bool,
 }
 
-pub enum UpstreamStream {
-    Tcp(TcpStream),
-    Shadowsocks(Box<ShadowsocksStream>),
-}
-
-impl std::fmt::Debug for UpstreamStream {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::Tcp(_) => f.write_str("UpstreamStream::Tcp(..)"),
-            Self::Shadowsocks(_) => f.write_str("UpstreamStream::Shadowsocks(..)"),
-        }
-    }
-}
-
-impl UpstreamStream {
-    pub fn into_tcp(self) -> Result<TcpStream> {
-        match self {
-            Self::Tcp(stream) => Ok(stream),
-            Self::Shadowsocks(_) => Err(ProxyError::Config(
-                "shadowsocks upstreams are not supported when general.use_middle_proxy = true"
-                    .to_string(),
-            )),
-        }
-    }
-}
-
-impl AsyncRead for UpstreamStream {
-    fn poll_read(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut ReadBuf<'_>,
-    ) -> Poll<std::io::Result<()>> {
-        match self.get_mut() {
-            Self::Tcp(stream) => Pin::new(stream).poll_read(cx, buf),
-            Self::Shadowsocks(stream) => Pin::new(stream.as_mut()).poll_read(cx, buf),
-        }
-    }
-}
-
-impl AsyncWrite for UpstreamStream {
-    fn poll_write(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<std::io::Result<usize>> {
-        match self.get_mut() {
-            Self::Tcp(stream) => Pin::new(stream).poll_write(cx, buf),
-            Self::Shadowsocks(stream) => Pin::new(stream.as_mut()).poll_write(cx, buf),
-        }
-    }
-
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        match self.get_mut() {
-            Self::Tcp(stream) => Pin::new(stream).poll_flush(cx),
-            Self::Shadowsocks(stream) => Pin::new(stream.as_mut()).poll_flush(cx),
-        }
-    }
-
-    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        match self.get_mut() {
-            Self::Tcp(stream) => Pin::new(stream).poll_shutdown(cx),
-            Self::Shadowsocks(stream) => Pin::new(stream.as_mut()).poll_shutdown(cx),
-        }
-    }
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum UpstreamRouteKind {
     Direct,
     Socks4,
     Socks5,
-    Shadowsocks,
 }
 
 #[derive(Debug, Clone)]
@@ -276,7 +194,6 @@ pub struct UpstreamApiSummarySnapshot {
     pub direct_total: usize,
     pub socks4_total: usize,
     pub socks5_total: usize,
-    pub shadowsocks_total: usize,
 }
 
 #[derive(Debug, Clone)]
@@ -336,8 +253,7 @@ impl UpstreamManager {
         connect_failfast_hard_errors: bool,
         stats: Arc<Stats>,
     ) -> Self {
-        let states = configs
-            .into_iter()
+        let states = configs.into_iter()
             .filter(|c| c.enabled)
             .map(UpstreamState::new)
             .collect();
@@ -395,13 +311,20 @@ impl UpstreamManager {
                 summary.unhealthy_total += 1;
             }
 
-            let (route_kind, address) = Self::describe_upstream(&upstream.config.upstream_type);
-            match route_kind {
-                UpstreamRouteKind::Direct => summary.direct_total += 1,
-                UpstreamRouteKind::Socks4 => summary.socks4_total += 1,
-                UpstreamRouteKind::Socks5 => summary.socks5_total += 1,
-                UpstreamRouteKind::Shadowsocks => summary.shadowsocks_total += 1,
-            }
+            let (route_kind, address) = match &upstream.config.upstream_type {
+                UpstreamType::Direct { .. } => {
+                    summary.direct_total += 1;
+                    (UpstreamRouteKind::Direct, "direct".to_string())
+                }
+                UpstreamType::Socks4 { address, .. } => {
+                    summary.socks4_total += 1;
+                    (UpstreamRouteKind::Socks4, address.clone())
+                }
+                UpstreamType::Socks5 { address, .. } => {
+                    summary.socks5_total += 1;
+                    (UpstreamRouteKind::Socks5, address.clone())
+                }
+            };
 
             let mut dc = Vec::with_capacity(NUM_DCS);
             for dc_idx in 0..NUM_DCS {
@@ -429,18 +352,6 @@ impl UpstreamManager {
         Some(UpstreamApiSnapshot { summary, upstreams })
     }
 
-    fn describe_upstream(upstream_type: &UpstreamType) -> (UpstreamRouteKind, String) {
-        match upstream_type {
-            UpstreamType::Direct { .. } => (UpstreamRouteKind::Direct, "direct".to_string()),
-            UpstreamType::Socks4 { address, .. } => (UpstreamRouteKind::Socks4, address.clone()),
-            UpstreamType::Socks5 { address, .. } => (UpstreamRouteKind::Socks5, address.clone()),
-            UpstreamType::Shadowsocks { url, .. } => (
-                UpstreamRouteKind::Shadowsocks,
-                sanitize_shadowsocks_url(url).unwrap_or_else(|_| "invalid".to_string()),
-            ),
-        }
-    }
-
     pub fn api_policy_snapshot(&self) -> UpstreamApiPolicySnapshot {
         UpstreamApiPolicySnapshot {
             connect_retry_attempts: self.connect_retry_attempts,
@@ -628,44 +539,44 @@ impl UpstreamManager {
         // Scope filter:
         //   If scope is set: only scoped and matched items
         //   If scope is not set: only unscoped items
-        let filtered_upstreams: Vec<usize> = upstreams
-            .iter()
+        let filtered_upstreams : Vec<usize> = upstreams.iter()
             .enumerate()
             .filter(|(_, u)| {
-                scope.map_or(u.config.scopes.is_empty(), |req_scope| {
-                    u.config
-                        .scopes
-                        .split(',')
-                        .map(str::trim)
-                        .any(|s| s == req_scope)
-                })
+                scope.map_or(
+                    u.config.scopes.is_empty(),
+                    |req_scope| {
+                        u.config.scopes
+                            .split(',')
+                            .map(str::trim)
+                            .any(|s| s == req_scope)
+                    }
+                )
             })
             .map(|(i, _)| i)
             .collect();
 
         // Healthy filter
-        let healthy: Vec<usize> = filtered_upstreams
-            .iter()
+        let healthy: Vec<usize> = filtered_upstreams.iter()
             .filter(|&&i| upstreams[i].healthy)
             .copied()
             .collect();
 
         if filtered_upstreams.is_empty() {
-            if Self::should_emit_warn(self.no_upstreams_warn_epoch_ms.as_ref(), 5_000) {
-                warn!(
-                    scope = scope,
-                    "No upstreams available! Using first (direct?)"
-                );
+            if Self::should_emit_warn(
+                self.no_upstreams_warn_epoch_ms.as_ref(),
+                5_000,
+            ) {
+                warn!(scope = scope, "No upstreams available! Using first (direct?)");
             }
             return None;
         }
 
         if healthy.is_empty() {
-            if Self::should_emit_warn(self.no_healthy_warn_epoch_ms.as_ref(), 5_000) {
-                warn!(
-                    scope = scope,
-                    "No healthy upstreams available! Using random."
-                );
+            if Self::should_emit_warn(
+                self.no_healthy_warn_epoch_ms.as_ref(),
+                5_000,
+            ) {
+                warn!(scope = scope, "No healthy upstreams available! Using random.");
             }
             return Some(filtered_upstreams[rand::rng().gen_range(0..filtered_upstreams.len())]);
         }
@@ -674,18 +585,14 @@ impl UpstreamManager {
             return Some(healthy[0]);
         }
 
-        let weights: Vec<(usize, f64)> = healthy
-            .iter()
-            .map(|&i| {
-                let base = upstreams[i].config.weight as f64;
-                let latency_factor = upstreams[i]
-                    .effective_latency(dc_idx)
-                    .map(|ms| if ms > 1.0 { 1000.0 / ms } else { 1000.0 })
-                    .unwrap_or(1.0);
+        let weights: Vec<(usize, f64)> = healthy.iter().map(|&i| {
+            let base = upstreams[i].config.weight as f64;
+            let latency_factor = upstreams[i].effective_latency(dc_idx)
+                .map(|ms| if ms > 1.0 { 1000.0 / ms } else { 1000.0 })
+                .unwrap_or(1.0);
 
-                (i, base * latency_factor)
-            })
-            .collect();
+            (i, base * latency_factor)
+        }).collect();
 
         let total: f64 = weights.iter().map(|(_, w)| w).sum();
 
@@ -713,34 +620,8 @@ impl UpstreamManager {
     }
 
     /// Connect to target through a selected upstream.
-    pub async fn connect(
-        &self,
-        target: SocketAddr,
-        dc_idx: Option<i16>,
-        scope: Option<&str>,
-    ) -> Result<UpstreamStream> {
-        let idx = self
-            .select_upstream(dc_idx, scope)
-            .await
-            .ok_or_else(|| ProxyError::Config("No upstreams available".to_string()))?;
-
-        let mut upstream = {
-            let guard = self.upstreams.read().await;
-            guard[idx].config.clone()
-        };
-
-        if let Some(s) = scope {
-            upstream.selected_scope = s.to_string();
-        }
-
-        let bind_rr = {
-            let guard = self.upstreams.read().await;
-            guard.get(idx).map(|u| u.bind_rr.clone())
-        };
-
-        let (stream, _) = self
-            .connect_selected_upstream(idx, upstream, target, dc_idx, bind_rr)
-            .await?;
+    pub async fn connect(&self, target: SocketAddr, dc_idx: Option<i16>, scope: Option<&str>) -> Result<TcpStream> {
+        let (stream, _) = self.connect_with_details(target, dc_idx, scope).await?;
         Ok(stream)
     }
 
@@ -751,9 +632,7 @@ impl UpstreamManager {
         dc_idx: Option<i16>,
         scope: Option<&str>,
     ) -> Result<(TcpStream, UpstreamEgressInfo)> {
-        let idx = self
-            .select_upstream(dc_idx, scope)
-            .await
+        let idx = self.select_upstream(dc_idx, scope).await
             .ok_or_else(|| ProxyError::Config("No upstreams available".to_string()))?;
 
         let mut upstream = {
@@ -771,20 +650,6 @@ impl UpstreamManager {
             guard.get(idx).map(|u| u.bind_rr.clone())
         };
 
-        let (stream, egress) = self
-            .connect_selected_upstream(idx, upstream, target, dc_idx, bind_rr)
-            .await?;
-        Ok((stream.into_tcp()?, egress))
-    }
-
-    async fn connect_selected_upstream(
-        &self,
-        idx: usize,
-        upstream: UpstreamConfig,
-        target: SocketAddr,
-        dc_idx: Option<i16>,
-        bind_rr: Option<Arc<AtomicUsize>>,
-    ) -> Result<(UpstreamStream, UpstreamEgressInfo)> {
         let connect_started_at = Instant::now();
         let mut last_error: Option<ProxyError> = None;
         let mut attempts_used = 0u32;
@@ -797,8 +662,8 @@ impl UpstreamManager {
                 break;
             }
             let remaining_budget = self.connect_budget.saturating_sub(elapsed);
-            let attempt_timeout =
-                Duration::from_secs(DIRECT_CONNECT_TIMEOUT_SECS).min(remaining_budget);
+            let attempt_timeout = Duration::from_secs(DIRECT_CONNECT_TIMEOUT_SECS)
+                .min(remaining_budget);
             if attempt_timeout.is_zero() {
                 last_error = Some(ProxyError::ConnectionTimeout {
                     addr: target.to_string(),
@@ -921,12 +786,9 @@ impl UpstreamManager {
         target: SocketAddr,
         bind_rr: Option<Arc<AtomicUsize>>,
         connect_timeout: Duration,
-    ) -> Result<(UpstreamStream, UpstreamEgressInfo)> {
+    ) -> Result<(TcpStream, UpstreamEgressInfo)> {
         match &config.upstream_type {
-            UpstreamType::Direct {
-                interface,
-                bind_addresses,
-            } => {
+            UpstreamType::Direct { interface, bind_addresses } => {
                 let bind_ip = Self::resolve_bind_address(
                     interface,
                     bind_addresses,
@@ -934,7 +796,9 @@ impl UpstreamManager {
                     bind_rr.as_deref(),
                     true,
                 );
-                if bind_ip.is_none() && bind_addresses.as_ref().is_some_and(|v| !v.is_empty()) {
+                if bind_ip.is_none()
+                    && bind_addresses.as_ref().is_some_and(|v| !v.is_empty())
+                {
                     return Err(ProxyError::Config(format!(
                         "No valid bind_addresses for target family {target}"
                     )));
@@ -949,10 +813,8 @@ impl UpstreamManager {
 
                 socket.set_nonblocking(true)?;
                 match socket.connect(&target.into()) {
-                    Ok(()) => {}
-                    Err(err)
-                        if err.raw_os_error() == Some(libc::EINPROGRESS)
-                            || err.kind() == std::io::ErrorKind::WouldBlock => {}
+                    Ok(()) => {},
+                    Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
                     Err(err) => return Err(ProxyError::Io(err)),
                 }
 
@@ -974,7 +836,7 @@ impl UpstreamManager {
 
                 let local_addr = stream.local_addr().ok();
                 Ok((
-                    UpstreamStream::Tcp(stream),
+                    stream,
                     UpstreamEgressInfo {
                         upstream_id,
                         route_kind: UpstreamRouteKind::Direct,
@@ -984,12 +846,8 @@ impl UpstreamManager {
                         socks_proxy_addr: None,
                     },
                 ))
-            }
-            UpstreamType::Socks4 {
-                address,
-                interface,
-                user_id,
-            } => {
+            },
+            UpstreamType::Socks4 { address, interface, user_id } => {
                 // Try to parse as SocketAddr first (IP:port), otherwise treat as hostname:port
                 let mut stream = if let Ok(proxy_addr) = address.parse::<SocketAddr>() {
                     // IP:port format - use socket with optional interface binding
@@ -1005,10 +863,8 @@ impl UpstreamManager {
 
                     socket.set_nonblocking(true)?;
                     match socket.connect(&proxy_addr.into()) {
-                        Ok(()) => {}
-                        Err(err)
-                            if err.raw_os_error() == Some(libc::EINPROGRESS)
-                                || err.kind() == std::io::ErrorKind::WouldBlock => {}
+                        Ok(()) => {},
+                        Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
                         Err(err) => return Err(ProxyError::Io(err)),
                     }
 
@@ -1032,16 +888,14 @@ impl UpstreamManager {
                     // Hostname:port format - use tokio DNS resolution
                     // Note: interface binding is not supported for hostnames
                     if interface.is_some() {
-                        warn!(
-                            "SOCKS4 interface binding is not supported for hostname addresses, ignoring"
-                        );
+                        warn!("SOCKS4 interface binding is not supported for hostname addresses, ignoring");
                     }
                     Self::connect_hostname_with_dns_override(address, connect_timeout).await?
                 };
 
                 // replace socks user_id with config.selected_scope, if set
-                let scope: Option<&str> =
-                    Some(config.selected_scope.as_str()).filter(|s| !s.is_empty());
+                let scope: Option<&str> = Some(config.selected_scope.as_str())
+                    .filter(|s| !s.is_empty());
                 let _user_id: Option<&str> = scope.or(user_id.as_deref());
 
                 let bound = match tokio::time::timeout(
@@ -1061,7 +915,7 @@ impl UpstreamManager {
                 let local_addr = stream.local_addr().ok();
                 let socks_proxy_addr = stream.peer_addr().ok();
                 Ok((
-                    UpstreamStream::Tcp(stream),
+                    stream,
                     UpstreamEgressInfo {
                         upstream_id,
                         route_kind: UpstreamRouteKind::Socks4,
@@ -1071,13 +925,8 @@ impl UpstreamManager {
                         socks_proxy_addr,
                     },
                 ))
-            }
-            UpstreamType::Socks5 {
-                address,
-                interface,
-                username,
-                password,
-            } => {
+            },
+            UpstreamType::Socks5 { address, interface, username, password } => {
                 // Try to parse as SocketAddr first (IP:port), otherwise treat as hostname:port
                 let mut stream = if let Ok(proxy_addr) = address.parse::<SocketAddr>() {
                     // IP:port format - use socket with optional interface binding
@@ -1093,10 +942,8 @@ impl UpstreamManager {
 
                     socket.set_nonblocking(true)?;
                     match socket.connect(&proxy_addr.into()) {
-                        Ok(()) => {}
-                        Err(err)
-                            if err.raw_os_error() == Some(libc::EINPROGRESS)
-                                || err.kind() == std::io::ErrorKind::WouldBlock => {}
+                        Ok(()) => {},
+                        Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
                         Err(err) => return Err(ProxyError::Io(err)),
                     }
 
@@ -1120,17 +967,15 @@ impl UpstreamManager {
                     // Hostname:port format - use tokio DNS resolution
                     // Note: interface binding is not supported for hostnames
                     if interface.is_some() {
-                        warn!(
-                            "SOCKS5 interface binding is not supported for hostname addresses, ignoring"
-                        );
+                        warn!("SOCKS5 interface binding is not supported for hostname addresses, ignoring");
                     }
                     Self::connect_hostname_with_dns_override(address, connect_timeout).await?
                 };
 
                 debug!(config = ?config, "Socks5 connection");
                 // replace socks user:pass with config.selected_scope, if set
-                let scope: Option<&str> =
-                    Some(config.selected_scope.as_str()).filter(|s| !s.is_empty());
+                let scope: Option<&str> = Some(config.selected_scope.as_str())
+                    .filter(|s| !s.is_empty());
                 let _username: Option<&str> = scope.or(username.as_deref());
                 let _password: Option<&str> = scope.or(password.as_deref());
 
@@ -1151,7 +996,7 @@ impl UpstreamManager {
                 let local_addr = stream.local_addr().ok();
                 let socks_proxy_addr = stream.peer_addr().ok();
                 Ok((
-                    UpstreamStream::Tcp(stream),
+                    stream,
                     UpstreamEgressInfo {
                         upstream_id,
                         route_kind: UpstreamRouteKind::Socks5,
@@ -1161,22 +1006,7 @@ impl UpstreamManager {
                         socks_proxy_addr,
                     },
                 ))
-            }
-            UpstreamType::Shadowsocks { url, interface } => {
-                let stream = connect_shadowsocks(url, interface, target, connect_timeout).await?;
-                let local_addr = stream.get_ref().local_addr().ok();
-                    Ok((
-                        UpstreamStream::Shadowsocks(Box::new(stream)),
-                        UpstreamEgressInfo {
-                        upstream_id,
-                        route_kind: UpstreamRouteKind::Shadowsocks,
-                        local_addr,
-                        direct_bind_ip: None,
-                        socks_bound_addr: None,
-                        socks_proxy_addr: None,
-                    },
-                ))
-            }
+            },
         }
     }
 
@@ -1193,9 +1023,7 @@ impl UpstreamManager {
     ) -> Vec<StartupPingResult> {
         let upstreams: Vec<(usize, UpstreamConfig, Arc<AtomicUsize>)> = {
             let guard = self.upstreams.read().await;
-            guard
-                .iter()
-                .enumerate()
+            guard.iter().enumerate()
                 .map(|(i, u)| (i, u.config.clone(), u.bind_rr.clone()))
                 .collect()
         };
@@ -1223,11 +1051,6 @@ impl UpstreamManager {
                 }
                 UpstreamType::Socks4 { address, .. } => format!("socks4://{}", address),
                 UpstreamType::Socks5 { address, .. } => format!("socks5://{}", address),
-                UpstreamType::Shadowsocks { url, .. } => {
-                    let address =
-                        sanitize_shadowsocks_url(url).unwrap_or_else(|_| "invalid".to_string());
-                    format!("shadowsocks://{address}")
-                }
             };
 
             let mut v6_results = Vec::with_capacity(NUM_DCS);
@@ -1238,14 +1061,8 @@ impl UpstreamManager {
 
                     let result = tokio::time::timeout(
                         Duration::from_secs(DC_PING_TIMEOUT_SECS),
-                        self.ping_single_dc(
-                            *upstream_idx,
-                            upstream_config,
-                            Some(bind_rr.clone()),
-                            addr_v6,
-                        ),
-                    )
-                    .await;
+                        self.ping_single_dc(*upstream_idx, upstream_config, Some(bind_rr.clone()), addr_v6)
+                    ).await;
 
                     let ping_result = match result {
                         Ok(Ok(rtt_ms)) => {
@@ -1295,14 +1112,8 @@ impl UpstreamManager {
 
                     let result = tokio::time::timeout(
                         Duration::from_secs(DC_PING_TIMEOUT_SECS),
-                        self.ping_single_dc(
-                            *upstream_idx,
-                            upstream_config,
-                            Some(bind_rr.clone()),
-                            addr_v4,
-                        ),
-                    )
-                    .await;
+                        self.ping_single_dc(*upstream_idx, upstream_config, Some(bind_rr.clone()), addr_v4)
+                    ).await;
 
                     let ping_result = match result {
                         Ok(Ok(rtt_ms)) => {
@@ -1351,7 +1162,7 @@ impl UpstreamManager {
                     Err(_) => {
                         warn!(dc = %dc_key, "Invalid dc_overrides key, skipping");
                         continue;
-                    }
+                    },
                     _ => continue,
                 };
                 let dc_idx = dc_num as usize;
@@ -1364,14 +1175,8 @@ impl UpstreamManager {
                             }
                             let result = tokio::time::timeout(
                                 Duration::from_secs(DC_PING_TIMEOUT_SECS),
-                                self.ping_single_dc(
-                                    *upstream_idx,
-                                    upstream_config,
-                                    Some(bind_rr.clone()),
-                                    addr,
-                                ),
-                            )
-                            .await;
+                                self.ping_single_dc(*upstream_idx, upstream_config, Some(bind_rr.clone()), addr)
+                            ).await;
 
                             let ping_result = match result {
                                 Ok(Ok(rtt_ms)) => DcPingResult {
@@ -1400,9 +1205,7 @@ impl UpstreamManager {
                                 v4_results.push(ping_result);
                             }
                         }
-                        Err(_) => {
-                            warn!(dc = %dc_idx, addr = %addr_str, "Invalid dc_overrides address, skipping")
-                        }
+                        Err(_) => warn!(dc = %dc_idx, addr = %addr_str, "Invalid dc_overrides address, skipping"),
                     }
                 }
             }
@@ -1578,8 +1381,12 @@ impl UpstreamManager {
         ipv6_enabled: bool,
         dc_overrides: HashMap<String, Vec<String>>,
     ) {
-        let groups =
-            Self::build_health_check_groups(prefer_ipv6, ipv4_enabled, ipv6_enabled, &dc_overrides);
+        let groups = Self::build_health_check_groups(
+            prefer_ipv6,
+            ipv4_enabled,
+            ipv6_enabled,
+            &dc_overrides,
+        );
         let required_healthy_groups = Self::required_healthy_group_count(groups.len());
         let mut endpoint_rotation: HashMap<(usize, i16, bool), usize> = HashMap::new();
 
@@ -1609,16 +1416,13 @@ impl UpstreamManager {
                     let mut group_ok = false;
                     let mut group_rtt_ms = None;
 
-                    for (is_primary, endpoints) in
-                        [(true, &group.primary), (false, &group.fallback)]
-                    {
+                    for (is_primary, endpoints) in [(true, &group.primary), (false, &group.fallback)] {
                         if endpoints.is_empty() {
                             continue;
                         }
 
                         let rotation_key = (i, group.dc_idx, is_primary);
-                        let start_idx =
-                            *endpoint_rotation.entry(rotation_key).or_insert(0) % endpoints.len();
+                        let start_idx = *endpoint_rotation.entry(rotation_key).or_insert(0) % endpoints.len();
                         let mut next_idx = (start_idx + 1) % endpoints.len();
 
                         for step in 0..endpoints.len() {
@@ -1740,7 +1544,8 @@ impl UpstreamManager {
             return None;
         }
 
-        UpstreamState::dc_array_idx(dc_idx).map(|idx| guard[0].dc_ip_pref[idx])
+        UpstreamState::dc_array_idx(dc_idx)
+            .map(|idx| guard[0].dc_ip_pref[idx])
     }
 
     /// Get preferred DC address based on config preference
@@ -1761,12 +1566,6 @@ impl UpstreamManager {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::sync::Arc;
-
-    use crate::stats::Stats;
-
-    const TEST_SHADOWSOCKS_URL: &str =
-        "ss://2022-blake3-aes-256-gcm:MDEyMzQ1Njc4OTAxMjM0NTY3ODkwMTIzNDU2Nzg5MDE=@127.0.0.1:8388";
 
     #[test]
     fn required_healthy_group_count_applies_three_group_threshold() {
@@ -1797,18 +1596,15 @@ mod tests {
 
         assert!(dc2.primary.iter().all(|addr| addr.is_ipv6()));
         assert!(dc2.fallback.iter().all(|addr| addr.is_ipv4()));
-        assert!(
-            dc2.primary
-                .contains(&"[2001:db8::10]:443".parse::<SocketAddr>().unwrap())
-        );
-        assert!(
-            dc2.fallback
-                .contains(&"203.0.113.10:443".parse::<SocketAddr>().unwrap())
-        );
-        assert!(
-            dc2.fallback
-                .contains(&"203.0.113.11:443".parse::<SocketAddr>().unwrap())
-        );
+        assert!(dc2
+            .primary
+            .contains(&"[2001:db8::10]:443".parse::<SocketAddr>().unwrap()));
+        assert!(dc2
+            .fallback
+            .contains(&"203.0.113.10:443".parse::<SocketAddr>().unwrap()));
+        assert!(dc2
+            .fallback
+            .contains(&"203.0.113.11:443".parse::<SocketAddr>().unwrap()));
     }
 
     #[test]
@@ -1830,14 +1626,12 @@ mod tests {
             .expect("override-only dc group must be present");
 
         assert_eq!(dc9.primary.len(), 2);
-        assert!(
-            dc9.primary
-                .contains(&"198.51.100.1:443".parse::<SocketAddr>().unwrap())
-        );
-        assert!(
-            dc9.primary
-                .contains(&"198.51.100.2:443".parse::<SocketAddr>().unwrap())
-        );
+        assert!(dc9
+            .primary
+            .contains(&"198.51.100.1:443".parse::<SocketAddr>().unwrap()));
+        assert!(dc9
+            .primary
+            .contains(&"198.51.100.2:443".parse::<SocketAddr>().unwrap()));
         assert!(dc9.fallback.is_empty());
     }
 
@@ -1884,36 +1678,4 @@ mod tests {
 
         assert_eq!(bind, None);
     }
-
-    #[test]
-    fn api_snapshot_reports_shadowsocks_as_sanitized_route() {
-        let manager = UpstreamManager::new(
-            vec![UpstreamConfig {
-                upstream_type: UpstreamType::Shadowsocks {
-                    url: TEST_SHADOWSOCKS_URL.to_string(),
-                    interface: None,
-                },
-                weight: 2,
-                enabled: true,
-                scopes: String::new(),
-                selected_scope: String::new(),
-            }],
-            1,
-            100,
-            1000,
-            1,
-            false,
-            Arc::new(Stats::new()),
-        );
-
-        let snapshot = manager.try_api_snapshot().expect("snapshot");
-        assert_eq!(snapshot.summary.configured_total, 1);
-        assert_eq!(snapshot.summary.shadowsocks_total, 1);
-        assert_eq!(snapshot.upstreams.len(), 1);
-        assert_eq!(
-            snapshot.upstreams[0].route_kind,
-            UpstreamRouteKind::Shadowsocks
-        );
-        assert_eq!(snapshot.upstreams[0].address, "127.0.0.1:8388");
-    }
 }