fix: address dns review issues

Handle TUN receive errors by marking the fake TCP stack closed and
clearing registered sockets instead of panicking.

Refuse new sockets on closed stacks and let listeners recreate stacks
when the reader task exits.

.cargo/audit.toml

@@ -0,0 +1,3 @@
+[advisories]
+# openidconnect 4.0.1 depends on rsa 0.9.10, and RUSTSEC-2023-0071 has no fixed upgrade.
+ignore = ["RUSTSEC-2023-0071"]

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-contrib/easytier-uptime/Cargo.toml

@@ -33,7 +33,7 @@ sea-orm-migration = { version = "1.1" }
 sqlx = { version = "0.8", features = ["sqlite", "runtime-tokio-rustls", "chrono", "uuid"] }
 
 # Validation
-validator = { version = "0.18", features = ["derive"] }
+validator = { version = "0.20", features = ["derive"] }
 thiserror = "1.0"
 jsonwebtoken = "9.0"
 

easytier-web/Cargo.toml

@@ -15,7 +15,7 @@ dashmap = "6.1"
 url = "2.2"
 async-trait = "0.1"
 
-maxminddb = "0.24"
+maxminddb = "0.27"
 once_cell = "1.18"
 
 axum = { version = "0.7", features = ["macros"] }

easytier-web/src/client_manager/mod.rs

