perf(smoltcp): unify device packet type on BytesMut for zero-copy TX

The smoltcp -> ZCPacket TX path still memcpy'd each outbound packet:
the device emitted Vec<u8>, and `Vec<u8> -> BytesMut` has no zero-copy
conversion in the bytes crate, so socks5/tcp_proxy paid a
`BytesMut::from(Bytes::from(data))` copy per packet.

Switch tokio_smoltcp's packet type to BytesMut end to end so smoltcp
writes into the headroom-reserved buffer that ZCPacket wraps directly.

- device: `Packet = BytesMut`; BufferTxToken allocates via
  BytesMut::with_capacity + resize; BufferRxToken derefs to &[u8].
- channel_device: Stream/Sink/channel carry BytesMut instead of Vec<u8>.
- socks5 / tcp_proxy: TX wraps the BytesMut via new_from_buf directly,
  removing the per-packet memcpy; inbound keeps an equivalent copy via
  BytesMut::from(payload).
- reactor: adapts through the Packet alias, no code change.

Cargo.lock

@@ -241,6 +241,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 +925,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",
@@ -2234,6 +2244,7 @@ dependencies = [
  "aes-gcm",
  "anyhow",
  "arc-swap",
+ "ariadne",
  "async-recursion",
  "async-ringbuf",
  "async-stream",
@@ -2272,7 +2283,7 @@ dependencies = [
  "gethostname 0.5.0",
  "git-version",
  "globwalk",
- "guarden",
+ "guarden 0.2.0",
  "hickory-client",
  "hickory-proto",
  "hickory-resolver",
@@ -2318,7 +2329,7 @@ dependencies = [
  "prost-reflect-build",
  "prost-wkt-types",
  "quinn",
- "quinn-plaintext",
+ "quinn-proto",
  "quote",
  "rand 0.8.5",
  "rcgen",
@@ -2330,6 +2341,7 @@ dependencies = [
  "rstest",
  "rust-i18n",
  "rustls",
+ "seahash",
  "serde",
  "serde_json",
  "serial_test",
@@ -2457,7 +2469,7 @@ dependencies = [
  "dashmap",
  "easytier",
  "futures",
- "guarden",
+ "guarden 0.1.2",
  "jsonwebtoken",
  "mimalloc",
  "mockall",
@@ -3593,7 +3605,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 +3631,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"
@@ -5578,7 +5613,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 2.0.0",
  "proc-macro2",
  "quote",
  "syn 2.0.117",
@@ -6710,11 +6745,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 +7057,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 +7631,7 @@ checksum = "1f168d99749d307be9de54d23fd226628d99768225ef08f6ffb52e0182a27746"
 dependencies = [
  "cfg-if",
  "glob",
- "proc-macro-crate 3.2.0",
+ "proc-macro-crate 3.5.0",
  "proc-macro2",
  "quote",
  "regex",
@@ -9932,8 +9955,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 +10030,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 +10076,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 +11935,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 +12319,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 +12556,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",

easytier-contrib/easytier-ohrs/src/exports/runtime_api.rs

@@ -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) => {

easytier-contrib/easytier-ohrs/src/kernel_bridge.rs

@@ -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};

easytier-contrib/easytier-ohrs/src/kernel_bridge/protocol.rs

@@ -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
 }

easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs

@@ -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;
             }

easytier-contrib/easytier-ohrs/src/lib.rs

@@ -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]

easytier-contrib/easytier-ohrs/src/runtime/state/runtime_state.rs

@@ -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,

easytier/Cargo.toml

@@ -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"
@@ -373,7 +375,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"]

easytier/locales/app.yml

@@ -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: "最初要连接的对等节点"

easytier/src/common/config.rs

@@ -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.

easytier/src/core.rs

@@ -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)?;
         };
     }
 

easytier/src/gateway/socks5.rs

@@ -28,7 +28,7 @@ use crate::{
         ip_reassembler::IpReassembler,
         tokio_smoltcp::{BufferSize, Net, NetConfig, channel_device},
     },
