fix(stats): use age-based GC check to avoid evicting fresh metrics

The millis-since-base staleness check (last > now - 180s) saturated the cutoff to 0 early in process life, so metrics stamped at 0ms failed the strict > 0 test and were wrongly evicted by the immediate first GC tick. Switch to age-based (now - last < 180s), equivalent to the original Instant semantics and robust under clock saturation, which fixes the flaky peer_conn_secure_mode_pubkey_and_encryption test.
docs(stats): fix misleading/stale comments and rename test_counter
2026-09-20 03:22:05 +00:00 · 2026-06-26 23:31:56 +08:00 · 2026-06-25 09:16:12 +08:00 · 2026-06-25 00:59:49 +08:00 · 2026-06-25 00:54:26 +08:00 · 2026-06-25 00:51:55 +08:00
25 changed files with 2012 additions and 543 deletions
@@ -129,6 +129,12 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "anes"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"
+
 [[package]]
 name = "anstream"
 version = "0.6.15"
@@ -241,6 +247,16 @@ dependencies = [
 "password-hash",
 ]

+[[package]]
+name = "ariadne"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "36f5e3dca4e09a6f340a61a0e9c7b61e030c69fc27bf29d73218f7e5e3b7638f"
+dependencies = [
+ "unicode-width 0.1.11",
+ "yansi",
+]
+
 [[package]]
 name = "arrayvec"
 version = "0.7.6"