@@ -245,32 +245,40 @@ impl ClientManager {
         }
 
         let location = if let Some(db) = &*geoip_db {
-            match db.lookup::<geoip2::City>(ip) {
-                Ok(city) => {
+            match db.lookup(ip).and_then(|result| result.decode::<geoip2::City>()) {
+                Ok(Some(city)) => {
                     let country = city
                         .country
-                        .and_then(|c| c.names)
-                        .and_then(|n| {
-                            n.get("zh-CN")
-                                .or_else(|| n.get("en"))
-                                .map(|s| s.to_string())
-                        })
+                        .names
+                        .simplified_chinese
+                        .or(city.country.names.english)
+                        .map(|s| s.to_string())
                         .unwrap_or_else(|| "海外".to_string());
 
-                    let city_name = city.city.and_then(|c| c.names).and_then(|n| {
-                        n.get("zh-CN")
-                            .or_else(|| n.get("en"))
-                            .map(|s| s.to_string())
-                    });
+                    let city_name = city
+                        .city
+                        .names
+                        .simplified_chinese
+                        .or(city.city.names.english)
+                        .map(|s| s.to_string());
 
-                    let region = city.subdivisions.map(|r| {
-                        r.iter()
-                            .filter_map(|x| x.names.as_ref())
-                            .filter_map(|x| x.get("zh-CN").or_else(|| x.get("en")))
+                    let region = if city.subdivisions.is_empty() {
+                        None
+                    } else {
+                        let region = city
+                            .subdivisions
+                            .iter()
+                            .filter_map(|x| x.names.simplified_chinese.or(x.names.english))
                             .map(|x| x.to_string())
                             .collect::<Vec<_>>()
-                            .join(",")
-                    });
+                            .join(",");
+
+                        if region.is_empty() {
+                            None
+                        } else {
+                            Some(region)
+                        }
+                    };
 
                     Location {
                         country,
@@ -278,6 +286,14 @@ impl ClientManager {
                         region,
                     }
                 }
+                Ok(None) => {
+                    tracing::debug!("GeoIP data not found for {}", ip);
+                    Location {
+                        country: "海外".to_string(),
+                        city: None,
+                        region: None,
+                    }
+                }
                 Err(err) => {
                     tracing::debug!("GeoIP lookup failed for {}: {}", ip, err);
                     Location {

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"
@@ -234,12 +236,11 @@ http_req = { git = "https://github.com/EasyTier/http_req.git", default-features
 ] }
 
 # for dns connector
-hickory-resolver = "0.25.2"
-hickory-proto = "0.25.2"
+hickory-resolver = "0.26.1"
+hickory-proto = "0.26.1"
 
 # for magic dns
-hickory-client = { version = "0.25.2", optional = true }
-hickory-server = { version = "0.25.2", features = [
+hickory-server = { version = "0.26.1", features = [
     "resolver",
 ], optional = true }
 
@@ -373,7 +374,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"]
@@ -399,7 +400,7 @@ jemalloc-prof = [
     "jemalloc-sys/stats",
 ]
 tracing = ["tokio/tracing", "dep:console-subscriber"]
-magic-dns = ["dep:hickory-client", "dep:hickory-server"]
+magic-dns = ["dep:hickory-server"]
 faketcp = ["dep:flume"]
 zstd = ["dep:zstd"]
 # For Network Extension on macOS

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;
@@ -569,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>>,
@@ -591,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 {
@@ -628,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))
     }
 }
 
@@ -1199,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)
@@ -1227,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;
 
@@ -1268,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/common/dns.rs

@@ -3,33 +3,40 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 
 use anyhow::Context;
-use hickory_proto::runtime::TokioRuntimeProvider;
-use hickory_proto::xfer::Protocol;
-use hickory_resolver::config::{LookupIpStrategy, NameServerConfig, ResolverConfig, ResolverOpts};
-use hickory_resolver::name_server::{GenericConnector, TokioConnectionProvider};
+use hickory_proto::rr::RData;
+use hickory_resolver::config::{
+    ConnectionConfig, LookupIpStrategy, NameServerConfig, ResolverConfig, ResolverOpts,
+};
+use hickory_resolver::net::runtime::TokioRuntimeProvider;
 use hickory_resolver::system_conf::read_system_conf;
-use hickory_resolver::{Resolver, TokioResolver};
+use hickory_resolver::TokioResolver;
 use once_cell::sync::Lazy;
 use tokio::net::lookup_host;
 
 use super::error::Error;
 
 pub fn get_default_resolver_config() -> ResolverConfig {
-    let mut default_resolve_config = ResolverConfig::new();
-    default_resolve_config.add_name_server(NameServerConfig::new(
-        "223.5.5.5:53".parse().unwrap(),
-        Protocol::Udp,
-    ));
-    default_resolve_config.add_name_server(NameServerConfig::new(
-        "180.184.1.1:53".parse().unwrap(),
-        Protocol::Udp,
-    ));
-    default_resolve_config
+    ResolverConfig::from_parts(
+        None,
+        vec![],
+        vec![
+            NameServerConfig::new(
+                "223.5.5.5".parse().unwrap(),
+                true,
+                vec![ConnectionConfig::udp()],
+            ),
+            NameServerConfig::new(
+                "180.184.1.1".parse().unwrap(),
+                true,
+                vec![ConnectionConfig::udp()],
+            ),
+        ],
+    )
 }
 
 pub static ALLOW_USE_SYSTEM_DNS_RESOLVER: Lazy<AtomicBool> = Lazy::new(|| AtomicBool::new(true));
 
-pub static RESOLVER: Lazy<Arc<Resolver<GenericConnector<TokioRuntimeProvider>>>> =
+pub static RESOLVER: Lazy<Arc<TokioResolver>> =
     Lazy::new(|| {
         let system_cfg = read_system_conf();
         let mut cfg = get_default_resolver_config();
@@ -41,9 +48,11 @@ pub static RESOLVER: Lazy<Arc<Resolver<GenericConnector<TokioRuntimeProvider>>>>
             opt = s.1;
         }
         opt.ip_strategy = LookupIpStrategy::Ipv4AndIpv6;
-        let builder = TokioResolver::builder_with_config(cfg, TokioConnectionProvider::default())
-            .with_options(opt);
-        Arc::new(builder.build())
+        let resolver = TokioResolver::builder_with_config(cfg, TokioRuntimeProvider::default())
+            .with_options(opt)
+            .build()
+            .expect("failed to build DNS resolver");
+        Arc::new(resolver)
     });
 
 pub async fn resolve_txt_record(domain_name: &str) -> Result<String, Error> {
@@ -53,12 +62,16 @@ pub async fn resolve_txt_record(domain_name: &str) -> Result<String, Error> {
         .await
         .with_context(|| format!("txt_lookup failed, domain_name: {}", domain_name))?;
 
-    let txt_record = response
+    let Some(RData::TXT(txt_record)) = response
+        .answers()
         .iter()
         .next()
-        .with_context(|| format!("no txt record found, domain_name: {}", domain_name))?;
+        .map(|record| &record.data)
+    else {
+        return Err(anyhow::anyhow!("no txt record found, domain_name: {}", domain_name).into());
+    };
 
-    let txt_data = String::from_utf8_lossy(&txt_record.txt_data()[0]);
+    let txt_data = String::from_utf8_lossy(&txt_record.txt_data[0]);
     tracing::info!(?txt_data, ?domain_name, "get txt record");
 
     Ok(txt_data.to_string())

easytier/src/connector/dns_connector.rs

@@ -13,7 +13,7 @@ use crate::{
 };
 use anyhow::Context;
 use dashmap::DashSet;
-use hickory_resolver::proto::rr::rdata::SRV;
+use hickory_resolver::proto::rr::{RData, rdata::SRV};
 use rand::{Rng as _, seq::SliceRandom};
 use strum::VariantArray;
 
@@ -85,12 +85,12 @@ impl DnsTunnelConnector {
 
     fn handle_one_srv_record(record: &SRV, protocol: IpScheme) -> Result<(url::Url, u64), Error> {
         // port must be non-zero
-        if record.port() == 0 {
+        if record.port == 0 {
             return Err(anyhow::anyhow!("port must be non-zero").into());
         }
 
-        let connector_dst = record.target().to_utf8();
-        let dst_url = format!("{}://{}:{}", protocol, connector_dst, record.port());
+        let connector_dst = record.target.to_utf8();
+        let dst_url = format!("{}://{}:{}", protocol, connector_dst, record.port);
 
         Ok((
             dst_url.parse().with_context(|| {
@@ -98,11 +98,11 @@ impl DnsTunnelConnector {
                     "parse dst_url failed, protocol: {}, connector_dst: {}, port: {}, dst_url: {}",
                     protocol,
                     connector_dst,
-                    record.port(),
+                    record.port,
                     dst_url
                 )
             })?,
-            record.priority() as _,
+            record.priority as _,
         ))
     }
 
@@ -129,7 +129,10 @@ impl DnsTunnelConnector {
                         format!("srv_lookup failed, srv_domain: {}", srv_domain)
                     })?;
                     tracing::info!(?response, ?srv_domain, "srv_lookup response");
-                    for record in response.iter() {
+                    for record in response.answers() {
+                        let RData::SRV(record) = &record.data else {
+                            continue;
+                        };
                         let parsed_record = Self::handle_one_srv_record(record, **protocol);
                         tracing::info!(?parsed_record, ?srv_domain, "parsed_record");
                         if let Err(e) = &parsed_record {

easytier/src/core.rs

@@ -1614,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;
@@ -1624,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;
     }
 
@@ -1644,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/instance/dns_server/config.rs

@@ -92,12 +92,11 @@ impl TryFrom<&Record> for rr::Record {
 
     fn try_from(value: &Record) -> Result<Self, Self::Error> {
         let name = value.name()?;
-        let mut record = Self::update0(name, value.ttl.as_secs() as u32, value.rr_type());
-        record.set_dns_class(rr::DNSClass::IN);
+        let ttl = value.ttl.as_secs() as u32;
         match value.rr_type {
             RecordType::A => {
                 let addr: Ipv4Addr = value.value.parse()?;
-                record.set_data(RData::A(rr::rdata::a::A(addr)));
+                Ok(Self::from_rdata(name, ttl, RData::A(rr::rdata::a::A(addr))))
             }
             RecordType::SOA => {
                 let soa = value.value.split_whitespace().collect::<Vec<_>>();
@@ -111,7 +110,7 @@ impl TryFrom<&Record> for rr::Record {
                 let retry: u32 = soa[4].parse()?;
                 let expire: u32 = soa[5].parse()?;
                 let minimum: u32 = soa[6].parse()?;
-                record.set_data(RData::SOA(rr::rdata::soa::SOA::new(
+                Ok(Self::from_rdata(name, ttl, RData::SOA(rr::rdata::soa::SOA::new(
                     mname,
                     rname,
                     serial,
@@ -119,11 +118,10 @@ impl TryFrom<&Record> for rr::Record {
                     retry.try_into().unwrap(),
                     expire.try_into().unwrap(),
                     minimum,
-                )));
+                ))))
             }
             _ => todo!(),
         }
-        Ok(record)
     }
 }
 

easytier/src/instance/dns_server/server.rs

@@ -3,14 +3,14 @@ use hickory_proto::op::Edns;
 use hickory_proto::rr;
 use hickory_proto::rr::LowerName;
 use hickory_resolver::config::ResolverOpts;
-use hickory_resolver::name_server::TokioConnectionProvider;
+use hickory_resolver::net::runtime::TokioRuntimeProvider;
 use hickory_resolver::system_conf::read_system_conf;
-use hickory_server::ServerFuture;
-use hickory_server::authority::{AuthorityObject, Catalog, ZoneType};
+use hickory_server::net::runtime::TokioTime;
+use hickory_server::server::Server as HickoryServer;
 use hickory_server::server::{Request, RequestHandler, ResponseHandler, ResponseInfo};
 use hickory_server::store::forwarder::ForwardConfig;
-use hickory_server::store::{forwarder::ForwardAuthority, in_memory::InMemoryAuthority};
-use std::io;
+use hickory_server::store::{forwarder::ForwardZoneHandler, in_memory::InMemoryZoneHandler};
+use hickory_server::zone_handler::{AxfrPolicy, Catalog, ZoneHandler, ZoneType};
 use std::net::SocketAddr;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -24,7 +24,7 @@ use crate::common::dns::get_default_resolver_config;
 use super::config::{GeneralConfig, Record, RunConfig};
 
 pub struct Server {
-    server: ServerFuture<CatalogRequestHandler>,
+    server: HickoryServer<CatalogRequestHandler>,
     catalog: Arc<RwLock<Catalog>>,
     general_config: GeneralConfig,
     udp_local_addr: Option<SocketAddr>,
@@ -52,7 +52,7 @@ impl CatalogRequestHandler {
 
 #[async_trait::async_trait]
 impl RequestHandler for CatalogRequestHandler {
-    async fn handle_request<R: ResponseHandler>(
+    async fn handle_request<R: ResponseHandler, T: hickory_server::net::runtime::Time>(
         &self,
         request: &Request,
         response_handle: R,
@@ -60,14 +60,14 @@ impl RequestHandler for CatalogRequestHandler {
         self.catalog
             .read()
             .await
-            .handle_request(request, response_handle)
+            .handle_request::<R, T>(request, response_handle)
             .await
     }
 }
 
-pub fn build_authority(domain: &str, records: &[Record]) -> Result<InMemoryAuthority> {
+pub fn build_authority(domain: &str, records: &[Record]) -> Result<InMemoryZoneHandler> {
     let zone = rr::Name::from_str(domain)?;
-    let mut authority = InMemoryAuthority::empty(zone, ZoneType::Primary, false);
+    let mut authority = InMemoryZoneHandler::empty(zone, ZoneType::Primary, AxfrPolicy::Deny);
     for record in records.iter() {
         let r = record.try_into()?;
         authority.upsert_mut(r, 0);
@@ -97,18 +97,16 @@ impl Server {
                 .name_servers()
                 .iter()
                 .filter(|&x| {
-                    !config
-                        .excluded_forward_nameservers()
-                        .contains(&x.socket_addr.ip())
+                    !config.excluded_forward_nameservers().contains(&x.ip)
                 })
                 .cloned()
                 .collect::<Vec<_>>()
                 .into(),
             options: Some(system_conf.1),
         };
-        let auth = ForwardAuthority::builder_with_config(
+        let auth = ForwardZoneHandler::builder_with_config(
             forward_config,
-            TokioConnectionProvider::default(),
+            TokioRuntimeProvider::default(),
         )
         .build()
         .unwrap();
@@ -117,7 +115,7 @@ impl Server {
 
         let catalog = Arc::new(RwLock::new(catalog));
         let handler = CatalogRequestHandler::new(catalog.clone());
-        let server = ServerFuture::new(handler);
+        let server = HickoryServer::new(handler);
 
         Ok(Self {
             server,
@@ -187,7 +185,7 @@ impl Server {
                 .with_context(|| format!("DNS Server failed to bind TCP address {}", address))?;
             self.tcp_local_addr = Some(tcp_listener.local_addr()?);
             self.server
-                .register_listener(tcp_listener, Duration::from_secs(5));
+                .register_listener(tcp_listener, Duration::from_secs(5), 1024);
         }
 
         if let Some(address) = self.general_config.listen_udp() {
@@ -203,11 +201,11 @@ impl Server {
         Ok(())
     }
 
-    pub async fn upsert(&self, name: LowerName, authority: Arc<dyn AuthorityObject>) {
+    pub async fn upsert(&self, name: LowerName, authority: Arc<dyn ZoneHandler>) {
         self.catalog.write().await.upsert(name, vec![authority]);
     }
 
-    pub async fn remove(&self, name: &LowerName) -> Option<Vec<Arc<dyn AuthorityObject>>> {
+    pub async fn remove(&self, name: &LowerName) -> Option<Vec<Arc<dyn ZoneHandler>>> {
         self.catalog.write().await.remove(name)
     }
 
@@ -216,11 +214,16 @@ impl Server {
         update: &Request,
         response_edns: Option<Edns>,
         response_handle: R,
-    ) -> io::Result<ResponseInfo> {
+    ) -> ResponseInfo {
         self.catalog
             .write()
             .await
-            .update(update, response_edns, response_handle)
+            .update(
+                update,
+                response_edns.as_ref(),
+                <TokioTime as hickory_server::net::runtime::Time>::current_time(),
+                response_handle,
+            )
             .await
     }
 
@@ -237,7 +240,12 @@ impl Server {
         self.catalog
             .read()
             .await
-            .lookup(request, response_edns, response_handle)
+            .lookup(
+                request,
+                response_edns.as_ref(),
+                <TokioTime as hickory_server::net::runtime::Time>::current_time(),
+                response_handle,
+            )
             .await
     }
 
@@ -257,11 +265,14 @@ mod tests {
         GeneralConfigBuilder, RecordBuilder, RecordType, RunConfigBuilder,
     };
     use anyhow::Result;
-    use hickory_client::client::{Client, ClientHandle};
     use hickory_proto::rr;
-    use hickory_proto::runtime::TokioRuntimeProvider;
-    use hickory_proto::udp::UdpClientStream;
+    use hickory_resolver::TokioResolver;
+    use hickory_resolver::config::{
+        ConnectionConfig, NameServerConfig, ResolverConfig, ResolverOpts,
+    };
+    use hickory_resolver::net::runtime::TokioRuntimeProvider;
     use maplit::hashmap;
+    use std::net::Ipv4Addr;
     use std::time::Duration;
 
     #[tokio::test]
@@ -314,23 +325,34 @@ mod tests {
         server.run().await?;
 
         let local_addr = server.udp_local_addr().unwrap();
-        let stream = UdpClientStream::builder(local_addr, TokioRuntimeProvider::default()).build();
-        let (mut client, background) = Client::connect(stream).await?;
-        let background_task = tokio::spawn(background);
-        let response = client
-            .query(
-                rr::Name::from_str("www.et.internal")?,
-                rr::DNSClass::IN,
-                rr::RecordType::A,
-            )
+        let mut connection = ConnectionConfig::udp();
+        connection.port = local_addr.port();
+        let resolver_config = ResolverConfig::from_parts(
+            None,
+            vec![],
+            vec![NameServerConfig::new(
+                local_addr.ip(),
+                true,
+                vec![connection],
+            )],
+        );
+        let resolver = TokioResolver::builder_with_config(
+            resolver_config,
+            TokioRuntimeProvider::default(),
+        )
+        .with_options(ResolverOpts::default())
+        .build()?;
+        let response = resolver
+            .lookup(rr::Name::from_str("www.et.internal")?, rr::RecordType::A)
             .await?;
-        drop(background_task);
 
         println!("Response: {:?}", response);
 
         assert_eq!(response.answers().len(), 1);
-        let expected_record: rr::Record = configured_record.try_into()?;
-        assert_eq!(response.answers().first().unwrap(), &expected_record);
+        let Some(rr::RData::A(ip)) = response.answers().first().map(|record| &record.data) else {
+            panic!("unexpected DNS response: {response:?}");
+        };
+        assert_eq!(ip.0, Ipv4Addr::new(123, 123, 123, 123));
 
         server.shutdown().await?;
         Ok(())

easytier/src/instance/dns_server/server_instance.rs

@@ -39,9 +39,10 @@ use anyhow::Context;
 use cidr::Ipv4Inet;
 use dashmap::DashMap;
 use hickory_proto::rr::LowerName;
-use hickory_proto::serialize::binary::{BinDecodable, BinEncoder};
-use hickory_server::authority::{MessageRequest, MessageResponse};
+use hickory_proto::serialize::binary::BinEncoder;
+use hickory_server::net::{NetError, udp as dns_udp, xfer::Protocol};
 use hickory_server::server::{Request, RequestHandler, ResponseHandler, ResponseInfo};
+use hickory_server::zone_handler::MessageResponse;
 use multimap::MultiMap;
 use pnet::packet::icmp::{IcmpTypes, MutableIcmpPacket};
 use pnet::packet::ipv4::Ipv4Packet;
@@ -54,7 +55,7 @@ use pnet::packet::{
 };
 use std::net::{SocketAddr, SocketAddrV4};
 use std::sync::Mutex;
-use std::{collections::BTreeMap, io, net::Ipv4Addr, str::FromStr, sync::Arc, time::Duration};
+use std::{collections::BTreeMap, net::Ipv4Addr, str::FromStr, sync::Arc, time::Duration};
 
 static NIC_PIPELINE_NAME: &str = "magic_dns_server";
 
@@ -266,25 +267,25 @@ impl ResponseHandler for ResponseWrapper {
             impl RecordIter<'a>,
             impl RecordIter<'a>,
         >,
-    ) -> io::Result<ResponseInfo> {
+    ) -> Result<ResponseInfo, NetError> {
         let mut buffer = self
             .response
             .lock()
-            .map_err(|_| io::Error::other("lock poisoned"))?;
+            .map_err(|_| NetError::Msg("lock poisoned".to_string()))?;
+        buffer.clear();
 
         let mut encoder = BinEncoder::new(&mut buffer);
 
         // `max_size` should be u16::MAX for protocol other than UDP.
-        let max_size = if let Some(edns) = response.get_edns() {
-            edns.max_payload()
-        } else {
-            hickory_proto::udp::MAX_RECEIVE_BUFFER_SIZE as u16
-        };
+        let max_size = response
+            .edns()
+            .map(|edns| edns.max_payload())
+            .unwrap_or(dns_udp::MAX_RECEIVE_BUFFER_SIZE as u16);
 
         encoder.set_max_size(max_size);
         response
             .destructive_emit(&mut encoder)
-            .map_err(io::Error::other)
+            .map_err(NetError::from)
     }
 }
 
@@ -360,11 +361,12 @@ impl MagicDnsServerInstanceData {
             (
                 src_port,
                 dst_port,
-                Request::new(
-                    MessageRequest::from_bytes(request_payload).ok()?,
+                Request::from_bytes(
+                    request_payload.to_vec(),
                     SocketAddr::from(SocketAddrV4::new(src_ip, src_port)),
-                    hickory_proto::xfer::Protocol::Udp,
-                ),
+                    Protocol::Udp,
+                )
+                .ok()?,
                 request_payload.len(),
             )
         };
@@ -375,7 +377,7 @@ impl MagicDnsServerInstanceData {
             self.dns_server
                 .read_catalog()
                 .await
-                .handle_request(
+                .handle_request::<ResponseWrapper, hickory_server::net::runtime::TokioTime>(
                     &request,
                     ResponseWrapper {
                         response: response_payload_arc.clone(),

easytier/src/instance/dns_server/tests.rs

@@ -1,13 +1,16 @@
-use std::net::{Ipv4Addr, SocketAddr};
+use std::net::Ipv4Addr;
 use std::str::FromStr as _;
 use std::sync::Arc;
 use std::time::Duration;
 
 use cidr::Ipv4Inet;
-use hickory_client::client::{Client, ClientHandle as _};
 use hickory_proto::rr;
-use hickory_proto::runtime::TokioRuntimeProvider;
-use hickory_proto::udp::UdpClientStream;
+use hickory_resolver::TokioResolver;
+use hickory_resolver::config::{
+    ConnectionConfig, NameServerConfig, ResolverConfig, ResolverOpts,
+};
+use hickory_resolver::net::runtime::TokioRuntimeProvider;
+use hickory_resolver::net::{DnsError, NetError};
 use tokio::sync::Notify;
 use tokio_util::sync::CancellationToken;
 
@@ -66,55 +69,56 @@ pub async fn prepare_env_with_tld_dns_zone(
 }
 
 pub async fn check_dns_record(fake_ip: &Ipv4Addr, domain: &str, expected_ip: &str) {
-    let stream = UdpClientStream::builder(
-        SocketAddr::new((*fake_ip).into(), 53),
-        TokioRuntimeProvider::default(),
-    )
-    .build();
-    let (mut client, background) = Client::connect(stream).await.unwrap();
-    let background_task = tokio::spawn(background);
-    let response = client
-        .query(
-            rr::Name::from_str(domain).unwrap(),
-            rr::DNSClass::IN,
-            rr::RecordType::A,
-        )
+    let resolver = build_test_resolver(fake_ip);
+    let response = resolver
+        .lookup(rr::Name::from_str(domain).unwrap(), rr::RecordType::A)
         .await
         .unwrap_or_else(|e| panic!("DNS query failed unexpectedly for domain '{domain}': {e}"));
-    background_task.abort();
-    let _ = background_task.await;
 
     println!("Response: {:?}", response);
 
-    assert_eq!(response.answers().len(), 1, "{:?}", response.answers());
+    assert_eq!(response.answers().len(), 1, "{:?}", response);
     let resp = response.answers().first().unwrap();
+    let rr::RData::A(ip) = &resp.data else {
+        panic!("unexpected DNS response: {response:?}");
+    };
     assert_eq!(
-        resp.clone().into_parts().rdata.into_a().unwrap().0,
+        ip.0,
         expected_ip.parse::<Ipv4Addr>().unwrap()
     );
 }
 
 pub async fn check_dns_record_missing(fake_ip: &Ipv4Addr, domain: &str) {
-    let stream = UdpClientStream::builder(
-        SocketAddr::new((*fake_ip).into(), 53),
-        TokioRuntimeProvider::default(),
-    )
-    .build();
-    let (mut client, background) = Client::connect(stream).await.unwrap();
-    let background_task = tokio::spawn(background);
-    let response = client
-        .query(
-            rr::Name::from_str(domain).unwrap(),
-            rr::DNSClass::IN,
-            rr::RecordType::A,
-        )
-        .await
-        .unwrap_or_else(|e| {
+    let resolver = build_test_resolver(fake_ip);
+    let response = resolver
+        .lookup(rr::Name::from_str(domain).unwrap(), rr::RecordType::A)
+        .await;
+    match response {
+        Ok(response) => assert!(response.answers().is_empty(), "{:?}", response),
+        Err(NetError::Dns(DnsError::NoRecordsFound(_))) => {}
+        Err(e) => {
             panic!("DNS query for missing record failed unexpectedly for domain '{domain}': {e}")
-        });
-    background_task.abort();
-    let _ = background_task.await;
-    assert!(response.answers().is_empty(), "{:?}", response.answers());
+        }
+    }
+}
+
+fn build_test_resolver(fake_ip: &Ipv4Addr) -> TokioResolver {
+    let mut connection = ConnectionConfig::udp();
+    connection.port = 53;
+    let config = ResolverConfig::from_parts(
+        None,
+        vec![],
+        vec![NameServerConfig::new(
+            (*fake_ip).into(),
+            true,
+            vec![connection],
+        )],
+    );
+
+    TokioResolver::builder_with_config(config, TokioRuntimeProvider::default())
+        .with_options(ResolverOpts::default())
+        .build()
+        .unwrap()
 }
 
 #[tokio::test]

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