-    tunnel::packet_def::{PacketType, ZCPacket},
+    tunnel::packet_def::{PacketType, ZCPacket, ZCPacketType},
 };
 use anyhow::Context;
 use dashmap::DashMap;
@@ -363,7 +363,10 @@ impl Socks5ServerNet {
             let mut smoltcp_stack_receiver = packet_recv.lock().await;
             while let Some(packet) = smoltcp_stack_receiver.recv().await {
                 tracing::trace!(?packet, "receive from peer send to smoltcp packet");
-                if let Err(e) = stack_sink.send(Ok(packet.payload().to_vec())).await {
+                if let Err(e) = stack_sink
+                    .send(Ok(bytes::BytesMut::from(packet.payload())))
+                    .await
+                {
                     tracing::error!("send to smoltcp stack failed: {:?}", e);
                 }
             }
@@ -377,13 +380,16 @@ impl Socks5ServerNet {
                     "receive from smoltcp stack and send to peer mgr packet, len = {}",
                     data.len()
                 );
-                let Some(ipv4) = Ipv4Packet::new(&data) else {
-                    tracing::error!(?data, "smoltcp stack stream get non ipv4 packet");
+                let packet = ZCPacket::new_from_buf(data, ZCPacketType::NIC);
+                let Some(ipv4) = Ipv4Packet::new(packet.payload()) else {
+                    tracing::error!(
+                        payload_len = packet.payload_len(),
+                        "smoltcp stack stream get non ipv4 packet"
+                    );
                     continue;
                 };
 
                 let dst = ipv4.get_destination();
-                let packet = ZCPacket::new_with_payload(&data);
                 let Some(peer_manager) = peer_manager.upgrade() else {
                     tracing::warn!("peer manager is gone, smoltcp sender exited");
                     return;
@@ -412,7 +418,8 @@ impl Socks5ServerNet {
                     tcp_tx_size: 1024 * 128,
                     ..Default::default()
                 }),
-            ),
+            )
+            .with_packet_tx_headroom(ZCPacketType::NIC.get_packet_offsets().payload_offset),
         );
 
         let forward_tasks = Arc::new(std::sync::Mutex::new(forward_tasks));

easytier/src/gateway/tcp_proxy.rs

@@ -39,6 +39,8 @@ use super::CidrSet;
 
 #[cfg(feature = "smoltcp")]
 use super::tokio_smoltcp::{self, Net, NetConfig, channel_device};
+#[cfg(feature = "smoltcp")]
+use crate::tunnel::packet_def::ZCPacketType;
 
 #[async_trait::async_trait]
 pub(crate) trait NatDstConnector: Send + Sync + Clone + 'static {