@@ -915,7 +931,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bfcfdc083699101d5a7965e49925975f2f55060f94f9a05e7187be95d530ca59"
 dependencies = [
 "once_cell",
- "proc-macro-crate 3.2.0",
+ "proc-macro-crate 3.5.0",
 "proc-macro2",
 "quote",
 "syn 2.0.117",
@@ -1129,6 +1145,12 @@ dependencies = [
 "toml 0.9.12+spec-1.1.0",
 ]

+[[package]]
+name = "cast"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
+
 [[package]]
 name = "cc"
 version = "1.2.10"
@@ -1238,6 +1260,33 @@ dependencies = [
 "windows-targets 0.52.6",
 ]

+[[package]]
+name = "ciborium"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e"
+dependencies = [
+ "ciborium-io",
+ "ciborium-ll",
+ "serde",
+]
+
+[[package]]
+name = "ciborium-io"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757"
+
+[[package]]
+name = "ciborium-ll"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9"
+dependencies = [
+ "ciborium-io",
+ "half",
+]
+
 [[package]]
 name = "cidr"
 version = "0.3.1"
@@ -1583,6 +1632,42 @@ dependencies = [
 "cfg-if",
 ]

+[[package]]
+name = "criterion"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f"
+dependencies = [
+ "anes",
+ "cast",
+ "ciborium",
+ "clap",
+ "criterion-plot",
+ "is-terminal",
+ "itertools 0.10.5",
+ "num-traits",
+ "once_cell",
+ "oorandom",
+ "plotters",
+ "rayon",
+ "regex",
+ "serde",
+ "serde_derive",
+ "serde_json",
+ "tinytemplate",
+ "walkdir",
+]
+
+[[package]]
+name = "criterion-plot"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1"
+dependencies = [
+ "cast",
+ "itertools 0.10.5",
+]
+
 [[package]]
 name = "critical-section"
 version = "1.2.0"
@@ -2234,6 +2319,7 @@ dependencies = [
 "aes-gcm",
 "anyhow",
 "arc-swap",
+ "ariadne",
 "async-recursion",
 "async-ringbuf",
 "async-stream",
@@ -2255,6 +2341,7 @@ dependencies = [
 "clap_complete",
 "clap_complete_nushell",
 "console-subscriber",
+ "criterion",
 "crossbeam",
 "ctor 0.8.0",
 "dashmap",
@@ -2265,6 +2352,7 @@ dependencies = [
 "derive_builder",
 "derive_more 2.1.1",
 "encoding",
+ "fastant",
 "flume 0.12.0",
 "forwarded-header-value",
 "futures",
@@ -2272,7 +2360,7 @@ dependencies = [
 "gethostname 0.5.0",
 "git-version",
 "globwalk",
- "guarden",
+ "guarden 0.2.0",
 "hickory-client",
 "hickory-proto",
 "hickory-resolver",
@@ -2318,7 +2406,7 @@ dependencies = [
 "prost-reflect-build",
 "prost-wkt-types",
 "quinn",
- "quinn-plaintext",
+ "quinn-proto",
 "quote",
 "rand 0.8.5",
 "rcgen",
@@ -2330,6 +2418,7 @@ dependencies = [
 "rstest",
 "rust-i18n",
 "rustls",
+ "seahash",
 "serde",
 "serde_json",
 "serial_test",
@@ -2457,7 +2546,7 @@ dependencies = [
 "dashmap",
 "easytier",
 "futures",
- "guarden",
+ "guarden 0.1.2",
 "jsonwebtoken",
 "mimalloc",
 "mockall",
@@ -2836,6 +2925,16 @@ dependencies = [
 "pin-project-lite",
 ]

+[[package]]
+name = "fastant"
+version = "0.1.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e825441bfb2d831c47c97d05821552db8832479f44c571b97fededbf0099c07"
+dependencies = [
+ "small_ctor",
+ "web-time",
+]
+
 [[package]]
 name = "fastbloom"
 version = "0.9.0"
@@ -3593,7 +3692,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ca87812d87fa82896df1adfb5c111cdeaae3edb6da028f5df002dcbd7df71454"
 dependencies = [
 "futures",
- "guarden-macros",
+ "guarden-macros 0.1.2",
+ "tokio",
+]
+
+[[package]]
+name = "guarden"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8408903291a7d0cc74169d5de4dd1919a9a402a2f67fcd7df3303ed045fae73"
+dependencies = [
+ "futures-core",
+ "guarden-macros 0.2.0",
 "tokio",
 ]

@@ -3608,6 +3718,18 @@ dependencies = [
 "syn 2.0.117",
 ]

+[[package]]
+name = "guarden-macros"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e0ef28f1077c259f9e7e238e234a78ce18cedbf0251fd2135f5fc23c40e79fe"
+dependencies = [
+ "proc-macro-crate 3.5.0",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.117",
+]
+
 [[package]]
 name = "h2"
 version = "0.4.7"
@@ -4460,6 +4582,17 @@ dependencies = [
 "once_cell",
 ]

+[[package]]
+name = "is-terminal"
+version = "0.4.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
+dependencies = [
+ "hermit-abi",
+ "libc",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "is-wsl"
 version = "0.4.0"
@@ -5578,7 +5711,7 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "680998035259dcfcafe653688bf2aa6d3e2dc05e98be6ab46afb089dc84f1df8"
 dependencies = [
- "proc-macro-crate 3.2.0",
+ "proc-macro-crate 3.5.0",
 "proc-macro2",
 "quote",
 "syn 2.0.117",
@@ -5820,6 +5953,12 @@ dependencies = [
 "portable-atomic",
 ]

+[[package]]
+name = "oorandom"
+version = "11.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
+
 [[package]]
 name = "opaque-debug"
 version = "0.3.1"
@@ -6450,6 +6589,34 @@ dependencies = [
 "time",
 ]

+[[package]]
+name = "plotters"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747"
+dependencies = [
+ "num-traits",
+ "plotters-backend",
+ "plotters-svg",
+ "wasm-bindgen",
+ "web-sys",
+]
+
+[[package]]
+name = "plotters-backend"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a"
+
+[[package]]
+name = "plotters-svg"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670"
+dependencies = [
+ "plotters-backend",
+]
+
 [[package]]
 name = "pnet"
 version = "0.35.0"
@@ -6710,11 +6877,11 @@ dependencies = [

 [[package]]
 name = "proc-macro-crate"
-version = "3.2.0"
+version = "3.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ecf48c7ca261d60b74ab1a7b20da18bede46776b2e55535cb958eb595c5fa7b"
+checksum = "e67ba7e9b2b56446f1d419b1d807906278ffa1a658a8a5d8a39dcb1f5a78614f"
 dependencies = [
- "toml_edit 0.22.20",
+ "toml_edit 0.25.12+spec-1.1.0",
 ]

 [[package]]
@@ -7022,18 +7189,6 @@ dependencies = [
 "web-time",
 ]

-[[package]]
-name = "quinn-plaintext"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3e617feaeb6493018fa35fc47ae8b630ac8903d8159e9e747018841b99bad3d"
-dependencies = [
- "bytes",
- "quinn-proto",
- "seahash",
- "tracing",
-]
-
 [[package]]
 name = "quinn-proto"
 version = "0.11.12"
@@ -7608,7 +7763,7 @@ checksum = "1f168d99749d307be9de54d23fd226628d99768225ef08f6ffb52e0182a27746"
 dependencies = [
 "cfg-if",
 "glob",
- "proc-macro-crate 3.2.0",
+ "proc-macro-crate 3.5.0",
 "proc-macro2",
 "quote",
 "regex",
@@ -8655,6 +8810,12 @@ dependencies = [
 "autocfg",
 ]

+[[package]]
+name = "small_ctor"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "88414a5ca1f85d82cc34471e975f0f74f6aa54c40f062efa42c0080e7f763f81"
+
 [[package]]
 name = "smallvec"
 version = "1.13.2"
@@ -9813,6 +9974,16 @@ dependencies = [
 "zerovec",
 ]

+[[package]]
+name = "tinytemplate"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc"
+dependencies = [
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "tinyvec"
 version = "1.8.0"
@@ -9932,8 +10103,7 @@ dependencies = [
 [[package]]
 name = "tokio-websockets"
 version = "0.13.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dad543404f98bfc969aeb71994105c592acfc6c43323fddcd016bb208d1c65cb"
+source = "git+https://github.com/EasyTier/tokio-websockets#dc9771c7c215882349c3cb328877550a3593df21"
 dependencies = [
 "base64 0.22.1",
 "bytes",
@@ -10008,6 +10178,15 @@ dependencies = [
 "serde_core",
 ]

+[[package]]
+name = "toml_datetime"
+version = "1.1.1+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7"
+dependencies = [
+ "serde_core",
+]
+
 [[package]]
 name = "toml_edit"
 version = "0.19.15"
@@ -10045,6 +10224,18 @@ dependencies = [
 "winnow 0.6.18",
 ]

+[[package]]
+name = "toml_edit"
+version = "0.25.12+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2153edc6955a6c354fad8f5efd38b6a8769bdccf9fe50f8e1329f81b0baa5d7"
+dependencies = [
+ "indexmap 2.14.0",
+ "toml_datetime 1.1.1+spec-1.1.0",
+ "toml_parser",
+ "winnow 1.0.1",
+]
+
 [[package]]
 name = "toml_parser"
 version = "1.1.2+spec-1.1.0"
@@ -11892,6 +12083,9 @@ name = "winnow"
 version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09dac053f1cd375980747450bfc7250c264eaae0583872e845c0c7cd578872b5"
+dependencies = [
+ "memchr",
+]

 [[package]]
 name = "winreg"
@@ -12273,7 +12467,7 @@ version = "5.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "897e79616e84aac4b2c46e9132a4f63b93105d54fe8c0e8f6bffc21fa8d49222"
 dependencies = [
- "proc-macro-crate 3.2.0",
+ "proc-macro-crate 3.5.0",
 "proc-macro2",
 "quote",
 "syn 2.0.117",
@@ -12510,7 +12704,7 @@ version = "5.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b59b012ebe9c46656f9cc08d8da8b4c726510aef12559da3e5f1bf72780752c"
 dependencies = [
- "proc-macro-crate 3.2.0",
+ "proc-macro-crate 3.5.0",
 "proc-macro2",
 "quote",
 "syn 2.0.117",
@@ -118,11 +118,7 @@ pub(crate) fn set_tun_fd(
        })
 }

-pub(crate) fn get_runtime_snapshot() -> RuntimeAggregateState {
-    get_runtime_snapshot_inner()
-}
-
-pub(crate) fn get_runtime_snapshot_inner() -> RuntimeAggregateState {
+pub(crate) fn collect_runtime_state() -> RuntimeAggregateState {
    let infos = match ASYNC_RUNTIME.block_on(INSTANCE_MANAGER.collect_network_infos()) {
        Ok(infos) => infos,
        Err(err) => {
@@ -3,6 +3,4 @@ mod routing;
 mod socket_server;

 pub(crate) use routing::aggregate_requested_tun_routes;
-pub use socket_server::{
-    set_snapshot_broadcast_enabled, start_local_socket_server, stop_local_socket_server,
-};
+pub use socket_server::{start_local_socket_server, stop_local_socket_server};
@@ -32,6 +32,13 @@ pub(crate) fn send_local_socket_message(
    Ok(())
 }

+fn shrink_clients_if_sparse(clients: &mut Vec<UnixStream>) {
+    let sparse_limit = clients.len().saturating_mul(2).max(4);
+    if clients.capacity() > sparse_limit {
+        clients.shrink_to_fit();
+    }
+}
+
 pub(crate) fn broadcast_local_socket_message(
    clients: &mut Vec<UnixStream>,
    message_type: &str,
@@ -45,6 +52,7 @@ pub(crate) fn broadcast_local_socket_message(
            active_clients.push(client);
        }
    }
+    shrink_clients_if_sparse(&mut active_clients);
    *clients = active_clients;
    delivered
 }
@@ -79,6 +87,7 @@ pub(crate) fn broadcast_local_socket_json_payload_message(
            active_clients.push(client);
        }
    }
+    shrink_clients_if_sparse(&mut active_clients);
    *clients = active_clients;
    delivered
 }
@@ -1,13 +1,20 @@
 use super::protocol::{
    TunRequestPayload, broadcast_local_socket_json_payload_message, broadcast_local_socket_message,
 };
-use crate::INSTANCE_MANAGER;
+use crate::collect_runtime_state_inner;
 use crate::config::repository::kernel_socket_path;
-use crate::get_runtime_snapshot_inner;
 use crate::kernel_bridge::routing::aggregate_tun_routes;
+use crate::runtime::state::runtime_state::{
+    PeerConnInfo as RuntimePeerConnInfo, RuntimeAggregateState, peer_conn_to_view,
+};
+use crate::{ASYNC_RUNTIME, INSTANCE_MANAGER};
 use easytier::common::global_ctx::{EventBusSubscriber, GlobalCtxEvent};
+use easytier::proto::api::instance::ListPeerRequest;
+use easytier::proto::rpc_types::controller::BaseController;
 use once_cell::sync::Lazy;
+use serde::Serialize;
 use std::collections::{HashMap, HashSet};
+use std::hash::Hash;
 use std::io::ErrorKind;
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::PathBuf;
@@ -23,13 +30,75 @@ struct LocalSocketState {
 }

 static LOCAL_SOCKET_STATE: Lazy<Mutex<Option<LocalSocketState>>> = Lazy::new(|| Mutex::new(None));
-static SNAPSHOT_BROADCAST_ENABLED: AtomicBool = AtomicBool::new(true);
 const SOCKET_TICK_INTERVAL: Duration = Duration::from_millis(250);
+const TRAFFIC_STATS_INTERVAL: Duration = Duration::from_secs(1);
+const INSTANCE_POLL_INTERVAL: Duration = Duration::from_secs(1);
 const TUN_FAST_CHECK_WINDOW: Duration = Duration::from_secs(8);
 const EVENT_RECEIVER_SYNC_INTERVAL: Duration = Duration::from_secs(1);

-pub fn set_snapshot_broadcast_enabled(enabled: bool) {
-    SNAPSHOT_BROADCAST_ENABLED.store(enabled, Ordering::Relaxed);
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct TrafficStatsPayload {
+    instances: Vec<InstanceTrafficStats>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct InstanceTrafficStats {
+    config_id: String,
+    instance_id: String,
+    rx_bytes: i64,
+    tx_bytes: i64,
+    peers: Vec<PeerTrafficStats>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct PeerTrafficStats {
+    peer_id: i64,
+    rx_bytes: i64,
+    tx_bytes: i64,
+    total_bytes: i64,
+    latency_us: i64,
+    loss_rate: f64,
+}
+
+struct PendingPeerEvent {
+    event: &'static str,
+    instance_id: String,
+    peer_id: i64,
+    conn: Option<RuntimePeerConnInfo>,
+}
+
+#[derive(Default)]
+struct DrainedKernelEvents {
+    tun_refresh: bool,
+    topology_lost: bool,
+    peer_events: Vec<PendingPeerEvent>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct RuntimePeerEventPayload {
+    event: &'static str,
+    config_id: String,
+    instance_id: String,
+    peer_id: i64,
+    conn: Option<RuntimePeerConnInfo>,
+}
+
+fn shrink_hash_map_if_sparse<K: Eq + Hash, V>(map: &mut HashMap<K, V>) {
+    let sparse_limit = map.len().saturating_mul(2).max(8);
+    if map.capacity() > sparse_limit {
+        map.shrink_to_fit();
+    }
+}
+
+fn shrink_hash_set_if_sparse<T: Eq + Hash>(set: &mut HashSet<T>) {
+    let sparse_limit = set.len().saturating_mul(2).max(8);
+    if set.capacity() > sparse_limit {
+        set.shrink_to_fit();
+    }
 }

 fn sync_tun_event_receivers(receivers: &mut HashMap<String, EventBusSubscriber>) {
@@ -44,36 +113,67 @@ fn sync_tun_event_receivers(receivers: &mut HashMap<String, EventBusSubscriber>)
        }
    }
    receivers.retain(|instance_id, _| active_instance_ids.contains(instance_id));
+    shrink_hash_map_if_sparse(receivers);
 }

 fn event_needs_tun_refresh(event: &GlobalCtxEvent) -> bool {
    matches!(
        event,
        GlobalCtxEvent::DhcpIpv4Changed(_, _)
-            | GlobalCtxEvent::DhcpIpv4Conflicted(_)
-            | GlobalCtxEvent::PublicIpv6Changed(_, _)
-            | GlobalCtxEvent::PublicIpv6RoutesUpdated(_, _)
            | GlobalCtxEvent::ProxyCidrsUpdated(_, _)
-            | GlobalCtxEvent::ConfigPatched(_)
-            | GlobalCtxEvent::PeerAdded(_)
-            | GlobalCtxEvent::PeerRemoved(_)
-            | GlobalCtxEvent::PeerConnAdded(_)
-            | GlobalCtxEvent::PeerConnRemoved(_)
+            | GlobalCtxEvent::PublicIpv6RoutesUpdated(_, _)
    )
 }

-fn drain_tun_refresh_events(receivers: &mut HashMap<String, EventBusSubscriber>) -> bool {
-    let mut refresh_needed = false;
+fn drain_kernel_events(receivers: &mut HashMap<String, EventBusSubscriber>) -> DrainedKernelEvents {
+    let mut drained = DrainedKernelEvents::default();
    let mut closed_receivers = Vec::new();
    for (instance_id, receiver) in receivers.iter_mut() {
        loop {
            match receiver.try_recv() {
                Ok(event) => {
-                    refresh_needed = event_needs_tun_refresh(&event) || refresh_needed;
+                    drained.tun_refresh = event_needs_tun_refresh(&event) || drained.tun_refresh;
+                    match event {
+                        GlobalCtxEvent::PeerAdded(peer_id) => {
+                            drained.peer_events.push(PendingPeerEvent {
+                                event: "peer_added",
+                                instance_id: instance_id.clone(),
+                                peer_id: peer_id as i64,
+                                conn: None,
+                            });
+                        }
+                        GlobalCtxEvent::PeerRemoved(peer_id) => {
+                            drained.peer_events.push(PendingPeerEvent {
+                                event: "peer_removed",
+                                instance_id: instance_id.clone(),
+                                peer_id: peer_id as i64,
+                                conn: None,
+                            });
+                        }
+                        GlobalCtxEvent::PeerConnAdded(conn_info) => {
+                            let peer_id = conn_info.peer_id as i64;
+                            drained.peer_events.push(PendingPeerEvent {
+                                event: "peer_conn_added",
+                                instance_id: instance_id.clone(),
+                                peer_id,
+                                conn: Some(peer_conn_to_view(conn_info)),
+                            });
+                        }
+                        GlobalCtxEvent::PeerConnRemoved(conn_info) => {
+                            let peer_id = conn_info.peer_id as i64;
+                            drained.peer_events.push(PendingPeerEvent {
+                                event: "peer_conn_removed",
+                                instance_id: instance_id.clone(),
+                                peer_id,
+                                conn: Some(peer_conn_to_view(conn_info)),
+                            });
+                        }
+                        _ => {}
+                    }
                }
                Err(tokio::sync::broadcast::error::TryRecvError::Empty) => break,
                Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) => {
-                    refresh_needed = true;
+                    drained.topology_lost = true;
                    continue;
                }
                Err(tokio::sync::broadcast::error::TryRecvError::Closed) => {
@@ -86,7 +186,124 @@ fn drain_tun_refresh_events(receivers: &mut HashMap<String, EventBusSubscriber>)
    for instance_id in closed_receivers {
        receivers.remove(&instance_id);
    }
-    refresh_needed
+    drained
+}
+
+fn broadcast_runtime_peer_events(
+    clients: &mut Vec<UnixStream>,
+    peer_events: Vec<PendingPeerEvent>,
+) {
+    for event in peer_events {
+        let payload = RuntimePeerEventPayload {
+            event: event.event,
+            config_id: event.instance_id.clone(),
+            instance_id: event.instance_id,
+            peer_id: event.peer_id,
+            conn: event.conn,
+        };
+        match serde_json::to_string(&payload) {
+            Ok(json) => {
+                let _ = broadcast_local_socket_json_payload_message(
+                    clients,
+                    "runtime_peer_event",
+                    &json,
+                );
+            }
+            Err(err) => {
+                ohrs_log_error!("[Rust] serialize runtime peer event failed: {}", err);
+            }
+        }
+    }
+}
+
+fn tun_candidate_ids(snapshot: &RuntimeAggregateState) -> HashSet<String> {
+    snapshot
+        .instances
+        .iter()
+        .filter(|instance| instance.running && instance.tun_required)
+        .map(|instance| instance.instance_id.clone())
+        .collect()
+}
+
+fn collect_traffic_stats() -> TrafficStatsPayload {
+    let services = INSTANCE_MANAGER
+        .iter()
+        .filter_map(|instance| {
+            instance
+                .value()
+                .get_api_service()
+                .map(|api_service| (instance.key().to_string(), api_service))
+        })
+        .collect::<Vec<_>>();
+
+    let instances = ASYNC_RUNTIME.block_on(async {
+        let mut instances = Vec::new();
+        for (instance_id, api_service) in services {
+            let peers = match api_service
+                .get_peer_manage_service()
+                .list_peer(BaseController::default(), ListPeerRequest::default())
+                .await
+            {
+                Ok(response) => response.peer_infos,
+                Err(err) => {
+                    ohrs_log_debug!(
+                        "[Rust] collect traffic stats list_peer failed instance={}: {}",
+                        instance_id,
+                        err
+                    );
+                    continue;
+                }
+            };
+
+            let mut instance_rx_bytes = 0i64;
+            let mut instance_tx_bytes = 0i64;
+            let mut peer_stats = Vec::with_capacity(peers.len());
+
+            for peer in peers {
+                let mut peer_rx_bytes = 0i64;
+                let mut peer_tx_bytes = 0i64;
+                let mut latency_us = i64::MAX;
+                let mut loss_rate = 0f64;
+
+                for conn in peer.conns {
+                    if let Some(stats) = conn.stats {
+                        let rx_bytes = stats.rx_bytes as i64;
+                        let tx_bytes = stats.tx_bytes as i64;
+                        peer_rx_bytes += rx_bytes;
+                        peer_tx_bytes += tx_bytes;
+                        latency_us = latency_us.min(stats.latency_us as i64);
+                    }
+                    loss_rate = loss_rate.max(conn.loss_rate as f64);
+                }
+
+                instance_rx_bytes += peer_rx_bytes;
+                instance_tx_bytes += peer_tx_bytes;
+                peer_stats.push(PeerTrafficStats {
+                    peer_id: peer.peer_id as i64,
+                    rx_bytes: peer_rx_bytes,
+                    tx_bytes: peer_tx_bytes,
+                    total_bytes: peer_rx_bytes + peer_tx_bytes,
+                    latency_us: if latency_us == i64::MAX {
+                        -1
+                    } else {
+                        latency_us
+                    },
+                    loss_rate,
+                });
+            }
+
+            instances.push(InstanceTrafficStats {
+                config_id: instance_id.clone(),
+                instance_id,
+                rx_bytes: instance_rx_bytes,
+                tx_bytes: instance_tx_bytes,
+                peers: peer_stats,
+            });
+        }
+        instances
+    });
+
+    TrafficStatsPayload { instances }
 }

 pub fn start_local_socket_server() -> bool {
@@ -131,21 +348,25 @@ pub fn start_local_socket_server() -> bool {
    let stop_flag = std::sync::Arc::new(AtomicBool::new(false));
    let worker_stop_flag = stop_flag.clone();
    let worker = thread::spawn(move || {
-        let mut last_snapshot_json = String::new();
+        let mut last_topology_json = String::new();
        let mut delivered_tun_requests = HashSet::new();
        let mut last_tun_route_signatures = HashMap::<String, String>::new();
        let mut tun_fast_until = Instant::now() + TUN_FAST_CHECK_WINDOW;
        let mut tun_bootstrap_done = false;
        let mut last_event_receiver_sync_at: Option<Instant> = None;
+        let mut last_traffic_stats_at: Option<Instant> = None;
+        let mut last_instance_poll_at: Option<Instant> = None;
        let mut tun_event_receivers = HashMap::<String, EventBusSubscriber>::new();
        let mut clients = Vec::<UnixStream>::new();

        while !worker_stop_flag.load(Ordering::Relaxed) {
+            let mut full_topology_dirty = false;
            let mut accepted_client = false;
            loop {
                match listener.accept() {
                    Ok((stream, _addr)) => {
                        accepted_client = true;
+                        full_topology_dirty = true;
                        clients.push(stream);
                        tun_fast_until = Instant::now() + TUN_FAST_CHECK_WINDOW;
                        tun_bootstrap_done = false;
@@ -158,15 +379,21 @@ pub fn start_local_socket_server() -> bool {
                }
            }

-            let snapshot_enabled = SNAPSHOT_BROADCAST_ENABLED.load(Ordering::Relaxed);
            if clients.is_empty() {
-                if !last_snapshot_json.is_empty() {
-                    last_snapshot_json.clear();
+                if !last_topology_json.is_empty() {
+                    last_topology_json.clear();
+                    last_topology_json.shrink_to_fit();
                }
                delivered_tun_requests.clear();
+                shrink_hash_set_if_sparse(&mut delivered_tun_requests);
                last_tun_route_signatures.clear();
+                shrink_hash_map_if_sparse(&mut last_tun_route_signatures);
                tun_event_receivers.clear();
+                shrink_hash_map_if_sparse(&mut tun_event_receivers);
+                clients.shrink_to_fit();
                last_event_receiver_sync_at = None;
+                last_traffic_stats_at = None;
+                last_instance_poll_at = None;
                tun_bootstrap_done = false;
                thread::sleep(SOCKET_TICK_INTERVAL);
                continue;
@@ -181,115 +408,143 @@ pub fn start_local_socket_server() -> bool {
                sync_tun_event_receivers(&mut tun_event_receivers);
                last_event_receiver_sync_at = Some(now);
            }
-            if drain_tun_refresh_events(&mut tun_event_receivers) {
+            let drained_events = drain_kernel_events(&mut tun_event_receivers);
+            let tun_refresh = drained_events.tun_refresh;
+            let topology_lost = drained_events.topology_lost;
+            let peer_events = drained_events.peer_events;
+            if topology_lost {
+                full_topology_dirty = true;
+            }
+            if tun_refresh {
                tun_bootstrap_done = false;
                tun_fast_until = now + TUN_FAST_CHECK_WINDOW;
            }
-            let should_collect_snapshot = snapshot_enabled
-                || accepted_client
-                || (!tun_bootstrap_done && now < tun_fast_until);
-            if !should_collect_snapshot {
-                if !last_snapshot_json.is_empty() {
-                    last_snapshot_json.clear();
+            if !peer_events.is_empty() {
+                broadcast_runtime_peer_events(&mut clients, peer_events);
+            }
+            let should_collect_traffic_stats = last_traffic_stats_at
+                .map(|last| now.duration_since(last) >= TRAFFIC_STATS_INTERVAL)
+                .unwrap_or(true);
+            if should_collect_traffic_stats {
+                last_traffic_stats_at = Some(now);
+                match serde_json::to_string(&collect_traffic_stats()) {
+                    Ok(json) => {
+                        let _ = broadcast_local_socket_json_payload_message(
+                            &mut clients,
+                            "traffic_stats",
+                            &json,
+                        );
+                    }
+                    Err(err) => {
+                        ohrs_log_error!("[Rust] serialize traffic stats failed: {}", err);
+                    }
                }
+            }
+            let should_poll_instance = last_instance_poll_at
+                .map(|last| now.duration_since(last) >= INSTANCE_POLL_INTERVAL)
+                .unwrap_or(true);
+            let should_collect_topology = accepted_client
+                || full_topology_dirty
+                || tun_refresh
+                || should_poll_instance
+                || (!tun_bootstrap_done && now < tun_fast_until);
+            if !should_collect_topology {
                thread::sleep(SOCKET_TICK_INTERVAL);
                continue;
            }

-            let snapshot = get_runtime_snapshot_inner();
-            if snapshot_enabled {
-                let snapshot_json = match serde_json::to_string(&snapshot) {
-                    Ok(json) => json,
-                    Err(err) => {
-                        ohrs_log_error!("[Rust] serialize runtime snapshot failed: {}", err);
-                        thread::sleep(SOCKET_TICK_INTERVAL);
-                        continue;
+            let snapshot = collect_runtime_state_inner();
+            last_instance_poll_at = Some(now);
+            match serde_json::to_string(&snapshot) {
+                Ok(json) => {
+                    if accepted_client || full_topology_dirty || json != last_topology_json {
+                        let _ = broadcast_local_socket_json_payload_message(
+                            &mut clients,
+                            "runtime_topology",
+                            &json,
+                        );
+                        last_topology_json = json;
                    }
-                };
-
-                if accepted_client || snapshot_json != last_snapshot_json {
-                    let _ = broadcast_local_socket_json_payload_message(
-                        &mut clients,
-                        "runtime_snapshot",
-                        &snapshot_json,
-                    );
-                    last_snapshot_json = snapshot_json;
                }
-            } else if !last_snapshot_json.is_empty() {
-                last_snapshot_json.clear();
+                Err(err) => {
+                    ohrs_log_error!("[Rust] serialize runtime topology failed: {}", err);
+                }
            }

+            let active_tun_candidate_ids = tun_candidate_ids(&snapshot);
+            delivered_tun_requests
+                .retain(|instance_id| active_tun_candidate_ids.contains(instance_id));
+            last_tun_route_signatures
+                .retain(|instance_id, _| active_tun_candidate_ids.contains(instance_id));
+            shrink_hash_set_if_sparse(&mut delivered_tun_requests);
+            shrink_hash_map_if_sparse(&mut last_tun_route_signatures);
            let mut saw_running_instance = false;
            let mut saw_tun_candidate = false;
            for instance in snapshot.instances.iter() {
                if instance.running {
                    saw_running_instance = true;
                }
-                if instance.running && instance.tun_required {
-                    saw_tun_candidate = true;
-                    let virtual_ipv4 = instance
-                        .my_node_info
-                        .as_ref()
-                        .and_then(|info| info.virtual_ipv4.clone());
-                    let virtual_ipv4_cidr = instance
-                        .my_node_info
-                        .as_ref()
-                        .and_then(|info| info.virtual_ipv4_cidr.clone());
-                    if clients.is_empty() {
+                if !(instance.running && instance.tun_required) {
+                    continue;
+                }
+
+                saw_tun_candidate = true;
+                let virtual_ipv4 = instance
+                    .my_node_info
+                    .as_ref()
+                    .and_then(|info| info.virtual_ipv4.clone());
+                let virtual_ipv4_cidr = instance
+                    .my_node_info
+                    .as_ref()
+                    .and_then(|info| info.virtual_ipv4_cidr.clone());
+                if clients.is_empty() {
+                    continue;
+                }
+                if virtual_ipv4.is_none() || virtual_ipv4_cidr.is_none() {
+                    continue;
+                }
+                let aggregated_routes = aggregate_tun_routes(instance);
+                let route_signature = serde_json::to_string(&(
+                    &virtual_ipv4,
+                    &virtual_ipv4_cidr,
+                    &aggregated_routes,
+                    instance.magic_dns_enabled,
+                    instance.need_exit_node,
+                ))
+                .unwrap_or_else(|_| "[]".to_string());
+                let should_send = !delivered_tun_requests.contains(&instance.instance_id)
+                    || last_tun_route_signatures
+                        .get(&instance.instance_id)
+                        .map(|value| value != &route_signature)
+                        .unwrap_or(true);
+                if !should_send {
+                    continue;
+                }
+                let payload = TunRequestPayload {
+                    config_id: instance.config_id.clone(),
+                    instance_id: instance.instance_id.clone(),
+                    display_name: instance.display_name.clone(),
+                    virtual_ipv4,
+                    virtual_ipv4_cidr,
+                    aggregated_routes,
+                    magic_dns_enabled: instance.magic_dns_enabled,
+                    need_exit_node: instance.need_exit_node,
+                };
+                let payload_json = match serde_json::to_string(&payload) {
+                    Ok(json) => json,
+                    Err(err) => {
+                        ohrs_log_error!("[Rust] serialize tun request failed: {}", err);
                        continue;
                    }
-                    if virtual_ipv4.is_none() || virtual_ipv4_cidr.is_none() {
-                        continue;
-                    }
-                    let aggregated_routes = aggregate_tun_routes(instance);
-                    let route_signature = serde_json::to_string(&(
-                        &virtual_ipv4,
-                        &virtual_ipv4_cidr,
-                        &aggregated_routes,
-                        instance.magic_dns_enabled,
-                        instance.need_exit_node,
-                    ))
-                    .unwrap_or_else(|_| "[]".to_string());
-                    let should_send = accepted_client
-                        || !delivered_tun_requests.contains(&instance.instance_id)
-                        || last_tun_route_signatures
-                            .get(&instance.instance_id)
-                            .map(|value| value != &route_signature)
-                            .unwrap_or(true);
-                    if !should_send {
-                        continue;
-                    }
-                    let payload = TunRequestPayload {
-                        config_id: instance.config_id.clone(),
-                        instance_id: instance.instance_id.clone(),
-                        display_name: instance.display_name.clone(),
-                        virtual_ipv4,
-                        virtual_ipv4_cidr,
-                        aggregated_routes,
-                        magic_dns_enabled: instance.magic_dns_enabled,
-                        need_exit_node: instance.need_exit_node,
-                    };
-                    let payload_json = match serde_json::to_string(&payload) {
-                        Ok(json) => json,
-                        Err(err) => {
-                            ohrs_log_error!("[Rust] serialize tun request failed: {}", err);
-                            continue;
-                        }
-                    };
-                    if broadcast_local_socket_message(&mut clients, "tun_request", &payload_json) {
-                        delivered_tun_requests.insert(instance.instance_id.clone());
-                        last_tun_route_signatures
-                            .insert(instance.instance_id.clone(), route_signature);
-                    }
-                } else {
-                    delivered_tun_requests.remove(&instance.instance_id);
-                    last_tun_route_signatures.remove(&instance.instance_id);
+                };
+                if broadcast_local_socket_message(&mut clients, "tun_request", &payload_json) {
+                    delivered_tun_requests.insert(instance.instance_id.clone());
+                    last_tun_route_signatures.insert(instance.instance_id.clone(), route_signature);
                }
            }
-            if !snapshot_enabled
-                && (!delivered_tun_requests.is_empty()
-                    || (saw_running_instance && !saw_tun_candidate)
-                    || now >= tun_fast_until)
+            if !delivered_tun_requests.is_empty()
+                || (saw_running_instance && !saw_tun_candidate)
+                || now >= tun_fast_until
            {
                tun_bootstrap_done = true;
            }
@@ -63,7 +63,7 @@ use easytier::proto::api::manage::NetworkConfig;
 use easytier::proto::api::manage::NetworkingMethod;
 use easytier::web_client::{WebClient, WebClientHooks, run_web_client};
 use kernel_bridge::{
-    set_snapshot_broadcast_enabled, start_local_socket_server as start_local_socket_server_inner,
+    start_local_socket_server as start_local_socket_server_inner,
    stop_local_socket_server as stop_local_socket_server_inner,
 };
 use napi_derive_ohos::napi;
@@ -517,18 +517,8 @@ mod tests {
    }
 }

-#[napi]
-pub fn get_runtime_snapshot() -> RuntimeAggregateState {
-    exports::runtime_api::get_runtime_snapshot()
-}
-
-#[napi]
-pub fn set_kernel_snapshot_enabled(enabled: bool) {
-    set_snapshot_broadcast_enabled(enabled);
-}
-
-pub(crate) fn get_runtime_snapshot_inner() -> RuntimeAggregateState {
-    exports::runtime_api::get_runtime_snapshot_inner()
+pub(crate) fn collect_runtime_state_inner() -> RuntimeAggregateState {
+    exports::runtime_api::collect_runtime_state()
 }

 #[napi]
@@ -324,7 +324,7 @@ fn route_to_view(route: api::instance::Route) -> RouteView {
    }
 }

-fn peer_conn_to_view(conn: api::instance::PeerConnInfo) -> PeerConnInfo {
+pub(crate) fn peer_conn_to_view(conn: api::instance::PeerConnInfo) -> PeerConnInfo {
    let stats = conn.stats.map(|stats| PeerConnStats {
        rx_bytes: stats.rx_bytes as i64,
        tx_bytes: stats.tx_bytes as i64,
@@ -51,7 +51,7 @@ time = "0.3"
 toml = "0.8.12"
 chrono = { version = "0.4.37", features = ["serde"] }

-guarden = "0.1"
+guarden = "0.2"

 delegate = "0.13.5"

@@ -82,7 +82,8 @@ pin-project-lite = "0.2.13"
 atomic_refcell = "0.1.13"

 quinn = { version = "0.11.8", optional = true, features = ["ring"] }
-quinn-plaintext = { version = "0.3.0", optional = true }
+quinn-proto = { version = "0.11.12", optional = true }
+seahash = { version = "4.1.0", optional = true }

 rustls = { version = "0.23.0", features = [
    "ring", "tls12"
@@ -90,7 +91,7 @@ rustls = { version = "0.23.0", features = [
 rcgen = { version = "0.12.1", optional = true }

 # for websocket
-tokio-websockets = { version = "0.13.2", optional = true, features = [
+tokio-websockets = { version = "0.13.2", git = "https://github.com/EasyTier/tokio-websockets", optional = true, features = [
    "rustls-webpki-roots",
    "client",
    "server",
@@ -133,6 +134,7 @@ prost-wkt-types = "0.7.1"
 pbjson = "0.9.0"

 anyhow = "1.0"
+ariadne = "0.5"

 url = { version = "2.5", features = ["serde"] }
 percent-encoding = "2.3.1"
@@ -213,6 +215,7 @@ smoltcp = { git = "https://github.com/smoltcp-rs/smoltcp.git", rev = "0a926767a6
    "async",
 ] }
 parking_lot = { version = "0.12.0" }
+fastant = "0.1"

 wildmatch = "2.3.4"

@@ -339,6 +342,11 @@ futures-util = "0.3.31"
 maplit = "1.0.2"
 tempfile = "3.22.0"
 ctor = "0.8.0"
+criterion = { version = "0.5", features = ["html_reports"] }
+
+[[bench]]
+name = "counter_contention"
+harness = false

 [target.'cfg(target_os = "linux")'.dev-dependencies]
 defguard_wireguard_rs = "0.4.2"
@@ -373,7 +381,7 @@ full = [
    "zstd",
 ]
 wireguard = ["dep:boringtun", "dep:ring"]
-quic = ["dep:quinn", "dep:quinn-plaintext", "dep:rustls", "dep:rcgen"]
+quic = ["dep:quinn", "dep:quinn-proto", "dep:seahash", "dep:rustls", "dep:rcgen"]
 kcp = ["dep:kcp-sys"]
 mimalloc = ["dep:mimalloc"]
 aes-gcm = ["dep:aes-gcm"]
@@ -0,0 +1,443 @@
+//! Compare counter implementations under tokio-task contention.
+//!
+//! Groups:
+//!   - `contention_scaling` : N tokio tasks share one counter, total work fixed.
+//!     Variants: `single_atomic`, `cas_saturating`, `sharded_atomic`,
+//!     `thread_local_cell`, `unsafe_cell` (unsound, for reference only).
+//!   - `single_thread_write`: per-`add` cost with no contention (floor cost).
+//!   - `read_cost`          : per-`get()` cost.
+//!   - `counter_handle`     : the REAL production hot path. Measures the actual
+//!     `stats_manager::CounterHandle::add` (single-atomic `fetch_add` + lock-free
+//!     fastant `touch`) against reconstructed baselines:
+//!       * `prod` - real `CounterHandle` (this code's version)
+//!       * `baseline_cas_mutex` - pre-optimization: single `AtomicU64` with `fetch_update` (CAS) + `Mutex<Instant>` touch
+//!       * `baseline_fetchadd_mutex` - `fetch_add` + `Mutex<Instant>` touch (isolates the lock-free fastant touch)
+//!
+//! Run: `cargo bench -p easytier --bench counter_contention`
+
+use std::cell::{Cell, UnsafeCell};
+use std::hint::black_box;
+use std::sync::Arc;
+use std::sync::LazyLock;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::thread::available_parallelism;
+use std::time::Instant;
+
+use criterion::{
+    BenchmarkId, Criterion, Throughput, criterion_group, criterion_main, measurement::WallTime,
+};
+use easytier::common::stats_manager::{CounterHandle, MetricName, StatsManager};
+use parking_lot::Mutex;
+
+const COUNTER_SHARDS: usize = 16;
+const TOTAL_WORK: u64 = 8_000_000;
+// The handle path does a counter update plus a timestamp `touch` per `add`,
+// so it is heavier per op than the counter-only groups; use a smaller total to
+// keep the bench fast.
+const HANDLE_TOTAL_WORK: u64 = 2_000_000;
+const TASK_COUNTS: &[usize] = &[1, 2, 4, 8, 16, 32];
+
+trait Counter: Send + Sync {
+    fn add(&self, delta: u64);
+    fn get(&self) -> u64;
+}
+
+// ---------------------------------------------------------------------------
+// 1. SingleAtomic: one atomic, fetch_add. Baseline; contends across cores.
+// ---------------------------------------------------------------------------
+
+struct SingleAtomic(AtomicU64);
+
+impl Default for SingleAtomic {
+    fn default() -> Self {
+        Self(AtomicU64::new(0))
+    }
+}
+
+impl Counter for SingleAtomic {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        self.0.fetch_add(delta, Ordering::Relaxed);
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        self.0.load(Ordering::Relaxed)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 2. CasSaturating: fetch_update with saturating_add (the original PR `add`).
+//    A CAS loop that can retry under contention.
+// ---------------------------------------------------------------------------
+
+struct CasSaturating(AtomicU64);
+
+impl Default for CasSaturating {
+    fn default() -> Self {
+        Self(AtomicU64::new(0))
+    }
+}
+
+impl Counter for CasSaturating {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        let _ = self
+            .0
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| {
+                Some(c.saturating_add(delta))
+            });
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        self.0.load(Ordering::Relaxed)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 3. ShardedAtomic: 16 cache-aligned shards + per-thread shard index.
+//    Comparison-only variant (production `stats_manager::Counter` is
+//    single-atomic; sharding was evaluated and dropped as no benefit for the
+//    default 1-16 worker deployments).
+// ---------------------------------------------------------------------------
+
+thread_local! {
+    static SHARD_IDX: Cell<usize> = Cell::new({
+        static NEXT: AtomicUsize = AtomicUsize::new(0);
+        NEXT.fetch_add(1, Ordering::Relaxed) % COUNTER_SHARDS
+    });
+}
+
+#[repr(align(64))]
+struct Shard {
+    value: AtomicU64,
+}
+
+struct ShardedAtomic {
+    shards: Box<[Shard]>,
+}
+
+impl Default for ShardedAtomic {
+    fn default() -> Self {
+        let mut shards = Vec::with_capacity(COUNTER_SHARDS);
+        for _ in 0..COUNTER_SHARDS {
+            shards.push(Shard {
+                value: AtomicU64::new(0),
+            });
+        }
+        Self {
+            shards: shards.into_boxed_slice(),
+        }
+    }
+}
+
+impl Counter for ShardedAtomic {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        let i = SHARD_IDX.with(|c| c.get());
+        self.shards[i].value.fetch_add(delta, Ordering::Relaxed);
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        self.shards
+            .iter()
+            .map(|s| s.value.load(Ordering::Relaxed))
+            .sum()
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 4. ThreadLocalCell: per-thread Cell<u64> accumulation. Zero-atomic writes.
+//    `get()` flushes the caller thread's local into a shared aggregate, so the
+//    measured read cost reflects a flush-based read. Exact totals would require
+//    flushing every thread (not modeled here).
+// ---------------------------------------------------------------------------
+
+thread_local! {
+    static TLS_DELTA: Cell<u64> = const { Cell::new(0) };
+}
+
+struct ThreadLocalCell {
+    shared: AtomicU64,
+}
+
+impl Default for ThreadLocalCell {
+    fn default() -> Self {
+        Self {
+            shared: AtomicU64::new(0),
+        }
+    }
+}
+
+impl Counter for ThreadLocalCell {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        TLS_DELTA.with(|c| c.set(c.get() + delta));
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        let local = TLS_DELTA.with(|c| c.replace(0));
+        self.shared.fetch_add(local, Ordering::Relaxed) + local
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 5. UnsafeCellCounter: a plain u64 mutated through UnsafeCell with manual
+//    `unsafe impl Send/Sync`. This is UNSOUND under concurrent access (data
+//    race / UB) and is exactly what the original code did "for speed". It is
+//    included only to measure the speed ceiling the author was chasing, and to
+//    show that its `get()` returns wrong totals under contention (lost updates).
+// ---------------------------------------------------------------------------
+
+struct UnsafeCellCounter(UnsafeCell<u64>);
+
+// SAFETY: deliberately unsound; see above.
+unsafe impl Send for UnsafeCellCounter {}
+unsafe impl Sync for UnsafeCellCounter {}
+
+impl Default for UnsafeCellCounter {
+    fn default() -> Self {
+        Self(UnsafeCell::new(0))
+    }
+}
+
+impl Counter for UnsafeCellCounter {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        // SAFETY: UNSOUND under concurrent access (data race).
+        unsafe {
+            *self.0.get() += delta;
+        }
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        // SAFETY: UNSOUND under concurrent writers (data race).
+        unsafe { *self.0.get() }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Production counter handle + reconstructed baselines for the `counter_handle`
+// group. These measure the full hot path (`add` = counter update + `touch`
+// timestamp), which is what actually runs per packet in `peer_manager`.
+// ---------------------------------------------------------------------------
+
+// The real production `CounterHandle`. `CounterHandle::add` does a single
+// `AtomicU64::fetch_add` then a lock-free fastant `touch`.
+impl Counter for CounterHandle {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        // Fully-qualified to avoid infinite recursion through the trait method.
+        CounterHandle::add(self, delta);
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        CounterHandle::get(self)
+    }
+}
+
+// A faithful replica of the pre-optimization design: a SINGLE `AtomicU64`
+// (unsharded) plus a `Mutex<Instant>` timestamp. `use_cas` selects whether the
+// counter write is a `fetch_update` saturating CAS (the original PR `add`) or a
+// plain `fetch_add`.
+struct BaselineHandle {
+    counter: AtomicU64,
+    last_updated: Mutex<Instant>,
+    use_cas: bool,
+}
+
+impl Default for BaselineHandle {
+    fn default() -> Self {
+        Self {
+            counter: AtomicU64::new(0),
+            last_updated: Mutex::new(Instant::now()),
+            use_cas: false,
+        }
+    }
+}
+
+impl Counter for BaselineHandle {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        if self.use_cas {
+            let _ = self
+                .counter
+                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| {
+                    Some(c.saturating_add(delta))
+                });
+        } else {
+            self.counter.fetch_add(delta, Ordering::Relaxed);
+        }
+        *self.last_updated.lock() = Instant::now();
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        self.counter.load(Ordering::Relaxed)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Harness: a shared multi-thread tokio runtime sized to the host's parallelism.
+// ---------------------------------------------------------------------------
+
+static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
+    let workers = available_parallelism().map(|n| n.get()).unwrap_or(1);
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(workers)
+        .enable_all()
+        .build()
+        .expect("failed to build tokio runtime")
+});
+
+fn bench_contention<C: Counter + Default + 'static>(
+    group: &mut criterion::BenchmarkGroup<'_, WallTime>,
+    name: &str,
+    n_tasks: usize,
+    per_task: u64,
+) {
+    let counter: Arc<C> = Arc::new(C::default());
+    group.bench_with_input(BenchmarkId::new(name, n_tasks), &n_tasks, |b, &n| {
+        b.iter(|| {
+            let counter = counter.clone();
+            RUNTIME.block_on(async move {
+                let mut handles = Vec::with_capacity(n);
+                for _ in 0..n {
+                    let c = counter.clone();
+                    handles.push(tokio::spawn(async move {
+                        for _ in 0..per_task {
+                            c.add(black_box(1));
+                        }
+                    }));
+                }
+                for handle in handles {
+                    let _ = handle.await;
+                }
+                black_box(counter.get());
+            });
+        });
+    });
+}
+
+fn contention_scaling(c: &mut Criterion) {
+    let mut group = c.benchmark_group("contention_scaling");
+    group.throughput(Throughput::Elements(TOTAL_WORK));
+    for &n in TASK_COUNTS {
+        let per = TOTAL_WORK / n as u64;
+        bench_contention::<SingleAtomic>(&mut group, "single_atomic", n, per);
+        bench_contention::<CasSaturating>(&mut group, "cas_saturating", n, per);
+        bench_contention::<ShardedAtomic>(&mut group, "sharded_atomic", n, per);
+        bench_contention::<ThreadLocalCell>(&mut group, "thread_local_cell", n, per);
+        bench_contention::<UnsafeCellCounter>(&mut group, "unsafe_cell", n, per);
+    }
+    group.finish();
+}
+
+fn single_thread_write<C: Counter + Default>(
+    group: &mut criterion::BenchmarkGroup<'_, WallTime>,
+    name: &str,
+) {
+    let counter = C::default();
+    group.bench_function(name, |b| {
+        b.iter(|| {
+            counter.add(black_box(1));
+        });
+    });
+}
+
+fn single_thread_write_group(c: &mut Criterion) {
+    let mut group = c.benchmark_group("single_thread_write");
+    group.throughput(Throughput::Elements(1));
+    single_thread_write::<SingleAtomic>(&mut group, "single_atomic");
+    single_thread_write::<CasSaturating>(&mut group, "cas_saturating");
+    single_thread_write::<ShardedAtomic>(&mut group, "sharded_atomic");
+    single_thread_write::<ThreadLocalCell>(&mut group, "thread_local_cell");
+    single_thread_write::<UnsafeCellCounter>(&mut group, "unsafe_cell");
+    group.finish();
+}
+
+fn read_cost<C: Counter + Default>(
+    group: &mut criterion::BenchmarkGroup<'_, WallTime>,
+    name: &str,
+) {
+    let counter = C::default();
+    counter.add(1000);
+    group.bench_function(name, |b| {
+        b.iter(|| black_box(counter.get()));
+    });
+}
+
+fn read_cost_group(c: &mut Criterion) {
+    let mut group = c.benchmark_group("read_cost");
+    read_cost::<SingleAtomic>(&mut group, "single_atomic");
+    read_cost::<CasSaturating>(&mut group, "cas_saturating");
+    read_cost::<ShardedAtomic>(&mut group, "sharded_atomic");
+    read_cost::<ThreadLocalCell>(&mut group, "thread_local_cell");
+    read_cost::<UnsafeCellCounter>(&mut group, "unsafe_cell");
+    group.finish();
+}
+
+fn bench_handle(
+    group: &mut criterion::BenchmarkGroup<'_, WallTime>,
+    name: &str,
+    n_tasks: usize,
+    per_task: u64,
+    counter: Arc<dyn Counter>,
+) {
+    group.bench_with_input(BenchmarkId::new(name, n_tasks), &n_tasks, |b, &n| {
+        b.iter(|| {
+            let counter = counter.clone();
+            RUNTIME.block_on(async move {
+                let mut handles = Vec::with_capacity(n);
+                for _ in 0..n {
+                    let c = counter.clone();
+                    handles.push(tokio::spawn(async move {
+                        for _ in 0..per_task {
+                            c.add(black_box(1));
+                        }
+                    }));
+                }
+                for handle in handles {
+                    let _ = handle.await;
+                }
+                black_box(counter.get());
+            });
+        });
+    });
+}
+
+fn counter_handle(c: &mut Criterion) {
+    let mut group = c.benchmark_group("counter_handle");
+    group.throughput(Throughput::Elements(HANDLE_TOTAL_WORK));
+
+    // StatsManager::new() spawns a background cleanup task, which needs a tokio
+    // runtime context; bind it to our shared RUNTIME for the lifetime of the
+    // group.
+    let _rt_guard = RUNTIME.enter();
+    let stats = StatsManager::new();
+
+    let prod: Arc<dyn Counter> = Arc::new(stats.get_simple_counter(MetricName::TrafficBytesTx));
+    let cas: Arc<dyn Counter> = Arc::new(BaselineHandle {
+        use_cas: true,
+        ..Default::default()
+    });
+    let fam: Arc<dyn Counter> = Arc::new(BaselineHandle {
+        use_cas: false,
+        ..Default::default()
+    });
+
+    for &n in TASK_COUNTS {
+        let per = HANDLE_TOTAL_WORK / n as u64;
+        bench_handle(&mut group, "prod", n, per, prod.clone());
+        bench_handle(&mut group, "baseline_cas_mutex", n, per, cas.clone());
+        bench_handle(&mut group, "baseline_fetchadd_mutex", n, per, fam.clone());
+    }
+    group.finish();
+}
+
+// Keep the default measurement config; pass CLI flags to speed up a run, e.g.
+// `-- --measurement-time 2 --sample-size 30 --warm-up-time 500`.
+criterion_group! {
+    name = benches;
+    config = Criterion::default();
+    targets = contention_scaling, single_thread_write_group, read_cost_group, counter_handle
+}
+criterion_main!(benches);
@@ -49,8 +49,8 @@ core_clap:
    en: "manually specify the public IPv6 subnet to share, instead of auto-detecting from system routes"
    zh-CN: "手动指定要共享的公网 IPv6 子网，不自动从系统路由检测"
  dhcp:
-    en: "automatically determine and set IP address by Easytier. The subnet is derived from a connected peer's IPv4 or defaults to 10.126.126.0/24. Warning, if there is an IP conflict in the network when using DHCP, the IP will be automatically changed. Optionally specify a CIDR subnet (e.g. -d 10.0.0.0/24, prefix <= /30) to pin the DHCP address range."
-    zh-CN: "由Easytier自动确定并设置IP地址。子网从已连接对等节点的IPv4派生，或默认使用10.126.126.0/24。警告：在使用DHCP时，如果网络中出现IP冲突，IP将自动更改。可选指定CIDR子网（如 -d 10.0.0.0/24，前缀 <= /30）来固定DHCP地址范围。"
+    en: "automatically determine and set IP address by Easytier, and the IP address starts from 10.0.0.1 by default. Warning, if there is an IP conflict in the network when using DHCP, the IP will be automatically changed."
+    zh-CN: "由Easytier自动确定并设置IP地址，默认从10.0.0.1开始。警告：在使用DHCP时，如果网络中出现IP冲突，IP将自动更改。"
  peers:
    en: "peers to connect initially"
    zh-CN: "最初要连接的对等节点"
@@ -114,11 +114,11 @@ core_clap:
    en: "encryption algorithm to use, supported: '', 'xor', 'chacha20', 'aes-gcm', 'aes-gcm-256', 'openssl-aes128-gcm', 'openssl-aes256-gcm', 'openssl-chacha20'. Empty string means default (aes-gcm)"
    zh-CN: "要使用的加密算法，支持：''（默认aes-gcm）、'xor'、'chacha20'、'aes-gcm'、'aes-gcm-256'、'openssl-aes128-gcm'、'openssl-aes256-gcm'、'openssl-chacha20'"
  multi_thread:
-    en: "use multi-thread runtime, default is single-thread"
-    zh-CN: "使用多线程运行时，默认为单线程"
+    en: "multi-thread tokio runtime (default on). Only affects launcher-based deployments (GUI/mobile/web/Windows service); the easytier-core CLI always runs single-threaded."
+    zh-CN: "多线程 tokio 运行时（默认开启）。仅对 launcher 部署（GUI/移动端/web/Windows 服务）生效；easytier-core CLI 始终为单线程。"
  multi_thread_count:
-    en: "the number of threads to use, default is 2, only effective when multi-thread is enabled, must be greater than 2"
-    zh-CN: "使用的线程数，默认为2，仅在多线程模式下有效。取值必须大于2"
+    en: "the number of worker threads, default 2, only effective when multi-thread is enabled, minimum 2"
+    zh-CN: "worker 线程数，默认 2，仅在启用多线程时生效，最小为 2"
  disable_ipv6:
    en: "do not use ipv6"
    zh-CN: "不使用IPv6"
@@ -6,6 +6,7 @@ use std::{
 };

 use anyhow::Context;
+use ariadne::{CharSet, Config as AriadneConfig, IndexType, Label, Report, ReportKind, Source};
 use base64::{Engine as _, prelude::BASE64_STANDARD};
 use clap::ValueEnum;
 use clap::builder::PossibleValue;
@@ -185,9 +186,6 @@ pub trait ConfigLoader: Send + Sync {
    fn get_dhcp(&self) -> bool;
    fn set_dhcp(&self, dhcp: bool);

-    fn get_dhcp_cidr(&self) -> Option<cidr::Ipv4Cidr>;
-    fn set_dhcp_cidr(&self, cidr: Option<cidr::Ipv4Cidr>);
-
    fn add_proxy_cidr(
        &self,
        cidr: cidr::Ipv4Cidr,
@@ -538,7 +536,6 @@ struct Config {
    ipv6_public_addr_auto: Option<bool>,
    ipv6_public_addr_prefix: Option<String>,
    dhcp: Option<bool>,
-    dhcp_cidr: Option<String>,
    network_identity: Option<NetworkIdentity>,
    listeners: Option<Vec<url::Url>>,
    mapped_listeners: Option<Vec<url::Url>>,
@@ -573,6 +570,35 @@ struct Config {
    source: Option<ConfigSourceConfig>,
 }

+fn format_toml_parse_error(source_name: &str, config_str: &str, error: &toml::de::Error) -> String {
+    let message = format!("failed to parse config TOML from {source_name}");
+
+    let Some(span) = error.span() else {
+        return format!("{message}\ndetail: {error}");
+    };
+
+    let mut output = Vec::new();
+    let report = Report::build(ReportKind::Error, (source_name, span.clone()))
+        .with_config(
+            AriadneConfig::default()
+                .with_color(false)
+                .with_char_set(CharSet::Ascii)
+                .with_index_type(IndexType::Byte),
+        )
+        .with_message(&message)
+        .with_label(Label::new((source_name, span)).with_message(error.message()))
+        .finish();
+
+    if report
+        .write((source_name, Source::from(config_str)), &mut output)
+        .is_ok()
+    {
+        String::from_utf8_lossy(&output).into_owned()
+    } else {
+        format!("{message}\ndetail: {error}")
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct TomlConfigLoader {
    config: Arc<Mutex<Config>>,
@@ -595,12 +621,39 @@ impl TomlConfigLoader {
    }

    pub fn new_from_str(config_str: &str) -> Result<Self, anyhow::Error> {
-        let mut config = toml::de::from_str::<Config>(config_str)
-            .with_context(|| format!("failed to parse config file: {}", config_str))?;
+        Self::new_from_str_with_source("inline config", config_str)
+    }
+
+    pub fn new(config_path: &PathBuf) -> Result<Self, anyhow::Error> {
+        let config_str = std::fs::read_to_string(config_path)
+            .with_context(|| format!("failed to read config file: {}", config_path.display()))?;
+
+        let source_name = config_path.display().to_string();
+        Self::new_from_str_with_source(&source_name, &config_str)
+    }
+
+    pub(crate) fn new_from_str_with_source(
+        source_name: &str,
+        config_str: &str,
+    ) -> Result<Self, anyhow::Error> {
+        let mut config = toml::de::from_str::<Config>(config_str).map_err(|err| {
+            let message = format_toml_parse_error(source_name, config_str, &err);
+            anyhow::Error::new(err).context(message)
+        })?;

        Self::normalize_config_source(&mut config);

-        config.flags_struct = Some(Self::gen_flags(config.flags.clone().unwrap_or_default()));
+        Self::new_from_config(config).map_err(|err| {
+            let message = format!("failed to load config from {source_name}: {err}");
+            err.context(message)
+        })
+    }
+
+    fn new_from_config(mut config: Config) -> Result<Self, anyhow::Error> {
+        config.flags_struct = Some(
+            Self::gen_flags(config.flags.clone().unwrap_or_default())
+                .context("failed to parse flags")?,
+        );
        let has_network_identity = config.network_identity.is_some();

        let config = TomlConfigLoader {
@@ -632,21 +685,15 @@ impl TomlConfigLoader {
        Ok(config)
    }

-    pub fn new(config_path: &PathBuf) -> Result<Self, anyhow::Error> {
-        let config_str = std::fs::read_to_string(config_path)
-            .with_context(|| format!("failed to read config file: {:?}", config_path))?;
-        let ret = Self::new_from_str(&config_str)?;
-
-        Ok(ret)
-    }
-
-    fn gen_flags(flags_hashmap: serde_json::Map<String, serde_json::Value>) -> Flags {
+    fn gen_flags(
+        flags_hashmap: serde_json::Map<String, serde_json::Value>,
+    ) -> serde_json::Result<Flags> {
        let mut merged_hashmap = match serde_json::to_value(gen_default_flags()) {
            Ok(serde_json::Value::Object(map)) => map,
            _ => serde_json::Map::new(),
        };
        merged_hashmap.extend(flags_hashmap);
-        serde_json::from_value(serde_json::Value::Object(merged_hashmap)).unwrap()
+        serde_json::from_value(serde_json::Value::Object(merged_hashmap))
    }
 }

@@ -765,26 +812,13 @@ impl ConfigLoader for TomlConfigLoader {
    }

    fn get_dhcp(&self) -> bool {
-        let config = self.config.lock().unwrap();
-        config.dhcp.unwrap_or_default() || config.dhcp_cidr.is_some()
+        self.config.lock().unwrap().dhcp.unwrap_or_default()
    }

    fn set_dhcp(&self, dhcp: bool) {
        self.config.lock().unwrap().dhcp = Some(dhcp);
    }

-    fn get_dhcp_cidr(&self) -> Option<cidr::Ipv4Cidr> {
-        let locked_config = self.config.lock().unwrap();
-        locked_config
-            .dhcp_cidr
-            .as_ref()
-            .and_then(|s| s.parse().ok())
-    }
-
-    fn set_dhcp_cidr(&self, cidr: Option<cidr::Ipv4Cidr>) {
-        self.config.lock().unwrap().dhcp_cidr = cidr.map(|c| c.to_string());
-    }
-
    fn add_proxy_cidr(
        &self,
        cidr: cidr::Ipv4Cidr,
@@ -1216,13 +1250,13 @@ pub async fn load_config_from_file(
            .read_to_string(&mut stdin)
            .await
            .context("failed to read config from stdin")?;
-        let config = TomlConfigLoader::new_from_str(&stdin)?;
+        let config = TomlConfigLoader::new_from_str_with_source("stdin", &stdin)?;
        return Ok((config, ConfigFileControl::STATIC_CONFIG));
    }

    let config_str = tokio::fs::read_to_string(config_file)
        .await
-        .with_context(|| format!("failed to read config file: {:?}", config_file))?;
+        .with_context(|| format!("failed to read config file: {}", config_file.display()))?;

    let (expanded_config_str, uses_env_vars) = if disable_env_parsing {
        (config_str.clone(), false)
@@ -1244,8 +1278,8 @@ pub async fn load_config_from_file(
        );
    }

-    let config = TomlConfigLoader::new_from_str(&expanded_config_str)
-        .with_context(|| format!("failed to load config file: {:?}", config_file))?;
+    let source_name = config_file.display().to_string();
+    let config = TomlConfigLoader::new_from_str_with_source(&source_name, &expanded_config_str)?;

    let mut control = ConfigFileControl::from_path(config_file.clone()).await;

@@ -1285,6 +1319,96 @@ pub mod tests {
    use std::path::PathBuf;
    use tempfile::NamedTempFile;

+    #[test]
+    fn invalid_toml_error_includes_location_and_source_line() {
+        let error = TomlConfigLoader::new_from_str("dhcp = \"yes\"").unwrap_err();
+        let display = error.to_string();
+
+        assert!(display.contains("failed to parse config TOML"));
+        assert!(display.contains("inline config"));
+        assert!(display.contains("dhcp = \"yes\""));
+        assert!(display.contains("^"));
+        assert!(display.contains("invalid type: string"));
+        assert!(!display.contains("<unknown>"));
+        assert!(
+            error
+                .chain()
+                .any(|err| err.downcast_ref::<toml::de::Error>().is_some())
+        );
+    }
+
+    #[test]
+    fn invalid_file_toml_error_includes_config_source() {
+        let mut config_file = NamedTempFile::new().unwrap();
+        writeln!(config_file, "dhcp = \"yes\"").unwrap();
+
+        let error = TomlConfigLoader::new(&config_file.path().to_path_buf()).unwrap_err();
+        let error = error.to_string();
+
+        assert!(error.contains(config_file.path().to_string_lossy().as_ref()));
+        assert!(error.contains("failed to parse config TOML"));
+        assert!(error.contains("dhcp = \"yes\""));
+        assert!(error.contains("^"));
+        assert!(error.contains("invalid type: string"));
+        assert!(!error.contains("<unknown>"));
+    }
+
+    #[test]
+    fn invalid_stdin_toml_error_includes_config_source_in_display() {
+        let error = TomlConfigLoader::new_from_str_with_source("stdin", "dhcp = \"yes\"")
+            .unwrap_err()
+            .to_string();
+
+        assert!(error.contains("stdin"));
+        assert!(error.contains("failed to parse config TOML"));
+        assert!(error.contains("dhcp = \"yes\""));
+        assert!(error.contains("^"));
+        assert!(error.contains("invalid type: string"));
+        assert!(!error.contains("<unknown>"));
+    }
+
+    #[test]
+    fn invalid_toml_error_handles_non_ascii_before_error() {
+        let error = TomlConfigLoader::new_from_str("hostname = \"节点\"\ndhcp = \"yes\"")
+            .unwrap_err()
+            .to_string();
+
+        assert!(error.contains("dhcp = \"yes\""));
+        assert!(error.contains("^"));
+        assert!(error.contains("invalid type: string"));
+    }
+
+    #[test]
+    fn invalid_toml_error_handles_non_ascii_before_error_on_same_line() {
+        let error = TomlConfigLoader::new_from_str("hostname = \"节点\" dhcp = \"yes\"")
+            .unwrap_err()
+            .to_string();
+
+        assert!(error.contains("failed to parse config TOML"));
+        assert!(error.contains("inline config:1:"));
+        assert!(error.contains("hostname = \"节点\" dhcp = \"yes\""));
+        assert!(error.contains("expected newline"));
+        assert!(!error.contains("<unknown>"));
+    }
+
+    #[test]
+    fn invalid_file_flags_error_includes_config_source_in_display() {
+        let mut config_file = NamedTempFile::new().unwrap();
+        writeln!(config_file, "[flags]").unwrap();
+        writeln!(config_file, "socket_mark = \"bad\"").unwrap();
+
+        let error = TomlConfigLoader::new(&config_file.path().to_path_buf()).unwrap_err();
+
+        let display = error.to_string();
+        assert!(display.contains(config_file.path().to_string_lossy().as_ref()));
+        assert!(display.contains("failed to load config"));
+        assert!(display.contains("failed to parse flags"));
+
+        // with_context preserves the cause chain so callers can inspect the root reason.
+        let chain: Vec<String> = error.chain().map(|e| e.to_string()).collect();
+        assert!(chain.iter().any(|m| m.contains("failed to parse flags")));
+    }
+
    #[test]
    fn socket_mark_config_file_roundtrip_none_some_and_zero() {
        // Omitting the flag leaves socket_mark unset (None) -> SO_MARK untouched.
@@ -1,9 +1,10 @@
 use dashmap::DashMap;
 use serde::{Deserialize, Serialize};
-use std::cell::UnsafeCell;
 use std::fmt;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::sync::OnceLock;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
 use tokio::time::interval;
 use tokio_util::task::AbortOnDropHandle;

@@ -374,136 +375,106 @@ impl Default for LabelSet {
    }
 }

-/// UnsafeCounter provides a high-performance counter using UnsafeCell
+/// Counter provides a high-performance atomic counter
 #[derive(Debug)]
-pub struct UnsafeCounter {
-    value: UnsafeCell<u64>,
+pub struct Counter {
+    value: AtomicU64,
 }

-impl Default for UnsafeCounter {
+impl Default for Counter {
    fn default() -> Self {
        Self::new()
    }
 }

-impl UnsafeCounter {
+impl Counter {
    pub fn new() -> Self {
        Self {
-            value: UnsafeCell::new(0),
+            value: AtomicU64::new(0),
        }
    }

    pub fn new_with_value(initial: u64) -> Self {
        Self {
-            value: UnsafeCell::new(initial),
+            value: AtomicU64::new(initial),
        }
    }

    /// Increment the counter by the given amount
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn add(&self, delta: u64) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = (*ptr).saturating_add(delta);
-        }
+    pub fn add(&self, delta: u64) {
+        self.value.fetch_add(delta, Ordering::Relaxed);
    }

    /// Increment the counter by 1
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn inc(&self) {
-        unsafe {
-            self.add(1);
-        }
+    pub fn inc(&self) {
+        self.add(1);
    }

    /// Get the current value of the counter
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is modifying this counter simultaneously.
-    pub unsafe fn get(&self) -> u64 {
-        let ptr = self.value.get();
-        unsafe { *ptr }
+    pub fn get(&self) -> u64 {
+        self.value.load(Ordering::Relaxed)
    }

    /// Reset the counter to zero
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn reset(&self) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = 0;
-        }
+    pub fn reset(&self) {
+        self.value.store(0, Ordering::Relaxed);
    }

    /// Set the counter to a specific value
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn set(&self, value: u64) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = value;
-        }
+    pub fn set(&self, value: u64) {
+        self.value.store(value, Ordering::Relaxed);
    }
 }

-// UnsafeCounter is Send + Sync because the safety is guaranteed by the caller
-unsafe impl Send for UnsafeCounter {}
-unsafe impl Sync for UnsafeCounter {}
+/// Epoch used to convert a monotonic clock reading into a storable `u64`
+/// millisecond count for `MetricData::last_updated`. Lazily initialized on first
+/// use. Backed by `fastant`, which uses the TSC on x86_64 Linux (and falls back
+/// to `std::time::Instant` elsewhere), making `now_millis()` cheap enough to
+/// call per packet.
+fn time_base() -> fastant::Instant {
+    static BASE: OnceLock<fastant::Instant> = OnceLock::new();
+    *BASE.get_or_init(fastant::Instant::now)
+}
+
+fn now_millis() -> u64 {
+    fastant::Instant::now()
+        .saturating_duration_since(time_base())
+        .as_millis() as u64
+}

 /// MetricData contains both the counter and last update timestamp
-/// Uses UnsafeCell for lock-free access
 #[derive(Debug)]
 struct MetricData {
-    counter: UnsafeCounter,
-    last_updated: UnsafeCell<Instant>,
+    counter: Counter,
+    last_updated: AtomicU64,
 }

 impl MetricData {
    fn new() -> Self {
        Self {
-            counter: UnsafeCounter::new(),
-            last_updated: UnsafeCell::new(Instant::now()),
+            counter: Counter::new(),
+            last_updated: AtomicU64::new(now_millis()),
        }
    }

    fn new_with_value(initial: u64) -> Self {
        Self {
-            counter: UnsafeCounter::new_with_value(initial),
-            last_updated: UnsafeCell::new(Instant::now()),
+            counter: Counter::new_with_value(initial),
+            last_updated: AtomicU64::new(now_millis()),
        }
    }

-    /// Update the last_updated timestamp
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this timestamp simultaneously.
-    unsafe fn touch(&self) {
-        let ptr = self.last_updated.get();
-        unsafe {
-            *ptr = Instant::now();
-        }
+    /// Update the last_updated timestamp. Lock-free.
+    fn touch(&self) {
+        self.last_updated.store(now_millis(), Ordering::Relaxed);
    }

-    /// Get the last updated timestamp
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is modifying this timestamp simultaneously.
-    unsafe fn get_last_updated(&self) -> Instant {
-        let ptr = self.last_updated.get();
-        unsafe { *ptr }
+    /// Last update time as milliseconds since `time_base()`.
+    fn last_updated_millis(&self) -> u64 {
+        self.last_updated.load(Ordering::Relaxed)
    }
 }

-// MetricData is Send + Sync because the safety is guaranteed by the caller
-unsafe impl Send for MetricData {}
-unsafe impl Sync for MetricData {}
-
 /// MetricKey uniquely identifies a metric with its name and labels
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct MetricKey {
@@ -546,39 +517,31 @@ impl CounterHandle {

    /// Increment the counter by the given amount
    pub fn add(&self, delta: u64) {
-        unsafe {
-            self.metric_data.counter.add(delta);
-            self.metric_data.touch();
-        }
+        self.metric_data.counter.add(delta);
+        self.metric_data.touch();
    }

    /// Increment the counter by 1
    pub fn inc(&self) {
-        unsafe {
-            self.metric_data.counter.inc();
-            self.metric_data.touch();
-        }
+        self.metric_data.counter.inc();
+        self.metric_data.touch();
    }

    /// Get the current value of the counter
    pub fn get(&self) -> u64 {
-        unsafe { self.metric_data.counter.get() }
+        self.metric_data.counter.get()
    }

    /// Reset the counter to zero
    pub fn reset(&self) {
-        unsafe {
-            self.metric_data.counter.reset();
-            self.metric_data.touch();
-        }
+        self.metric_data.counter.reset();
+        self.metric_data.touch();
    }

    /// Set the counter to a specific value
    pub fn set(&self, value: u64) {
-        unsafe {
-            self.metric_data.counter.set(value);
-            self.metric_data.touch();
-        }
+        self.metric_data.counter.set(value);
+        self.metric_data.touch();
    }
 }

@@ -614,9 +577,15 @@ impl StatsManager {
            loop {
                interval.tick().await;

-                let Some(cutoff_time) = Instant::now().checked_sub(Duration::from_secs(180)) else {
-                    continue;
-                };
+                // Drop metrics untouched for 180s and with no live handles.
+                // Compare in the millis-since-base domain so neither the hot
+                // path nor GC reconstructs an `Instant` or locks.
+                //
+                // Use an age-based check (`now - last < STALE`) rather than
+                // `last > now - STALE`: early in process life `now_millis()` is
+                // tiny, so `now - STALE` saturates to 0 and a metric stamped at
+                // 0 would fail a strict `> 0` test and be wrongly evicted.
+                let now = now_millis();

                let Some(counters) = counters_clone.upgrade() else {
                    break;
@@ -624,7 +593,7 @@ impl StatsManager {

                counters.retain(|_, metric_data: &mut Arc<MetricData>| {
                    Arc::strong_count(metric_data) > 1
-                        || unsafe { metric_data.get_last_updated() > cutoff_time }
+                        || now.saturating_sub(metric_data.last_updated_millis()) < 180_000
                });
                counters.shrink_to_fit();
            }
@@ -662,7 +631,7 @@ impl StatsManager {
            let key = entry.key();
            let metric_data = entry.value();

-            let value = unsafe { metric_data.counter.get() };
+            let value = metric_data.counter.get();

            metrics.push(MetricSnapshot {
                name: key.name,
@@ -695,7 +664,7 @@ impl StatsManager {
        let key = MetricKey::new(name, labels.clone());

        if let Some(metric_data) = self.counters.get(&key) {
-            let value = unsafe { metric_data.counter.get() };
+            let value = metric_data.counter.get();
            Some(MetricSnapshot {
                name,
                labels: labels.clone(),
@@ -793,20 +762,18 @@ mod tests {
    }

    #[tokio::test]
-    async fn test_unsafe_counter() {
-        let counter = UnsafeCounter::new();
+    async fn test_counter() {
+        let counter = Counter::new();

-        unsafe {
-            assert_eq!(counter.get(), 0);
-            counter.inc();
-            assert_eq!(counter.get(), 1);
-            counter.add(5);
-            assert_eq!(counter.get(), 6);
-            counter.set(10);
-            assert_eq!(counter.get(), 10);
-            counter.reset();
-            assert_eq!(counter.get(), 0);
-        }
+        assert_eq!(counter.get(), 0);
+        counter.inc();
+        assert_eq!(counter.get(), 1);
+        counter.add(5);
+        assert_eq!(counter.get(), 6);
+        counter.set(10);
+        assert_eq!(counter.get(), 10);
+        counter.reset();
+        assert_eq!(counter.get(), 0);
    }

    #[tokio::test]
@@ -947,12 +914,14 @@ mod tests {
        let counter = stats.get_simple_counter(MetricName::TrafficBytesForwarded);
        counter.set(1);

-        let cutoff_time = Instant::now().checked_add(Duration::from_secs(1)).unwrap();
+        // Cutoff 1s in the future, so every metric is stale by timestamp; only
+        // a live handle keeps a metric.
+        let cutoff_millis = now_millis() + 1_000;
        stats
            .counters
            .retain(|_, metric_data: &mut Arc<MetricData>| {
                Arc::strong_count(metric_data) > 1
-                    || unsafe { metric_data.get_last_updated() > cutoff_time }
+                    || metric_data.last_updated_millis() > cutoff_millis
            });

        assert_eq!(stats.metric_count(), 1);
@@ -963,11 +932,33 @@ mod tests {
            .counters
            .retain(|_, metric_data: &mut Arc<MetricData>| {
                Arc::strong_count(metric_data) > 1
-                    || unsafe { metric_data.get_last_updated() > cutoff_time }
+                    || metric_data.last_updated_millis() > cutoff_millis
            });
        assert_eq!(stats.metric_count(), 0);
    }

+    #[tokio::test]
+    async fn test_counter_handle_concurrent_increment() {
+        const THREADS: usize = 8;
+        const INCREMENTS_PER_THREAD: usize = 10_000;
+
+        let stats = StatsManager::new();
+        let counter = stats.get_simple_counter(MetricName::TrafficPacketsForwarded);
+
+        std::thread::scope(|scope| {
+            for _ in 0..THREADS {
+                let counter = counter.clone();
+                scope.spawn(move || {
+                    for _ in 0..INCREMENTS_PER_THREAD {
+                        counter.inc();
+                    }
+                });
+            }
+        });
+
+        assert_eq!(counter.get(), (THREADS * INCREMENTS_PER_THREAD) as u64);
+    }
+
    #[tokio::test]
    async fn test_stats_rpc_data_structures() {
        // Test GetStatsRequest
@@ -204,7 +204,7 @@ struct NetworkOptions {
        num_args = 0..=1,
        default_missing_value = "true"
    )]
-    dhcp: Option<String>,
+    dhcp: Option<bool>,

    #[arg(
        short,
@@ -907,25 +907,8 @@ impl NetworkOptions {
            cfg.set_network_identity(NetworkIdentity::new_credential(network_name));
        }

-        if let Some(ref dhcp) = self.dhcp {
-            if dhcp == "true" || dhcp == "1" {
-                cfg.set_dhcp(true);
-            } else if dhcp == "false" || dhcp == "0" {
-                cfg.set_dhcp(false);
-            } else {
-                // Treat as CIDR, e.g. "10.0.0.0/24"
-                cfg.set_dhcp(true);
-                let cidr: cidr::Ipv4Cidr = dhcp
-                    .parse()
-                    .with_context(|| format!("failed to parse dhcp cidr: {}", dhcp))?;
-                if cidr.network_length() > 30 {
-                    anyhow::bail!(
-                        "dhcp cidr prefix length must be <= 30, got /{}",
-                        cidr.network_length()
-                    );
-                }
-                cfg.set_dhcp_cidr(Some(cidr));
-            }
+        if let Some(dhcp) = self.dhcp {
+            cfg.set_dhcp(dhcp);
        }

        if let Some(ipv4) = &self.ipv4 {
@@ -1631,7 +1614,7 @@ pub async fn main() -> ExitCode {
    // Verify configurations
    if cli.check_config {
        if let Err(error) = validate_config(&cli).await {
-            log::error!(?error, "Config validation failed");
+            log::error!(%error, "Config validation failed");
            return ExitCode::FAILURE;
        } else {
            return ExitCode::SUCCESS;
@@ -1641,7 +1624,7 @@ pub async fn main() -> ExitCode {
    let mut ret_code = 0;

    if let Err(error) = run_main(cli).await {
-        log::error!(?error);
+        log::error!(%error);
        ret_code = 1;
    }

@@ -1661,12 +1644,13 @@ async fn validate_config(cli: &Cli) -> anyhow::Result<()> {
    for config_file in config_files {
        if config_file == &PathBuf::from("-") {
            let mut stdin = String::new();
-            _ = tokio::io::stdin().read_to_string(&mut stdin).await?;
-            TomlConfigLoader::new_from_str(stdin.as_str())
-                .with_context(|| "config source: stdin")?;
+            _ = tokio::io::stdin()
+                .read_to_string(&mut stdin)
+                .await
+                .context("failed to read config from stdin")?;
+            TomlConfigLoader::new_from_str_with_source("stdin", stdin.as_str())?;
        } else {
-            TomlConfigLoader::new(config_file)
-                .with_context(|| format!("config source: {:?}", config_file))?;
+            TomlConfigLoader::new(config_file)?;
        };
    }

@@ -23,6 +23,9 @@ pub static malloc_conf: &[u8] = b"retain:false\0";

 rust_i18n::i18n!("locales", fallback = "en");

+// The easytier-core CLI intentionally uses a single-thread runtime. The
+// `multi_thread` flag only affects launcher-based deployments (GUI / mobile /
+// web / Windows service); see launcher.rs:223.
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> std::process::ExitCode {
    core::main().await
@@ -829,11 +829,7 @@ impl Instance {
        let nic_ctx = self.nic_ctx.clone();
        let _peer_packet_receiver = self.peer_packet_receiver.clone();
        tokio::spawn(async move {
-            let default_ipv4_addr = if let Some(dhcp_cidr) = global_ctx_c.config.get_dhcp_cidr() {
-                Ipv4Inet::new(dhcp_cidr.first_address(), dhcp_cidr.network_length()).unwrap()
-            } else {
-                Ipv4Inet::new(Ipv4Addr::new(10, 126, 126, 0), 24).unwrap()
-            };
+            let default_ipv4_addr = Ipv4Inet::new(Ipv4Addr::new(10, 126, 126, 0), 24).unwrap();
            let mut current_dhcp_ip: Option<Ipv4Inet> = None;
            let mut next_sleep_time = 0;
            let nic_closed_notifier = Arc::new(Notify::new());
@@ -868,11 +864,7 @@ impl Instance {
                    used_ipv4.insert(peer_ipv4_addr.into());
                }

-                let dhcp_inet = if global_ctx_c.config.get_dhcp_cidr().is_some() {
-                    &default_ipv4_addr
-                } else {
-                    used_ipv4.iter().next().unwrap_or(&default_ipv4_addr)
-                };
+                let dhcp_inet = used_ipv4.iter().next().unwrap_or(&default_ipv4_addr);
                // if old ip is already in this subnet and not conflicted, use it
                if let Some(ip) = current_dhcp_ip
                    && ip.network() == dhcp_inet.network()
@@ -1818,83 +1810,4 @@ mod tests {

        assert!(InstanceConfigPatcher::validate_public_ipv6_patch(&global_ctx, &patch).is_ok());
    }
-
-    #[tokio::test]
-    async fn test_dhcp_cidr_allocates_ip_in_specified_subnet() {
-        use std::time::Duration;
-
-        use crate::common::config::{ConfigLoader, TomlConfigLoader};
-        use crate::instance::instance::Instance;
-        use crate::tunnel::common::tests::wait_for_condition;
-        use crate::tunnel::ring::RingTunnelConnector;
-
-        // inst1: static IP, no DHCP (acts as a peer so DHCP on inst2 can proceed)
-        let config1 = TomlConfigLoader::default();
-        config1.set_inst_name("dhcp_test_inst1".to_owned());
-        config1.set_ipv4(Some("192.168.200.1/24".parse().unwrap()));
-        let mut flags1 = config1.get_flags();
-        flags1.no_tun = true;
-        config1.set_flags(flags1);
-        config1.set_listeners(vec![]);
-
-        // inst2: DHCP enabled with specific CIDR
-        let config2 = TomlConfigLoader::default();
-        config2.set_inst_name("dhcp_test_inst2".to_owned());
-        config2.set_dhcp(true);
-        config2.set_dhcp_cidr(Some("172.20.0.0/24".parse().unwrap()));
-        let mut flags2 = config2.get_flags();
-        flags2.no_tun = true;
-        config2.set_flags(flags2);
-        config2.set_listeners(vec![]);
-
-        let mut inst1 = Instance::new(config1);
-        let mut inst2 = Instance::new(config2);
-
-        inst1.run().await.unwrap();
-        inst2.run().await.unwrap();
-
-        // Connect inst2 to inst1 via ring tunnel
-        inst2
-            .get_conn_manager()
-            .add_connector(RingTunnelConnector::new(
-                format!("ring://{}", inst1.id()).parse().unwrap(),
-            ));
-
-        // Wait for inst2 to see inst1 in routes
-        let pm2 = inst2.get_peer_manager();
-        wait_for_condition(
-            || async {
-                let routes = pm2.list_routes().await;
-                !routes.is_empty()
-            },
-            Duration::from_secs(5),
-        )
-        .await;
-
-        // Wait for DHCP to allocate an IP on inst2
-        let global_ctx2 = inst2.get_global_ctx();
-        wait_for_condition(
-            || async { global_ctx2.get_ipv4().is_some() },
-            Duration::from_secs(15),
-        )
-        .await;
-
-        // Verify allocated IP is within the specified CIDR 172.20.0.0/24
-        let allocated_ip = global_ctx2.get_ipv4().unwrap();
-        let expected_cidr: cidr::Ipv4Cidr = "172.20.0.0/24".parse().unwrap();
-        assert!(
-            expected_cidr.contains(&allocated_ip.address()),
-            "Allocated IP {:?} is not in expected CIDR {:?}",
-            allocated_ip,
-            expected_cidr
-        );
-        // Verify the network prefix length matches
-        assert_eq!(
-            allocated_ip.network_length(),
-            expected_cidr.network_length(),
-            "Allocated IP network length {} does not match expected {}",
-            allocated_ip.network_length(),
-            expected_cidr.network_length()
-        );
-    }
 }
@@ -637,19 +637,6 @@ impl NetworkConfig {
        );
        cfg.set_hostname(self.hostname.clone());
        cfg.set_dhcp(self.dhcp.unwrap_or_default());
-        if let Some(ref dhcp_cidr) = self.dhcp_cidr {
-            let cidr = dhcp_cidr
-                .parse::<cidr::Ipv4Cidr>()
-                .with_context(|| format!("failed to parse dhcp_cidr: {}", dhcp_cidr))?;
-            if cidr.network_length() > 30 {
-                anyhow::bail!(
-                    "dhcp_cidr prefix length must be <= 30, got /{}",
-                    cidr.network_length()
-                );
-            }
-            cfg.set_dhcp(true);
-            cfg.set_dhcp_cidr(Some(cidr));
-        }
        cfg.set_inst_name(self.network_name.clone().unwrap_or_default());

        // The web UI does not expose credential inputs directly, but imported/saved
@@ -1032,7 +1019,6 @@ impl NetworkConfig {
        }

        result.dhcp = Some(config.get_dhcp());
-        result.dhcp_cidr = config.get_dhcp_cidr().map(|c| c.to_string());

        let network_identity = config.get_network_identity();
        result.network_name = Some(network_identity.network_name.clone());
@@ -14,7 +14,7 @@ use std::{
 };

 use dashmap::{DashMap, DashSet};
-use guarden::defer;
+use guarden::{Guard, defer};
 use tokio::{
    sync::{
        Mutex,
@@ -33,7 +33,6 @@ use super::{
    peer_session::{PeerSession, PeerSessionAction},
    traffic_metrics::AggregateTrafficMetrics,
 };
-use crate::utils::BoxExt;
 use crate::{
    common::{
        PeerId,
@@ -380,9 +379,9 @@ impl PeerConn {
            session_filter,
            noise_handshake_result: None,

-            tunnel: Arc::new(Mutex::new(
-                guard!([mut mpsc_tunnel] mpsc_tunnel.close()).boxed(),
-            )),
+            tunnel: Arc::new(Mutex::new(Box::new(
+                guard!([mut mpsc_tunnel] mpsc_tunnel.close()),
+            ))),
            sink,
            recv: Mutex::new(Some(recv)),
            tunnel_info,
@@ -1533,9 +1533,22 @@ impl PeerManager {
    ) -> Result<(), Error> {
        let policy =
            Self::get_next_hop_policy(msg.peer_manager_header().unwrap().is_latency_first());
+        let is_latency_first = msg.peer_manager_header().unwrap().is_latency_first();
        let packet_type = msg.peer_manager_header().unwrap().packet_type;
        let msg_len = msg.buf_len() as u64;
-        let send_result = if peers.has_peer(dst_peer_id) {
+        let latency_first_gateway = if is_latency_first {
+            peers
+                .get_gateway_peer_id(dst_peer_id, policy.clone())
+                .await
+                .filter(|gateway| *gateway != dst_peer_id)
+        } else {
+            None
+        };
+        let send_result = if let Some(gateway) = latency_first_gateway
+            && (peers.has_peer(gateway) || foreign_network_client.has_next_hop(gateway))
+        {
+            relay_peer_map.send_msg(msg, dst_peer_id, policy).await
+        } else if peers.has_peer(dst_peer_id) {
            peers.send_msg_directly(msg, dst_peer_id).await
        } else if foreign_network_client.has_next_hop(dst_peer_id) {
            foreign_network_client.send_msg(msg, dst_peer_id).await
@@ -2185,6 +2198,7 @@ impl PeerManager {
 mod tests {
    use base64::Engine;
    use std::{
+        collections::HashMap,
        fmt::Debug,
        sync::Arc,
        time::{Duration, Instant},
@@ -2192,6 +2206,7 @@ mod tests {

    use crate::{
        common::{
+            PeerId,
            config::Flags,
            global_ctx::{NetworkIdentity, tests::get_mock_global_ctx},
            stats_manager::{LabelSet, LabelType, MetricName},
@@ -2206,7 +2221,7 @@ mod tests {
            peer_conn::tests::set_secure_mode_cfg,
            peer_manager::RouteAlgoType,
            peer_rpc::tests::register_service,
-            route_trait::NextHopPolicy,
+            route_trait::{NextHopPolicy, RouteCostCalculatorInterface},
            tests::{
                connect_peer_manager, create_mock_peer_manager_with_name, wait_route_appear,
                wait_route_appear_with_cost,
@@ -2250,6 +2265,16 @@ mod tests {
        ))
    }

+    struct TestCostCalculator {
+        costs: HashMap<(PeerId, PeerId), i32>,
+    }
+
+    impl RouteCostCalculatorInterface for TestCostCalculator {
+        fn calculate_cost(&self, src: PeerId, dst: PeerId) -> i32 {
+            *self.costs.get(&(src, dst)).unwrap_or(&1)
+        }
+    }
+
    #[test]
    fn recent_traffic_fanout_policy_only_marks_single_peer() {
        assert!(PeerManager::should_mark_recent_traffic_for_fanout(0));
@@ -2657,6 +2682,109 @@ mod tests {
        .await;
    }

+    #[tokio::test]
+    async fn send_msg_internal_uses_latency_first_gateway_for_direct_peer() {
+        let peer_mgr_a = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
+        let peer_mgr_b = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
+        let peer_mgr_c = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
+
+        connect_peer_manager(peer_mgr_a.clone(), peer_mgr_b.clone()).await;
+        connect_peer_manager(peer_mgr_b.clone(), peer_mgr_c.clone()).await;
+        connect_peer_manager(peer_mgr_a.clone(), peer_mgr_c.clone()).await;
+        wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.clone())
+            .await
+            .unwrap();
+        wait_route_appear(peer_mgr_b.clone(), peer_mgr_c.clone())
+            .await
+            .unwrap();
+        wait_route_appear(peer_mgr_a.clone(), peer_mgr_c.clone())
+            .await
+            .unwrap();
+
+        peer_mgr_a
+            .get_route()
+            .set_route_cost_fn(Box::new(TestCostCalculator {
+                costs: HashMap::from([
+                    ((peer_mgr_a.my_peer_id(), peer_mgr_c.my_peer_id()), 100),
+                    ((peer_mgr_a.my_peer_id(), peer_mgr_b.my_peer_id()), 1),
+                    ((peer_mgr_b.my_peer_id(), peer_mgr_c.my_peer_id()), 1),
+                ]),
+            }))
+            .await;
+
+        wait_for_condition(
+            || {
+                let peer_mgr_a = peer_mgr_a.clone();
+                let peer_mgr_b = peer_mgr_b.clone();
+                let peer_mgr_c = peer_mgr_c.clone();
+                async move {
+                    peer_mgr_a
+                        .get_route()
+                        .get_next_hop_with_policy(peer_mgr_c.my_peer_id(), NextHopPolicy::LeastCost)
+                        .await
+                        == Some(peer_mgr_b.my_peer_id())
+                }
+            },
+            Duration::from_secs(5),
+        )
+        .await;
+
+        let b_network_labels = network_labels(&peer_mgr_b);
+        let forwarded_bytes_before = metric_value(
+            &peer_mgr_b,
+            MetricName::TrafficBytesForwarded,
+            &b_network_labels,
+        );
+        let forwarded_packets_before = metric_value(
+            &peer_mgr_b,
+            MetricName::TrafficPacketsForwarded,
+            &b_network_labels,
+        );
+
+        let mut pkt = ZCPacket::new_with_payload(b"latency-first");
+        pkt.fill_peer_manager_hdr(
+            peer_mgr_a.my_peer_id(),
+            peer_mgr_c.my_peer_id(),
+            PacketType::Data as u8,
+        );
+        pkt.mut_peer_manager_header()
+            .unwrap()
+            .set_latency_first(true);
+        let pkt_len = pkt.buf_len() as u64;
+
+        PeerManager::send_msg_internal(
+            &peer_mgr_a.peers,
+            &peer_mgr_a.foreign_network_client,
+            &peer_mgr_a.relay_peer_map,
+            Some(&peer_mgr_a.traffic_metrics),
+            pkt,
+            peer_mgr_c.my_peer_id(),
+        )
+        .await
+        .unwrap();
+
+        wait_for_condition(
+            || {
+                let peer_mgr_b = peer_mgr_b.clone();
+                let b_network_labels = b_network_labels.clone();
+                async move {
+                    metric_value(
+                        &peer_mgr_b,
+                        MetricName::TrafficBytesForwarded,
+                        &b_network_labels,
+                    ) >= forwarded_bytes_before + pkt_len
+                        && metric_value(
+                            &peer_mgr_b,
+                            MetricName::TrafficPacketsForwarded,
+                            &b_network_labels,
+                        ) > forwarded_packets_before
+                }
+            },
+            Duration::from_secs(5),
+        )
+        .await;
+    }
+
    #[tokio::test]
    async fn send_msg_internal_records_control_metrics_for_direct_peer() {
        let peer_mgr_a = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
@@ -102,7 +102,6 @@ message NetworkConfig {
  optional bool disable_relay_data = 65;
  optional bool enable_udp_broadcast_relay = 66;
  optional uint32 socket_mark = 67;
-  optional string dhcp_cidr = 68;
 }

 message PortForwardConfig {
@@ -12,7 +12,7 @@ use std::{
    sync::Arc,
    task::{Context as TaskContext, Poll},
 };
-use tokio::{io::AsyncReadExt, net::TcpStream, sync::Mutex};
+use tokio::{io::AsyncReadExt, net::TcpStream};

 use crate::tunnel::{
    FromUrl, IpVersion, SinkError, SinkItem, StreamItem, Tunnel, TunnelConnector, TunnelError,
@@ -85,7 +85,7 @@ pub struct FakeTcpTunnelListener {
    addr: url::Url,
    os_listener: Option<tokio::net::TcpListener>,
    // interface_name -> fake tcp stack
-    stack_map: DashMap<String, Arc<Mutex<stack::Stack>>>,
+    stack_map: DashMap<String, Arc<stack::Stack>>,
    // a cache from ip addr to interface name
    ip_to_ifname: IpToIfNameCache,
 }
@@ -148,7 +148,7 @@ impl FakeTcpTunnelListener {
    async fn get_stack(
        &self,
        accept_result: &AcceptResult,
-    ) -> Result<Arc<Mutex<stack::Stack>>, TunnelError> {
+    ) -> Result<Arc<stack::Stack>, TunnelError> {
        let local_socket_addr = accept_result.local_addr;

        let interface_name = &accept_result.interface_name;
@@ -158,29 +158,38 @@ impl FakeTcpTunnelListener {
            IpAddr::V6(ip) => (None, Some(ip)),
        };

-        let ret = match self.stack_map.entry(interface_name.to_string()) {
-            dashmap::Entry::Occupied(entry) => entry.get().clone(),
-            dashmap::Entry::Vacant(entry) => {
-                let tun =
-                    create_tun_off_runtime(interface_name.to_string(), None, local_socket_addr)
-                        .await?;
-                tracing::info!(
-                    ?local_socket_addr,
-                    "create new stack with interface_name: {:?}",
-                    interface_name
-                );
-                let stack = Arc::new(Mutex::new(stack::Stack::new(
-                    tun,
-                    local_ip.unwrap_or(Ipv4Addr::UNSPECIFIED),
-                    local_ip6,
-                    accept_result.mac,
-                )));
-                entry.insert(stack.clone());
-                stack
-            }
-        };
+        if let Some(entry) = self.stack_map.get(interface_name) {
+            let stack = entry.clone();
+            drop(entry);

-        Ok(ret)
+            if !stack.is_closed() {
+                return Ok(stack);
+            }
+
+            tracing::warn!(
+                interface_name,
+                "fake_tcp stack reader_task finished, recreating stack"
+            );
+            self.stack_map.remove(interface_name);
+        }
+
+        let tun =
+            create_tun_off_runtime(interface_name.to_string(), None, local_socket_addr).await?;
+        tracing::info!(
+            ?local_socket_addr,
+            "create new stack with interface_name: {:?}",
+            interface_name
+        );
+        let stack = Arc::new(stack::Stack::new(
+            tun,
+            local_ip.unwrap_or(Ipv4Addr::UNSPECIFIED),
+            local_ip6,
+            accept_result.mac,
+        ));
+        self.stack_map
+            .insert(interface_name.to_string(), stack.clone());
+
+        Ok(stack)
    }
 }

@@ -215,19 +224,29 @@ impl TunnelListener for FakeTcpTunnelListener {
        let os_listener = tokio::net::TcpListener::bind(bind_addr).await?;
        tracing::info!(port, "FakeTcpTunnelListener listening");
        self.os_listener = Some(os_listener);
-        // self.stack.lock().await.listen(port);
        Ok(())
    }

    async fn accept(&mut self) -> Result<Box<dyn Tunnel>, TunnelError> {
        tracing::debug!("FakeTcpTunnelListener waiting for accept");
-        let res = self.do_accept().await?;
-        let stack = self.get_stack(&res).await?;
-        let socket = stack
-            .lock()
-            .await
-            .alloc_established_socket(res.local_addr, res.remote_addr, stack::State::Established)
-            .await;
+        let (res, stack, socket) = loop {
+            let res = self.do_accept().await?;
+            let stack = self.get_stack(&res).await?;
+            let socket = stack.try_alloc_established_socket(
+                res.local_addr,
+                res.remote_addr,
+                stack::State::Established,
+            );
+            let Some(socket) = socket else {
+                tracing::warn!(
+                    interface_name = res.interface_name,
+                    "fake_tcp stack closed while accepting connection, dropping accepted socket"
+                );
+                self.stack_map.remove(&res.interface_name);
+                continue;
+            };
+            break (res, stack, socket);
+        };

        tracing::info!(
            ?res,
@@ -236,7 +255,7 @@ impl TunnelListener for FakeTcpTunnelListener {
        );

        let info = TunnelInfo {
-            tunnel_type: get_faketcp_tunnel_type_str(stack.lock().await.driver_type()),
+            tunnel_type: get_faketcp_tunnel_type_str(stack.driver_type()),
            local_addr: Some(self.local_url().into()),
            remote_addr: Some(
                crate::tunnel::build_url_from_socket_addr(
@@ -354,12 +373,14 @@ impl TunnelConnector for FakeTcpTunnelConnector {
        let tun =
            create_tun_off_runtime(interface_name.clone(), Some(remote_addr), local_addr).await?;
        let local_ip = local_ip.unwrap_or("0.0.0.0".parse().unwrap());
-        let mut stack = stack::Stack::new(tun, local_ip, local_ip6, mac);
+        let stack = stack::Stack::new(tun, local_ip, local_ip6, mac);
        let driver_type = stack.driver_type();

        let socket = stack
-            .alloc_established_socket(local_addr, remote_addr, stack::State::SynSent)
-            .await;
+            .try_alloc_established_socket(local_addr, remote_addr, stack::State::SynSent)
+            .ok_or(TunnelError::InternalError(
+                "FakeTCP stack closed while allocating socket".into(),
+            ))?;

        let os_stream = os_socket.connect(remote_addr).await?;

@@ -54,7 +54,7 @@ use std::sync::{
 use tokio::sync::broadcast;
 use tokio::time;
 use tokio_util::task::AbortOnDropHandle;
-use tracing::{info, trace, warn};
+use tracing::{error, info, trace, warn};

 const TIMEOUT: time::Duration = time::Duration::from_secs(1);
 const RETRIES: usize = 6;
@@ -83,13 +83,33 @@ impl AddrTuple {
    }
 }

+#[derive(Default)]
+struct StackState {
+    tuples: HashMap<AddrTuple, flume::Sender<Bytes>>,
+    closed: bool,
+}
+
 struct Shared {
-    tuples: RwLock<HashMap<AddrTuple, flume::Sender<Bytes>>>,
+    state: RwLock<StackState>,
    listening: RwLock<HashSet<u16>>,
    tun: Arc<dyn Tun>,
    tuples_purge: broadcast::Sender<AddrTuple>,
 }

+impl Shared {
+    fn is_closed(&self) -> bool {
+        self.state.read().unwrap().closed
+    }
+
+    fn mark_closed_and_clear_tuples(&self) -> usize {
+        let mut state = self.state.write().unwrap();
+        state.closed = true;
+        let len = state.tuples.len();
+        state.tuples.clear();
+        len
+    }
+}
+
 pub struct Stack {
    shared: Arc<Shared>,
    local_ip: Ipv4Addr,
@@ -353,7 +373,17 @@ impl Drop for Socket {
    fn drop(&mut self) {
        let tuple = AddrTuple::new(self.local_addr, self.remote_addr);
        // dissociates ourself from the dispatch map
-        assert!(self.shared.tuples.write().unwrap().remove(&tuple).is_some());
+        let (removed, closed) = {
+            let mut state = self.shared.state.write().unwrap();
+            (state.tuples.remove(&tuple).is_some(), state.closed)
+        };
+        if !removed {
+            if closed {
+                trace!(?tuple, "Fake TCP tuple already removed after stack closed");
+            } else {
+                warn!(?tuple, "Fake TCP tuple missing while dropping socket");
+            }
+        }
        // purge cache
        let _ = self.shared.tuples_purge.send(tuple);

@@ -400,7 +430,7 @@ impl Stack {
    ) -> Stack {
        let (tuples_purge_tx, _tuples_purge_rx) = broadcast::channel(16);
        let shared = Arc::new(Shared {
-            tuples: RwLock::new(HashMap::new()),
+            state: RwLock::new(StackState::default()),
            tun: tun.clone(),
            listening: RwLock::new(HashSet::new()),
            tuples_purge: tuples_purge_tx.clone(),
@@ -426,19 +456,31 @@ impl Stack {
        self.shared.tun.driver_type()
    }

+    pub fn is_closed(&self) -> bool {
+        self.shared.is_closed() || self.reader_task.is_finished()
+    }
+
    /// Listens for incoming connections on the given `port`.
    pub fn listen(&mut self, port: u16) {
        assert!(self.shared.listening.write().unwrap().insert(port));
    }

-    pub async fn alloc_established_socket(
-        &mut self,
+    pub fn try_alloc_established_socket(
+        &self,
        local_addr: SocketAddr,
        remote_addr: SocketAddr,
        state: State,
-    ) -> Socket {
+    ) -> Option<Socket> {
        let tuple = AddrTuple::new(local_addr, remote_addr);
-        let mut tuples = self.shared.tuples.write().unwrap();
+        let mut stack_state = self.shared.state.write().unwrap();
+        if stack_state.closed || self.reader_task.is_finished() {
+            stack_state.closed = true;
+            warn!(
+                ?tuple,
+                "fake_tcp stack is closed, refusing to allocate socket"
+            );
+            return None;
+        }
        let (sock, incoming) = Socket::new(
            self.shared.clone(),
            // self.shared.tun.choose(&mut rng).unwrap().clone(),
@@ -450,8 +492,8 @@ impl Stack {
            Some(0), // Initial ACK
            state,
        );
-        assert!(tuples.insert(tuple, incoming).is_none());
-        sock
+        assert!(stack_state.tuples.insert(tuple, incoming).is_none());
+        Some(sock)
    }

    async fn reader_task(
@@ -466,7 +508,22 @@ impl Stack {

            tokio::select! {
                size = tun.recv(&mut buf) => {
-                    let size = size.unwrap();
+                    let size = match size {
+                        Ok(size) => size,
+                        Err(e) => {
+                            let shared_tuple_count = shared.mark_closed_and_clear_tuples();
+                            let cached_tuple_count = tuples.len();
+                            tuples.clear();
+                            error!(
+                                ?e,
+                                driver_type = tun.driver_type(),
+                                shared_tuple_count,
+                                cached_tuple_count,
+                                "fake_tcp tun recv failed, reader_task exiting"
+                            );
+                            break;
+                        }
+                    };
                    tracing::trace!(len = size, ?buf, "PnetTun received packet");
                    let buf = buf.split().freeze();

@@ -494,8 +551,8 @@ impl Stack {
                            } else {
                                trace!("Cache miss, checking the shared tuples table for connection");
                                let sender = {
-                                    let tuples = shared.tuples.read().unwrap();
-                                    tuples.get(&tuple).cloned()
+                                    let state = shared.state.read().unwrap();
+                                    state.tuples.get(&tuple).cloned()
                                };

                                if let Some(c) = sender {
@@ -532,11 +589,107 @@ impl Stack {
                    }
                },
                tuple = tuples_purge.recv() => {
-                    let tuple = tuple.unwrap();
-                    tuples.remove(&tuple);
-                    trace!("Removed cached tuple: {:?}", tuple);
+                    match tuple {
+                        Ok(tuple) => {
+                            tuples.remove(&tuple);
+                            trace!("Removed cached tuple: {:?}", tuple);
+                        }
+                        Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                            let cached_tuple_count = tuples.len();
+                            tuples.clear();
+                            warn!(
+                                skipped,
+                                cached_tuple_count,
+                                "fake_tcp tuples purge receiver lagged, cleared local cache"
+                            );
+                        }
+                        Err(broadcast::error::RecvError::Closed) => {
+                            let shared_tuple_count = shared.mark_closed_and_clear_tuples();
+                            let cached_tuple_count = tuples.len();
+                            tuples.clear();
+                            warn!(
+                                shared_tuple_count,
+                                cached_tuple_count,
+                                "fake_tcp tuples purge channel closed, reader_task exiting"
+                            );
+                            break;
+                        }
+                    }
                }
            }
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io;
+    use tokio::{
+        sync::Notify,
+        time::{Duration, timeout},
+    };
+
+    #[derive(Default)]
+    struct FailingTun {
+        fail: Notify,
+    }
+
+    impl FailingTun {
+        fn fail(&self) {
+            self.fail.notify_one();
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl Tun for FailingTun {
+        async fn recv(&self, _packet: &mut BytesMut) -> Result<usize, io::Error> {
+            self.fail.notified().await;
+            Err(io::Error::new(io::ErrorKind::BrokenPipe, "test tun closed"))
+        }
+
+        fn try_send(&self, _packet: &Bytes) -> Result<(), io::Error> {
+            Ok(())
+        }
+
+        fn driver_type(&self) -> &'static str {
+            "test"
+        }
+    }
+
+    #[tokio::test]
+    async fn reader_task_closes_sockets_on_tun_recv_error() {
+        let tun = Arc::new(FailingTun::default());
+        let mut stack = Stack::new(tun.clone(), Ipv4Addr::LOCALHOST, None, None);
+        let socket = stack
+            .try_alloc_established_socket(
+                SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 10_000),
+                SocketAddr::new(Ipv4Addr::new(192, 0, 2, 1).into(), 20_000),
+                State::Established,
+            )
+            .expect("socket allocation should succeed before tun failure");
+
+        tun.fail();
+
+        let join_result = timeout(Duration::from_secs(1), &mut stack.reader_task)
+            .await
+            .expect("reader task should exit after tun recv error");
+        assert!(join_result.is_ok());
+        assert!(stack.is_closed());
+
+        let mut buf = BytesMut::new();
+        let recv_result = timeout(Duration::from_secs(1), socket.recv(&mut buf))
+            .await
+            .expect("socket recv should not hang after reader task exits");
+        assert_eq!(recv_result, None);
+
+        let new_socket = stack.try_alloc_established_socket(
+            SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 10_001),
+            SocketAddr::new(Ipv4Addr::new(192, 0, 2, 1).into(), 20_001),
+            State::Established,
+        );
+        assert!(new_socket.is_none());
+
+        drop(socket);
+    }
+}
@@ -23,6 +23,250 @@ use std::{net::SocketAddr, sync::Arc, time::Duration};
 use tokio::net::UdpSocket;

 // region config
+mod crypto {
+    use crate::utils::BoxExt;
+    use bytes::{Buf, BytesMut};
+    use quinn_proto::crypto::{
+        ClientConfig, ExportKeyingMaterialError, KeyPair, Keys, ServerConfig, Session,
+        UnsupportedVersion,
+    };
+    use quinn_proto::transport_parameters::TransportParameters;
+    use quinn_proto::{
+        ConnectError, ConnectionId, Side, TransportError,
+        crypto::{CryptoError, HeaderKey, PacketKey},
+    };
+    use seahash::SeaHasher;
+    use std::any::Any;
+    use std::{hash::Hasher, sync::Arc};
+    use tracing::{error, instrument, trace};
+
+    #[derive(Debug, Clone, Copy)]
+    struct CryptoKey;
+
+    impl CryptoKey {
+        fn header(self) -> KeyPair<Box<dyn HeaderKey>> {
+            KeyPair {
+                local: Box::new(self),
+                remote: Box::new(self),
+            }
+        }
+
+        fn packet(self) -> KeyPair<Box<dyn PacketKey>> {
+            KeyPair {
+                local: Box::new(self),
+                remote: Box::new(self),
+            }
+        }
+
+        fn keys(self) -> Keys {
+            Keys {
+                header: self.header(),
+                packet: self.packet(),
+            }
+        }
+    }
+
+    impl HeaderKey for CryptoKey {
+        fn decrypt(&self, _: usize, _: &mut [u8]) {}
+        fn encrypt(&self, _: usize, _: &mut [u8]) {}
+        fn sample_size(&self) -> usize {
+            0
+        }
+    }
+
+    impl CryptoKey {
+        fn checksum(slices: &[&[u8]]) -> u64 {
+            let mut hasher = SeaHasher::default();
+            for slice in slices {
+                hasher.write(&(slice.len() as u64).to_le_bytes());
+                hasher.write(slice);
+            }
+            hasher.finish()
+        }
+    }
+
+    impl PacketKey for CryptoKey {
+        #[instrument(level = "trace")]
+        fn encrypt(&self, packet: u64, buf: &mut [u8], header_len: usize) {
+            let (header, rest) = buf.split_at_mut(header_len);
+            let (payload, tag) = rest.split_at_mut(rest.len() - self.tag_len());
+            let checksum = Self::checksum(&[header, payload]);
+            tag.copy_from_slice(&checksum.to_be_bytes());
+            trace!(checksum, ?header, ?payload, ?tag);
+        }
+
+        #[instrument(level = "trace")]
+        fn decrypt(
+            &self,
+            packet: u64,
+            header: &[u8],
+            payload: &mut BytesMut,
+        ) -> Result<(), CryptoError> {
+            let tag = payload.split_off(payload.len() - self.tag_len()).get_u64();
+            trace!(tag, ?payload);
+            let checksum = Self::checksum(&[header, payload]);
+            if checksum != tag {
+                error!(tag, checksum, "checksum mismatch");
+                return Err(CryptoError);
+            }
+            Ok(())
+        }
+
+        fn tag_len(&self) -> usize {
+            8
+        }
+
+        fn confidentiality_limit(&self) -> u64 {
+            u64::MAX
+        }
+
+        fn integrity_limit(&self) -> u64 {
+            1 << 36
+        }
+    }
+
+    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    enum HandshakeState {
+        EmitInitial,
+        EmitHandshake,
+        Done,
+    }
+
+    #[derive(Debug)]
+    struct QuicSession {
+        side: Side,
+        state: HandshakeState,
+        local: TransportParameters,
+        remote: Option<TransportParameters>,
+    }
+
+    impl QuicSession {
+        fn new(side: Side, params: TransportParameters) -> Self {
+            Self {
+                side,
+                state: HandshakeState::EmitInitial,
+                local: params,
+                remote: None,
+            }
+        }
+    }
+
+    impl Session for QuicSession {
+        fn initial_keys(&self, _: &ConnectionId, _: Side) -> Keys {
+            CryptoKey.keys()
+        }
+
+        fn handshake_data(&self) -> Option<Box<dyn Any>> {
+            self.remote.map(|params| params.boxed() as _)
+        }
+
+        fn peer_identity(&self) -> Option<Box<dyn Any>> {
+            None
+        }
+
+        fn early_crypto(&self) -> Option<(Box<dyn HeaderKey>, Box<dyn PacketKey>)> {
+            None
+        }
+
+        fn early_data_accepted(&self) -> Option<bool> {
+            Some(false)
+        }
+
+        #[instrument(level = "trace")]
+        fn is_handshaking(&self) -> bool {
+            self.remote.is_none() || self.state != HandshakeState::Done
+        }
+
+        #[instrument(level = "trace")]
+        fn read_handshake(&mut self, mut buf: &[u8]) -> Result<bool, TransportError> {
+            if self.remote.is_none() {
+                self.remote = Some(
+                    TransportParameters::read(self.side, &mut buf)
+                        .expect("failed to read transport parameters"),
+                );
+            }
+            Ok(true)
+        }
+
+        #[instrument(level = "trace")]
+        fn transport_parameters(&self) -> Result<Option<TransportParameters>, TransportError> {
+            Ok(self.remote)
+        }
+
+        #[instrument(level = "trace")]
+        fn write_handshake(&mut self, buf: &mut Vec<u8>) -> Option<Keys> {
+            match self.state {
+                HandshakeState::EmitInitial => {
+                    if self.side.is_client() {
+                        self.local.write(buf);
+                    }
+                    self.state = HandshakeState::EmitHandshake;
+                    Some(CryptoKey.keys())
+                }
+                HandshakeState::EmitHandshake => {
+                    if self.side.is_server() {
+                        self.local.write(buf);
+                    }
+                    self.state = HandshakeState::Done;
+                    Some(CryptoKey.keys())
+                }
+                HandshakeState::Done => None,
+            }
+        }
+
+        fn next_1rtt_keys(&mut self) -> Option<KeyPair<Box<dyn PacketKey>>> {
+            Some(CryptoKey.packet())
+        }
+
+        fn is_valid_retry(&self, _: &ConnectionId, _: &[u8], _: &[u8]) -> bool {
+            true
+        }
+
+        fn export_keying_material(
+            &self,
+            _: &mut [u8],
+            _: &[u8],
+            _: &[u8],
+        ) -> Result<(), ExportKeyingMaterialError> {
+            Ok(())
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct CryptoConfig;
+
+    impl ClientConfig for CryptoConfig {
+        #[instrument(level = "trace")]
+        fn start_session(
+            self: Arc<Self>,
+            version: u32,
+            server_name: &str,
+            params: &TransportParameters,
+        ) -> Result<Box<dyn Session>, ConnectError> {
+            Ok(Box::new(QuicSession::new(Side::Client, *params)))
+        }
+    }
+
+    impl ServerConfig for CryptoConfig {
+        fn initial_keys(&self, _: u32, _: &ConnectionId) -> Result<Keys, UnsupportedVersion> {
+            Ok(CryptoKey.keys())
+        }
+
+        fn retry_tag(&self, _: u32, _: &ConnectionId, _: &[u8]) -> [u8; 16] {
+            [0u8; 16]
+        }
+
+        #[instrument(level = "trace")]
+        fn start_session(
+            self: Arc<Self>,
+            version: u32,
+            params: &TransportParameters,
+        ) -> Box<dyn Session> {
+            Box::new(QuicSession::new(Side::Server, *params))
+        }
+    }
+}
+
 pub fn transport_config() -> Arc<TransportConfig> {
    let mut config = TransportConfig::default();

@@ -39,13 +283,13 @@ pub fn transport_config() -> Arc<TransportConfig> {
 }

 pub fn server_config() -> ServerConfig {
-    let mut config = quinn_plaintext::server_config();
+    let mut config = ServerConfig::with_crypto(Arc::new(crypto::CryptoConfig));
    config.transport_config(transport_config());
    config
 }

 pub fn client_config() -> ClientConfig {
-    let mut config = quinn_plaintext::client_config();
+    let mut config = ClientConfig::new(Arc::new(crypto::CryptoConfig));
    config.transport_config(transport_config());
    config
 }
@@ -1,7 +1,4 @@
-use std::{
-    cell::UnsafeCell,
-    sync::atomic::{AtomicU32, Ordering::Relaxed},
-};
+use std::sync::atomic::{AtomicU32, AtomicU64, Ordering::Relaxed};

 pub struct WindowLatency {
    latency_us_window: Vec<AtomicU32>,
@@ -63,34 +60,30 @@ impl WindowLatency {

 #[derive(Debug)]
 pub struct Throughput {
-    tx_bytes: UnsafeCell<u64>,
-    rx_bytes: UnsafeCell<u64>,
-    tx_packets: UnsafeCell<u64>,
-    rx_packets: UnsafeCell<u64>,
+    tx_bytes: AtomicU64,
+    rx_bytes: AtomicU64,
+    tx_packets: AtomicU64,
+    rx_packets: AtomicU64,
 }

 impl Clone for Throughput {
    fn clone(&self) -> Self {
        Self {
-            tx_bytes: UnsafeCell::new(unsafe { *self.tx_bytes.get() }),
-            rx_bytes: UnsafeCell::new(unsafe { *self.rx_bytes.get() }),
-            tx_packets: UnsafeCell::new(unsafe { *self.tx_packets.get() }),
-            rx_packets: UnsafeCell::new(unsafe { *self.rx_packets.get() }),
+            tx_bytes: AtomicU64::new(self.tx_bytes()),
+            rx_bytes: AtomicU64::new(self.rx_bytes()),
+            tx_packets: AtomicU64::new(self.tx_packets()),
+            rx_packets: AtomicU64::new(self.rx_packets()),
        }
    }
 }

-// add sync::Send and sync::Sync traits to Throughput
-unsafe impl Send for Throughput {}
-unsafe impl Sync for Throughput {}
-
 impl Default for Throughput {
    fn default() -> Self {
        Self {
-            tx_bytes: UnsafeCell::new(0),
-            rx_bytes: UnsafeCell::new(0),
-            tx_packets: UnsafeCell::new(0),
-            rx_packets: UnsafeCell::new(0),
+            tx_bytes: AtomicU64::new(0),
+            rx_bytes: AtomicU64::new(0),
+            tx_packets: AtomicU64::new(0),
+            rx_packets: AtomicU64::new(0),
        }
    }
 }
@@ -101,32 +94,68 @@ impl Throughput {
    }

    pub fn tx_bytes(&self) -> u64 {
-        unsafe { *self.tx_bytes.get() }
+        self.tx_bytes.load(Relaxed)
    }

    pub fn rx_bytes(&self) -> u64 {
-        unsafe { *self.rx_bytes.get() }
+        self.rx_bytes.load(Relaxed)
    }

    pub fn tx_packets(&self) -> u64 {
-        unsafe { *self.tx_packets.get() }
+        self.tx_packets.load(Relaxed)
    }

    pub fn rx_packets(&self) -> u64 {
-        unsafe { *self.rx_packets.get() }
+        self.rx_packets.load(Relaxed)
    }

    pub fn record_tx_bytes(&self, bytes: u64) {
-        unsafe {
-            *self.tx_bytes.get() += bytes;
-            *self.tx_packets.get() += 1;
-        }
+        self.tx_bytes.fetch_add(bytes, Relaxed);
+        self.tx_packets.fetch_add(1, Relaxed);
    }

    pub fn record_rx_bytes(&self, bytes: u64) {
-        unsafe {
-            *self.rx_bytes.get() += bytes;
-            *self.rx_packets.get() += 1;
-        }
+        self.rx_bytes.fetch_add(bytes, Relaxed);
+        self.rx_packets.fetch_add(1, Relaxed);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Throughput;
+    use std::sync::Arc;
+
+    #[test]
+    fn throughput_records_concurrent_tx_and_rx() {
+        const THREADS: usize = 8;
+        const RECORDS_PER_THREAD: usize = 10_000;
+        const TX_BYTES_PER_RECORD: u64 = 3;
+        const RX_BYTES_PER_RECORD: u64 = 7;
+
+        let throughput = Arc::new(Throughput::new());
+
+        std::thread::scope(|scope| {
+            for _ in 0..THREADS {
+                let throughput = Arc::clone(&throughput);
+                scope.spawn(move || {
+                    for _ in 0..RECORDS_PER_THREAD {
+                        throughput.record_tx_bytes(TX_BYTES_PER_RECORD);
+                        throughput.record_rx_bytes(RX_BYTES_PER_RECORD);
+                    }
+                });
+            }
+        });
+
+        let expected_packets = (THREADS * RECORDS_PER_THREAD) as u64;
+        assert_eq!(throughput.tx_packets(), expected_packets);
+        assert_eq!(throughput.rx_packets(), expected_packets);
+        assert_eq!(
+            throughput.tx_bytes(),
+            expected_packets * TX_BYTES_PER_RECORD
+        );
+        assert_eq!(
+            throughput.rx_bytes(),
+            expected_packets * RX_BYTES_PER_RECORD
+        );
    }
 }
@@ -118,6 +118,7 @@ impl WsTunnelListener {

        let (request, stream) = ServerBuilder::new()
            .limits(Limits::unlimited())
+            .max_headers(128)
            .accept(stream)
            .await?;

@@ -252,7 +253,8 @@ impl WsTunnelConnector {
            ),
        };

-        let c = ClientBuilder::from_uri(http::Uri::try_from(addr.to_string()).unwrap());
+        let c = ClientBuilder::from_uri(http::Uri::try_from(addr.to_string()).unwrap())
+            .max_headers(128);
        let stream: MaybeTlsStream<TcpStream> = if is_wss {
            init_crypto_provider();
            let tls_conn =
Author	SHA1	Message	Date
fanyang89	c12b73e1ae	fix(stats): use age-based GC check to avoid evicting fresh metrics The millis-since-base staleness check (last > now - 180s) saturated the cutoff to 0 early in process life, so metrics stamped at 0ms failed the strict > 0 test and were wrongly evicted by the immediate first GC tick. Switch to age-based (now - last < 180s), equivalent to the original Instant semantics and robust under clock saturation, which fixes the flaky peer_conn_secure_mode_pubkey_and_encryption test.	2026-06-26 23:31:56 +08:00
fanyang	1375cd1832	docs(stats): fix misleading/stale comments and rename test_counter - GC test: cutoff is in the future, so every metric is stale by timestamp (not "nothing is stale"); only live handles retain. - GC loop: drop the inaccurate "no Instant alloc" rationale. - bench: the handle path no longer calls Instant::now() (it uses fastant); reword the HANDLE_TOTAL_WORK rationale. - rename test_unsafe_counter -> test_counter to match the type rename.	2026-06-25 09:16:12 +08:00
fanyang	ebb97fd4f4	refactor(stats): rename UnsafeCounter to Counter The counter is now a plain atomic (no longer UnsafeCell, no longer sharded), so the "Unsafe" prefix is a misleading leftover. Rename to Counter. Also fix a stale bench comment that claimed the ShardedAtomic variant mirrors production.	2026-06-25 00:59:49 +08:00
fanyang	7d249979ea	Update outdated comment	2026-06-25 00:54:26 +08:00
fanyang	8fa161bc1b	Fix clippy	2026-06-25 00:51:55 +08:00
fanyang	2624b2740e	Code format	2026-06-25 00:48:40 +08:00
fanyang	bf8cff60bf	docs: correct misleading multi_thread flag help text The flag defaults to true and only affects launcher-based deployments (GUI/mobile/web/Windows service); the easytier-core CLI is intentionally single-threaded. Fix the help text and document the intent at the CLI entry.	2026-06-25 00:06:38 +08:00
fanyang	a247358ec1	test(bench): add counter contention benchmark Adds benches/counter_contention.rs comparing the production CounterHandle against reconstructed baselines (pre-optimization single-atomic + CAS + Mutex<Instant>) under tokio-task contention, plus counter-only variants. Uses the real stats_manager types so the numbers reflect shipped code. Placed between the sharding and fastant commits so each can be benchmarked independently: at this commit prod = sharded counter + Mutex<Instant> touch.	2026-06-25 00:06:38 +08:00
fanyang	c94d106714	perf(stats): lock-free fastant timestamp and fetch_add counter Switch the counter add() from a fetch_update CAS loop to fetch_add, and replace MetricData's Mutex<Instant> timestamp with a lock-free AtomicU64 storing millis since a lazily-initialized base. Back now_millis() with fastant (TSC on x86_64 Linux, std fallback elsewhere) so touch() is cheap enough to call per packet. GC and its test compare in the millis domain. This is the change that actually delivers the performance: the Mutex timestamp was the serialization bottleneck, and removing it (plus the TSC clock) is what makes the handle path fast. No sharding.	2026-06-25 00:06:38 +08:00
fanyang	721b863547	fix: make stats counters thread safe	2026-06-25 00:06:38 +08:00
fanyangandGitHub	034f5066cd	fix(faketcp): handle closed tun reader without panic (#2308 ) Handle TUN receive errors by marking the fake TCP stack closed and clearing registered sockets instead of panicking. Refuse new sockets on closed stacks and let listeners recreate stacks when the reader task exits.	2026-06-22 10:54:48 +08:00
fanyangandGitHub	9869ddaa4b	fix: clarify config parse errors (#2360 ) * fix: improve config parse diagnostics * fix: polish config error context * test: cover non-ascii config diagnostics	2026-06-21 21:56:49 +08:00
Luna YaoandGitHub	5ea6766238	fix: raise max_headers in ws handshake to 128 (#2366 )	2026-06-21 21:54:07 +08:00
Luna YaoandGitHub	5efbc8587f	upgrade guarden to 0.2.0 (#2365 )	2026-06-18 23:45:28 +08:00
HYecandGitHub	7632cd64da	Fix latency-first routing for direct peers (#2358 )	2026-06-16 20:58:07 +08:00
韩嘉乐andGitHub	16b666ad25	fix: route_update message is not lag (#2355 )	2026-06-16 00:00:48 +08:00
Luna YaoandGitHub	8909e88484	do not panic when fail to parse flags (#2349 )	2026-06-14 13:12:20 +08:00
Luna YaoandGitHub	5edc4cb1cd	fix: remove quinn-plaintext (#2345 ) Remove quinn-plaintext to fix connection errors caused by different hash values across platforms. On x64, maintain compatibility with quinn-plaintext.	2026-06-14 01:21:35 +08:00
韩嘉乐andGitHub	e7709f1cb5	[OHOS] feat: improve status manager (#2343 ) * fix: improve status manager	2026-06-11 17:27:10 +08:00