@@ -561,7 +563,10 @@ impl<C: NatDstConnector> TcpProxy<C> {
             self.tasks.lock().unwrap().spawn(async move {
                 while let Some(packet) = smoltcp_stack_receiver.recv().await {
                     tracing::trace!(?packet, "receive from peer send to smoltcp packet");
-                    if let Err(e) = stack_sink.send(Ok(packet.payload().to_vec())).await {
+                    if let Err(e) = stack_sink
+                        .send(Ok(bytes::BytesMut::from(packet.payload())))
+                        .await
+                    {
                         tracing::error!("send to smoltcp stack failed: {:?}", e);
                     }
                 }
@@ -575,13 +580,16 @@ impl<C: NatDstConnector> TcpProxy<C> {
                         ?data,
                         "receive from smoltcp stack and send to peer mgr packet"
                     );
-                    let Some(ipv4) = Ipv4Packet::new(&data) else {
-                        tracing::error!(?data, "smoltcp stack stream get non ipv4 packet");
+                    let packet = ZCPacket::new_from_buf(data, ZCPacketType::NIC);
+                    let Some(ipv4) = Ipv4Packet::new(packet.payload()) else {
+                        tracing::error!(
+                            payload_len = packet.payload_len(),
+                            "smoltcp stack stream get non ipv4 packet"
+                        );
                         continue;
                     };
 
                     let dst = ipv4.get_destination();
-                    let packet = ZCPacket::new_with_payload(&data);
                     let Some(peer_mgr) = peer_mgr.upgrade() else {
                         tracing::warn!("peer manager is gone, smoltcp sender exited");
                         return;
@@ -610,7 +618,8 @@ impl<C: NatDstConnector> TcpProxy<C> {
                         tcp_tx_size: 1024 * 16,
                         ..Default::default()
                     }),
-                ),
+                )
+                .with_packet_tx_headroom(ZCPacketType::NIC.get_packet_offsets().payload_offset),
             );
             net.set_any_ip(true);
             self.smoltcp_net.lock().await.replace(net);

easytier/src/gateway/tokio_smoltcp/channel_device.rs

@@ -1,3 +1,4 @@
+use bytes::BytesMut;
 use futures::{Sink, Stream};
 use smoltcp::phy::DeviceCapabilities;
 use std::{
@@ -12,15 +13,15 @@ use super::device::AsyncDevice;
 
 /// A device that send and receive packets using a channel.
 pub struct ChannelDevice {
-    recv: Receiver<io::Result<Vec<u8>>>,
-    send: PollSender<Vec<u8>>,
+    recv: Receiver<io::Result<BytesMut>>,
+    send: PollSender<BytesMut>,
     caps: DeviceCapabilities,
 }
 
 pub type ChannelDeviceNewRet = (
     ChannelDevice,
-    Sender<io::Result<Vec<u8>>>,
-    Receiver<Vec<u8>>,
+    Sender<io::Result<BytesMut>>,
+    Receiver<BytesMut>,
 );
 
 impl ChannelDevice {
@@ -43,25 +44,25 @@ impl ChannelDevice {
 }
 
 impl Stream for ChannelDevice {
-    type Item = io::Result<Vec<u8>>;
+    type Item = io::Result<BytesMut>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.recv.poll_recv(cx)
     }
 }
 
-fn map_err(e: PollSendError<Vec<u8>>) -> io::Error {
+fn map_err(e: PollSendError<BytesMut>) -> io::Error {
     io::Error::other(e)
 }
 
-impl Sink<Vec<u8>> for ChannelDevice {
+impl Sink<BytesMut> for ChannelDevice {
     type Error = io::Error;
 
     fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.send.poll_reserve(cx).map_err(map_err)
     }
 
-    fn start_send(mut self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
+    fn start_send(mut self: Pin<&mut Self>, item: BytesMut) -> Result<(), Self::Error> {
         self.send.send_item(item).map_err(map_err)
     }
 

easytier/src/gateway/tokio_smoltcp/device.rs

@@ -1,3 +1,4 @@
+use bytes::BytesMut;
 use futures::{Sink, Stream};
 pub use smoltcp::phy::DeviceCapabilities;
 use smoltcp::{
@@ -10,7 +11,7 @@ use std::{collections::VecDeque, io};
 pub const DEFAULT_MAX_BURST_SIZE: usize = 100;
 
 /// A packet used in `AsyncDevice`.
-pub type Packet = Vec<u8>;
+pub type Packet = BytesMut;
 
 /// A device that send and receive packets asynchronously.
 pub trait AsyncDevice:
@@ -33,6 +34,7 @@ where
 pub struct BufferDevice {
     caps: DeviceCapabilities,
     max_burst_size: usize,
+    tx_headroom: usize,
     recv_queue: VecDeque<Packet>,
     send_queue: VecDeque<Packet>,
 }
@@ -41,13 +43,11 @@ pub struct BufferDevice {
 pub struct BufferRxToken(Packet);
 
 impl RxToken for BufferRxToken {
-    fn consume<R, F>(mut self, f: F) -> R
+    fn consume<R, F>(self, f: F) -> R
     where
         F: FnOnce(&[u8]) -> R,
     {
-        let p = &mut self.0;
-
-        f(p)
+        f(&self.0[..])
     }
 }
 
@@ -59,8 +59,10 @@ impl<'d> TxToken for BufferTxToken<'d> {
     where
         F: FnOnce(&mut [u8]) -> R,
     {
-        let mut buffer = vec![0u8; len];
-        let result = f(&mut buffer);
+        let tx_headroom = self.0.tx_headroom;
+        let mut buffer = BytesMut::with_capacity(tx_headroom + len);
+        buffer.resize(tx_headroom + len, 0);
+        let result = f(&mut buffer[tx_headroom..]);
 
         self.0.send_queue.push_back(buffer);
 
@@ -98,11 +100,12 @@ impl Device for BufferDevice {
 }
 
 impl BufferDevice {
-    pub(crate) fn new(caps: DeviceCapabilities) -> BufferDevice {
+    pub(crate) fn new(caps: DeviceCapabilities, tx_headroom: usize) -> BufferDevice {
         let max_burst_size = caps.max_burst_size.unwrap_or(DEFAULT_MAX_BURST_SIZE);
         BufferDevice {
             caps,
             max_burst_size,
+            tx_headroom,
             recv_queue: VecDeque::with_capacity(max_burst_size),
             send_queue: VecDeque::with_capacity(max_burst_size),
         }
@@ -123,3 +126,27 @@ impl BufferDevice {
         self.recv_queue.is_empty()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn buffer_device_reserves_tx_headroom() {
+        let mut caps = DeviceCapabilities::default();
+        caps.max_burst_size = Some(1);
+        let mut device = BufferDevice::new(caps, 16);
+
+        let token = device.transmit(Instant::now()).unwrap();
+        token.consume(4, |buf| {
+            assert_eq!(buf.len(), 4);
+            buf.copy_from_slice(&[1, 2, 3, 4]);
+        });
+
+        let mut queue = device.take_send_queue();
+        let packet = queue.pop_front().unwrap();
+        assert_eq!(packet.len(), 20);
+        assert_eq!(&packet[..16], &[0; 16]);
+        assert_eq!(&packet[16..], &[1, 2, 3, 4]);
+    }
+}

easytier/src/gateway/tokio_smoltcp/mod.rs

@@ -51,6 +51,7 @@ pub struct NetConfig {
     pub ip_addr: IpCidr,
     pub gateway: Vec<IpAddress>,
     pub buffer_size: BufferSize,
+    pub(crate) packet_tx_headroom: usize,
 }
 
 impl NetConfig {
@@ -65,8 +66,14 @@ impl NetConfig {
             ip_addr,
             gateway,
             buffer_size: buffer_size.unwrap_or_default(),
+            packet_tx_headroom: 0,
         }
     }
+
+    pub fn with_packet_tx_headroom(mut self, packet_tx_headroom: usize) -> Self {
+        self.packet_tx_headroom = packet_tx_headroom;
+        self
+    }
 }
 
 /// `Net` is the main interface to the network stack.
@@ -97,7 +104,8 @@ impl Net {
     }
 
     fn new2<D: device::AsyncDevice + 'static>(device: D, config: NetConfig) -> Net {
-        let mut buffer_device = BufferDevice::new(device.capabilities().clone());
+        let mut buffer_device =
+            BufferDevice::new(device.capabilities().clone(), config.packet_tx_headroom);
         let mut iface = Interface::new(config.interface_config, &mut buffer_device, Instant::now());
         let ip_addr = config.ip_addr;
         iface.update_ip_addrs(|ip_addrs| {

easytier/src/instance/instance.rs

@@ -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()
-        );
-    }
 }

easytier/src/launcher.rs

@@ -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());

easytier/src/peers/foreign_network_manager.rs

@@ -14,7 +14,7 @@ use std::{
 };
 
 use dashmap::{DashMap, DashSet};
-use guarden::defer;
+use guarden::{Guard, defer};
 use tokio::{
     sync::{
         Mutex,

easytier/src/peers/peer_conn.rs

@@ -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,

easytier/src/peers/peer_manager.rs

@@ -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;

easytier/src/proto/api_manage.proto

@@ -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 {

easytier/src/tunnel/fake_tcp/mod.rs

@@ -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?;
 

easytier/src/tunnel/fake_tcp/stack.rs

@@ -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);
+    }
+}

easytier/src/tunnel/packet_def.rs

@@ -764,6 +764,7 @@ impl ZCPacket {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::{hint::black_box, time::Instant};
 
     #[test]
     fn test_zc_packet() {
@@ -809,4 +810,68 @@ mod tests {
 
         assert!(packet.mut_wg_tunnel_header().is_none());
     }
+
+    fn bench_smoltcp_zcpacket_construct(payload_len: usize, iterations: usize) {
+        let nic_offset = ZCPacketType::NIC.get_packet_offsets().payload_offset;
+
+        // Correctness check (outside the timed section): both construction paths
+        // must yield equivalent payloads for the perf comparison to be meaningful.
+        {
+            let data = vec![7u8; payload_len];
+            let p_copy = ZCPacket::new_with_payload(&data);
+
+            let mut buf = BytesMut::with_capacity(nic_offset + payload_len);
+            buf.resize(nic_offset + payload_len, 0);
+            buf[nic_offset..].fill(7);
+            let p_zero = ZCPacket::new_from_buf(buf, ZCPacketType::NIC);
+
+            assert_eq!(p_copy.payload(), p_zero.payload());
+        }
+
+        // copy path: smoltcp emits a bare payload buf; socks5/tcp_proxy copy it
+        // via ZCPacket::new_with_payload (pre-f5ce0848 behavior).
+        let now = Instant::now();
+        let mut checksum = 0usize;
+        for _ in 0..iterations {
+            let data = vec![7u8; payload_len];
+            let p = ZCPacket::new_with_payload(black_box(&data));
+            // black_box forces the side-effect-free construction to be emitted;
+            // payload_len is stable per run so it cannot skew the numbers.
+            checksum = checksum.wrapping_add(black_box(&p).payload_len());
+        }
+        let copy_elapsed = now.elapsed().as_secs_f64();
+
+        // zerocopy path: device reserves NIC headroom in the buf; socks5/tcp_proxy
+        // wrap it zero-copy via ZCPacket::new_from_buf (f5ce0848 behavior).
+        let now = Instant::now();
+        let mut checksum2 = 0usize;
+        for _ in 0..iterations {
+            let mut buf = BytesMut::with_capacity(nic_offset + payload_len);
+            buf.resize(nic_offset + payload_len, 0);
+            buf[nic_offset..].fill(7);
+            let p = ZCPacket::new_from_buf(black_box(buf), ZCPacketType::NIC);
+            checksum2 = checksum2.wrapping_add(black_box(&p).payload_len());
+        }
+        let zerocopy_elapsed = now.elapsed().as_secs_f64();
+
+        println!(
+            "smoltcp_zcpacket payload_len={} iterations={} copy_pps={:.0} copy_bytes_per_sec={:.0} zerocopy_pps={:.0} zerocopy_bytes_per_sec={:.0} speedup={:.2}x checksums={}/{}",
+            payload_len,
+            iterations,
+            iterations as f64 / copy_elapsed,
+            (payload_len * iterations) as f64 / copy_elapsed,
+            iterations as f64 / zerocopy_elapsed,
+            (payload_len * iterations) as f64 / zerocopy_elapsed,
+            copy_elapsed / zerocopy_elapsed,
+            checksum,
+            checksum2
+        );
+    }
+
+    #[test]
+    #[ignore = "benchmark helper; run with --ignored --nocapture"]
+    fn smoltcp_zcpacket_construct_bench() {
+        bench_smoltcp_zcpacket_construct(1280, 1_000_000);
+        bench_smoltcp_zcpacket_construct(4096, 500_000);
+    }
 }

easytier/src/tunnel/quic.rs

@@ -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
 }

easytier/src/tunnel/websocket.rs

@@ -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 =