fix: address dns review issues

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

@@ -236,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 }
 
@@ -401,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/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/common/stats_manager.rs

@@ -1,9 +1,8 @@
 use dashmap::DashMap;
-use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
+use std::cell::UnsafeCell;
 use std::fmt;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, Instant};
 use tokio::time::interval;
 use tokio_util::task::AbortOnDropHandle;
@@ -23,8 +22,6 @@ pub enum MetricName {
     PeerRpcDuration,
     /// RPC errors
     PeerRpcErrors,
-    /// RPC/control packets dropped because the peer RPC queue is unavailable
-    PeerRpcPacketQueueDrops,
 
     /// Data-plane traffic bytes sent
     TrafficBytesTx,
@@ -118,7 +115,6 @@ impl fmt::Display for MetricName {
             MetricName::PeerRpcServerRx => write!(f, "peer_rpc_server_rx"),
             MetricName::PeerRpcDuration => write!(f, "peer_rpc_duration_ms"),
             MetricName::PeerRpcErrors => write!(f, "peer_rpc_errors"),
-            MetricName::PeerRpcPacketQueueDrops => write!(f, "peer_rpc_packet_queue_drops"),
 
             MetricName::TrafficBytesTx => write!(f, "traffic_bytes_tx"),
             MetricName::TrafficBytesTxByInstance => write!(f, "traffic_bytes_tx_by_instance"),
@@ -378,10 +374,10 @@ impl Default for LabelSet {
     }
 }
 
-/// UnsafeCounter provides a high-performance atomic counter
+/// UnsafeCounter provides a high-performance counter using UnsafeCell
 #[derive(Debug)]
 pub struct UnsafeCounter {
-    value: AtomicU64,
+    value: UnsafeCell<u64>,
 }
 
 impl Default for UnsafeCounter {
@@ -393,79 +389,121 @@ impl Default for UnsafeCounter {
 impl UnsafeCounter {
     pub fn new() -> Self {
         Self {
-            value: AtomicU64::new(0),
+            value: UnsafeCell::new(0),
         }
     }
 
     pub fn new_with_value(initial: u64) -> Self {
         Self {
-            value: AtomicU64::new(initial),
+            value: UnsafeCell::new(initial),
         }
     }
 
     /// Increment the counter by the given amount
-    pub fn add(&self, delta: u64) {
-        let _ = self
-            .value
-            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
-                Some(current.saturating_add(delta))
-            });
+    /// # Safety
+    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
+    /// that no other thread is accessing this counter simultaneously.
+    pub unsafe fn add(&self, delta: u64) {
+        let ptr = self.value.get();
+        unsafe {
+            *ptr = (*ptr).saturating_add(delta);
+        }
     }
 
     /// Increment the counter by 1
-    pub fn inc(&self) {
-        self.add(1);
+    /// # Safety
+    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
+    /// that no other thread is accessing this counter simultaneously.
+    pub unsafe fn inc(&self) {
+        unsafe {
+            self.add(1);
+        }
     }
 
     /// Get the current value of the counter
-    pub fn get(&self) -> u64 {
-        self.value.load(Ordering::Relaxed)
+    /// # Safety
+    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
+    /// that no other thread is modifying this counter simultaneously.
+    pub unsafe fn get(&self) -> u64 {
+        let ptr = self.value.get();
+        unsafe { *ptr }
     }
 
     /// Reset the counter to zero
-    pub fn reset(&self) {
-        self.value.store(0, Ordering::Relaxed);
+    /// # Safety
+    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
+    /// that no other thread is accessing this counter simultaneously.
+    pub unsafe fn reset(&self) {
+        let ptr = self.value.get();
+        unsafe {
+            *ptr = 0;
+        }
     }
 
     /// Set the counter to a specific value
-    pub fn set(&self, value: u64) {
-        self.value.store(value, Ordering::Relaxed);
+    /// # Safety
+    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
+    /// that no other thread is accessing this counter simultaneously.
+    pub unsafe fn set(&self, value: u64) {
+        let ptr = self.value.get();
+        unsafe {
+            *ptr = value;
+        }
     }
 }
 
+// UnsafeCounter is Send + Sync because the safety is guaranteed by the caller
+unsafe impl Send for UnsafeCounter {}
+unsafe impl Sync for UnsafeCounter {}
+
 /// MetricData contains both the counter and last update timestamp
+/// Uses UnsafeCell for lock-free access
 #[derive(Debug)]
 struct MetricData {
     counter: UnsafeCounter,
-    last_updated: Mutex<Instant>,
+    last_updated: UnsafeCell<Instant>,
 }
 
 impl MetricData {
     fn new() -> Self {
         Self {
             counter: UnsafeCounter::new(),
-            last_updated: Mutex::new(Instant::now()),
+            last_updated: UnsafeCell::new(Instant::now()),
         }
     }
 
     fn new_with_value(initial: u64) -> Self {
         Self {
             counter: UnsafeCounter::new_with_value(initial),
-            last_updated: Mutex::new(Instant::now()),
+            last_updated: UnsafeCell::new(Instant::now()),
         }
     }
 
     /// Update the last_updated timestamp
-    fn touch(&self) {
-        *self.last_updated.lock() = Instant::now();
+    /// # Safety
+    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
+    /// that no other thread is accessing this timestamp simultaneously.
+    unsafe fn touch(&self) {
+        let ptr = self.last_updated.get();
+        unsafe {
+            *ptr = Instant::now();
+        }
     }
 
     /// Get the last updated timestamp
-    fn get_last_updated(&self) -> Instant {
-        *self.last_updated.lock()
+    /// # Safety
+    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
+    /// that no other thread is modifying this timestamp simultaneously.
+    unsafe fn get_last_updated(&self) -> Instant {
+        let ptr = self.last_updated.get();
+        unsafe { *ptr }
     }
 }
 
+// MetricData is Send + Sync because the safety is guaranteed by the caller
+unsafe impl Send for MetricData {}
+unsafe impl Sync for MetricData {}
+
 /// MetricKey uniquely identifies a metric with its name and labels
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct MetricKey {
@@ -508,31 +546,39 @@ impl CounterHandle {
 
     /// Increment the counter by the given amount
     pub fn add(&self, delta: u64) {
-        self.metric_data.counter.add(delta);
-        self.metric_data.touch();
+        unsafe {
+            self.metric_data.counter.add(delta);
+            self.metric_data.touch();
+        }
     }
 
     /// Increment the counter by 1
     pub fn inc(&self) {
-        self.metric_data.counter.inc();
-        self.metric_data.touch();
+        unsafe {
+            self.metric_data.counter.inc();
+            self.metric_data.touch();
+        }
     }
 
     /// Get the current value of the counter
     pub fn get(&self) -> u64 {
-        self.metric_data.counter.get()
+        unsafe { self.metric_data.counter.get() }
     }
 
     /// Reset the counter to zero
     pub fn reset(&self) {
-        self.metric_data.counter.reset();
-        self.metric_data.touch();
+        unsafe {
+            self.metric_data.counter.reset();
+            self.metric_data.touch();
+        }
     }
 
     /// Set the counter to a specific value
     pub fn set(&self, value: u64) {
-        self.metric_data.counter.set(value);
-        self.metric_data.touch();
+        unsafe {
+            self.metric_data.counter.set(value);
+            self.metric_data.touch();
+        }
     }
 }
 
@@ -578,7 +624,7 @@ impl StatsManager {
 
                 counters.retain(|_, metric_data: &mut Arc<MetricData>| {
                     Arc::strong_count(metric_data) > 1
-                        || metric_data.get_last_updated() > cutoff_time
+                        || unsafe { metric_data.get_last_updated() > cutoff_time }
                 });
                 counters.shrink_to_fit();
             }
@@ -616,7 +662,7 @@ impl StatsManager {
             let key = entry.key();
             let metric_data = entry.value();
 
-            let value = metric_data.counter.get();
+            let value = unsafe { metric_data.counter.get() };
 
             metrics.push(MetricSnapshot {
                 name: key.name,
@@ -649,7 +695,7 @@ impl StatsManager {
         let key = MetricKey::new(name, labels.clone());
 
         if let Some(metric_data) = self.counters.get(&key) {
-            let value = metric_data.counter.get();
+            let value = unsafe { metric_data.counter.get() };
             Some(MetricSnapshot {
                 name,
                 labels: labels.clone(),
@@ -750,15 +796,17 @@ mod tests {
     async fn test_unsafe_counter() {
         let counter = UnsafeCounter::new();
 
-        assert_eq!(counter.get(), 0);
-        counter.inc();
-        assert_eq!(counter.get(), 1);
-        counter.add(5);
-        assert_eq!(counter.get(), 6);
-        counter.set(10);
-        assert_eq!(counter.get(), 10);
-        counter.reset();
-        assert_eq!(counter.get(), 0);
+        unsafe {
+            assert_eq!(counter.get(), 0);
+            counter.inc();
+            assert_eq!(counter.get(), 1);
+            counter.add(5);
+            assert_eq!(counter.get(), 6);
+            counter.set(10);
+            assert_eq!(counter.get(), 10);
+            counter.reset();
+            assert_eq!(counter.get(), 0);
+        }
     }
 
     #[tokio::test]
@@ -903,7 +951,8 @@ mod tests {
         stats
             .counters
             .retain(|_, metric_data: &mut Arc<MetricData>| {
-                Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > cutoff_time
+                Arc::strong_count(metric_data) > 1
+                    || unsafe { metric_data.get_last_updated() > cutoff_time }
             });
 
         assert_eq!(stats.metric_count(), 1);
@@ -913,33 +962,12 @@ mod tests {
         stats
             .counters
             .retain(|_, metric_data: &mut Arc<MetricData>| {
-                Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > cutoff_time
+                Arc::strong_count(metric_data) > 1
+                    || unsafe { metric_data.get_last_updated() > cutoff_time }
             });
         assert_eq!(stats.metric_count(), 0);
     }
 
-    #[tokio::test]
-    async fn test_counter_handle_concurrent_increment() {
-        const THREADS: usize = 8;
-        const INCREMENTS_PER_THREAD: usize = 10_000;
-
-        let stats = StatsManager::new();
-        let counter = stats.get_simple_counter(MetricName::TrafficPacketsForwarded);
-
-        std::thread::scope(|scope| {
-            for _ in 0..THREADS {
-                let counter = counter.clone();
-                scope.spawn(move || {
-                    for _ in 0..INCREMENTS_PER_THREAD {
-                        counter.inc();
-                    }
-                });
-            }
-        });
-
-        assert_eq!(counter.get(), (THREADS * INCREMENTS_PER_THREAD) as u64);
-    }
-
     #[tokio::test]
     async fn test_stats_rpc_data_structures() {
         // Test GetStatsRequest

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

@@ -18,7 +18,7 @@ use guarden::{Guard, defer};
 use tokio::{
     sync::{
         Mutex,
-        mpsc::{self, Receiver, Sender, error::TrySendError},
+        mpsc::{self, UnboundedReceiver, UnboundedSender},
     },
     task::JoinSet,
 };
@@ -30,7 +30,7 @@ use crate::{
         error::Error,
         global_ctx::{ArcGlobalCtx, GlobalCtx, GlobalCtxEvent, NetworkIdentity, TrustedKeySource},
         join_joinset_background, shrink_dashmap,
-        stats_manager::{CounterHandle, LabelSet, LabelType, MetricName, StatsManager},
+        stats_manager::{LabelSet, LabelType, MetricName, StatsManager},
         token_bucket::TokenBucket,
     },
     peer_center::instance::{PeerCenterInstance, PeerMapWithPeerRpcManager},
@@ -64,35 +64,6 @@ use super::{
     },
 };
 
-const PEER_RPC_PACKET_QUEUE_CAPACITY: usize = 1024;
-
-fn try_enqueue_peer_rpc_packet(
-    sender: &Sender<ZCPacket>,
-    packet: ZCPacket,
-    dropped_packets: &CounterHandle,
-    queue_name: &'static str,
-) -> bool {
-    match sender.try_send(packet) {
-        Ok(()) => true,
-        Err(TrySendError::Full(_)) => {
-            dropped_packets.inc();
-            tracing::warn!(
-                queue = queue_name,
-                "drop peer rpc/control packet because queue is full"
-            );
-            false
-        }
-        Err(TrySendError::Closed(_)) => {
-            dropped_packets.inc();
-            tracing::warn!(
-                queue = queue_name,
-                "drop peer rpc/control packet because receiver is closed"
-            );
-            false
-        }
-    }
-}
-
 #[async_trait::async_trait]
 #[auto_impl::auto_impl(&, Box, Arc)]
 pub trait GlobalForeignNetworkAccessor: Send + Sync + 'static {
@@ -116,7 +87,7 @@ struct ForeignNetworkEntry {
     pm_packet_sender: Mutex<Option<PacketRecvChan>>,
 
     peer_rpc: Arc<PeerRpcManager>,
-    rpc_sender: Sender<ZCPacket>,
+    rpc_sender: UnboundedSender<ZCPacket>,
 
     packet_recv: Mutex<Option<PacketRecvChanReceiver>>,
 
@@ -341,12 +312,12 @@ impl ForeignNetworkEntry {
     fn build_rpc_tspt(
         my_peer_id: PeerId,
         peer_map: Arc<PeerMap>,
-    ) -> (Arc<PeerRpcManager>, Sender<ZCPacket>) {
+    ) -> (Arc<PeerRpcManager>, UnboundedSender<ZCPacket>) {
         struct RpcTransport {
             my_peer_id: PeerId,
             peer_map: Weak<PeerMap>,
 
-            packet_recv: Mutex<Receiver<ZCPacket>>,
+            packet_recv: Mutex<UnboundedReceiver<ZCPacket>>,
         }
 
         #[async_trait::async_trait]
@@ -388,8 +359,7 @@ impl ForeignNetworkEntry {
             }
         }
 
-        let (rpc_transport_sender, peer_rpc_tspt_recv) =
-            mpsc::channel(PEER_RPC_PACKET_QUEUE_CAPACITY);
+        let (rpc_transport_sender, peer_rpc_tspt_recv) = mpsc::unbounded_channel();
         let tspt = RpcTransport {
             my_peer_id,
             peer_map: Arc::downgrade(&peer_map),
@@ -508,9 +478,6 @@ impl ForeignNetworkEntry {
         let rx_packets = self
             .stats_mgr
             .get_counter(MetricName::TrafficPacketsRx, label_set.clone());
-        let rpc_queue_drops = self
-            .stats_mgr
-            .get_counter(MetricName::PeerRpcPacketQueueDrops, label_set.clone());
 
         self.tasks.lock().await.spawn(async move {
             while let Ok(mut zc_packet) = recv_packet_from_chan(&mut recv).await {
@@ -559,12 +526,7 @@ impl ForeignNetworkEntry {
                     {
                         rx_bytes.add(buf_len as u64);
                         rx_packets.inc();
-                        try_enqueue_peer_rpc_packet(
-                            &rpc_sender,
-                            zc_packet,
-                            &rpc_queue_drops,
-                            "foreign_network_peer_rpc",
-                        );
+                        rpc_sender.send(zc_packet).unwrap();
                         continue;
                     }
                     tracing::trace!(
@@ -1274,7 +1236,7 @@ impl Drop for ForeignNetworkManager {
 pub mod tests {
     use crate::{
         common::global_ctx::tests::get_mock_global_ctx_with_network,
-        common::stats_manager::{LabelSet, LabelType, MetricName, StatsManager},
+        common::stats_manager::{LabelSet, LabelType, MetricName},
         connector::udp_hole_punch::tests::{
             create_mock_peer_manager_with_mock_stun, replace_stun_info_collector,
         },
@@ -1291,7 +1253,6 @@ pub mod tests {
         },
     };
     use std::{collections::HashMap, time::Duration};
-    use tokio::sync::mpsc;
 
     use super::*;
 
@@ -1304,59 +1265,6 @@ pub mod tests {
             .unwrap_or(0)
     }
 
-    #[tokio::test]
-    async fn peer_rpc_queue_helper_enqueues_when_available() {
-        let (sender, mut receiver) = mpsc::channel(1);
-        let stats_manager = StatsManager::new();
-        let dropped_packets = stats_manager.get_simple_counter(MetricName::PeerRpcPacketQueueDrops);
-
-        assert!(try_enqueue_peer_rpc_packet(
-            &sender,
-            ZCPacket::new_with_payload(b"rpc"),
-            &dropped_packets,
-            "test_foreign_peer_rpc",
-        ));
-
-        assert_eq!(dropped_packets.get(), 0);
-        assert!(receiver.try_recv().is_ok());
-    }
-
-    #[tokio::test]
-    async fn peer_rpc_queue_helper_drops_when_full() {
-        let (sender, _receiver) = mpsc::channel(1);
-        sender
-            .try_send(ZCPacket::new_with_payload(b"existing"))
-            .unwrap();
-        let stats_manager = StatsManager::new();
-        let dropped_packets = stats_manager.get_simple_counter(MetricName::PeerRpcPacketQueueDrops);
-
-        assert!(!try_enqueue_peer_rpc_packet(
-            &sender,
-            ZCPacket::new_with_payload(b"overflow"),
-            &dropped_packets,
-            "test_foreign_peer_rpc",
-        ));
-
-        assert_eq!(dropped_packets.get(), 1);
-    }
-
-    #[tokio::test]
-    async fn peer_rpc_queue_helper_drops_when_closed() {
-        let (sender, receiver) = mpsc::channel(1);
-        drop(receiver);
-        let stats_manager = StatsManager::new();
-        let dropped_packets = stats_manager.get_simple_counter(MetricName::PeerRpcPacketQueueDrops);
-
-        assert!(!try_enqueue_peer_rpc_packet(
-            &sender,
-            ZCPacket::new_with_payload(b"closed"),
-            &dropped_packets,
-            "test_foreign_peer_rpc",
-        ));
-
-        assert_eq!(dropped_packets.get(), 1);
-    }
-
     async fn create_mock_peer_manager_for_foreign_network_ext(
         network: &str,
         secret: &str,

easytier/src/peers/peer_manager.rs

@@ -13,7 +13,7 @@ use std::{
 use tokio::{
     sync::{
         Mutex, RwLock,
-        mpsc::{self, Receiver, Sender, error::TrySendError},
+        mpsc::{self, UnboundedReceiver, UnboundedSender},
     },
     task::JoinSet,
 };
@@ -72,43 +72,14 @@ use super::{
     route_trait::{ArcRoute, Route},
 };
 
-const PEER_RPC_PACKET_QUEUE_CAPACITY: usize = 1024;
-
-fn try_enqueue_peer_rpc_packet(
-    sender: &Sender<ZCPacket>,
-    packet: ZCPacket,
-    dropped_packets: &CounterHandle,
-    queue_name: &'static str,
-) -> bool {
-    match sender.try_send(packet) {
-        Ok(()) => true,
-        Err(TrySendError::Full(_)) => {
-            dropped_packets.inc();
-            tracing::warn!(
-                queue = queue_name,
-                "drop peer rpc/control packet because queue is full"
-            );
-            false
-        }
-        Err(TrySendError::Closed(_)) => {
-            dropped_packets.inc();
-            tracing::warn!(
-                queue = queue_name,
-                "drop peer rpc/control packet because receiver is closed"
-            );
-            false
-        }
-    }
-}
-
 struct RpcTransport {
     my_peer_id: PeerId,
     peers: Weak<PeerMap>,
     // TODO: this seems can be removed
     foreign_peers: Mutex<Option<Weak<ForeignNetworkClient>>>,
 
-    packet_recv: Mutex<Receiver<ZCPacket>>,
-    peer_rpc_tspt_sender: Sender<ZCPacket>,
+    packet_recv: Mutex<UnboundedReceiver<ZCPacket>>,
+    peer_rpc_tspt_sender: UnboundedSender<ZCPacket>,
 
     encryptor: Arc<dyn Encryptor>,
     is_secure_mode_enabled: bool,
@@ -302,8 +273,7 @@ impl PeerManager {
             .unwrap_or(false);
 
         // TODO: remove these because we have impl pipeline processor.
-        let (peer_rpc_tspt_sender, peer_rpc_tspt_recv) =
-            mpsc::channel(PEER_RPC_PACKET_QUEUE_CAPACITY);
+        let (peer_rpc_tspt_sender, peer_rpc_tspt_recv) = mpsc::unbounded_channel();
         let rpc_tspt = Arc::new(RpcTransport {
             my_peer_id,
             peers: Arc::downgrade(&peers),
@@ -1273,8 +1243,7 @@ impl PeerManager {
 
         // for peer rpc packet
         struct PeerRpcPacketProcessor {
-            peer_rpc_tspt_sender: Sender<ZCPacket>,
-            dropped_packets: CounterHandle,
+            peer_rpc_tspt_sender: UnboundedSender<ZCPacket>,
         }
 
         #[async_trait::async_trait]
@@ -1285,27 +1254,15 @@ impl PeerManager {
                     || hdr.packet_type == PacketType::RpcReq as u8
                     || hdr.packet_type == PacketType::RpcResp as u8
                 {
-                    try_enqueue_peer_rpc_packet(
-                        &self.peer_rpc_tspt_sender,
-                        packet,
-                        &self.dropped_packets,
-                        "local_peer_rpc",
-                    );
+                    self.peer_rpc_tspt_sender.send(packet).unwrap();
                     None
                 } else {
                     Some(packet)
                 }
             }
         }
-        let peer_rpc_queue_drops = self.global_ctx.stats_manager().get_counter(
-            MetricName::PeerRpcPacketQueueDrops,
-            LabelSet::new().with_label_type(LabelType::NetworkName(
-                self.global_ctx.get_network_name().to_string(),
-            )),
-        );
         self.add_packet_process_pipeline(Box::new(PeerRpcPacketProcessor {
             peer_rpc_tspt_sender: self.peer_rpc_tspt.peer_rpc_tspt_sender.clone(),
-            dropped_packets: peer_rpc_queue_drops,
         }))
         .await;
     }
@@ -2252,7 +2209,7 @@ mod tests {
             PeerId,
             config::Flags,
             global_ctx::{NetworkIdentity, tests::get_mock_global_ctx},
-            stats_manager::{LabelSet, LabelType, MetricName, StatsManager},
+            stats_manager::{LabelSet, LabelType, MetricName},
         },
         connector::{
             create_connector_by_url, direct::PeerManagerForDirectConnector,
@@ -2283,9 +2240,7 @@ mod tests {
         },
     };
 
-    use tokio::sync::mpsc;
-
-    use super::{PeerManager, try_enqueue_peer_rpc_packet};
+    use super::PeerManager;
 
     async fn create_lazy_peer_manager() -> Arc<PeerManager> {
         let peer_mgr = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
@@ -2310,59 +2265,6 @@ mod tests {
         ))
     }
 
-    #[tokio::test]
-    async fn peer_rpc_queue_helper_enqueues_when_available() {
-        let (sender, mut receiver) = mpsc::channel(1);
-        let stats_manager = StatsManager::new();
-        let dropped_packets = stats_manager.get_simple_counter(MetricName::PeerRpcPacketQueueDrops);
-
-        assert!(try_enqueue_peer_rpc_packet(
-            &sender,
-            ZCPacket::new_with_payload(b"rpc"),
-            &dropped_packets,
-            "test_peer_rpc",
-        ));
-
-        assert_eq!(dropped_packets.get(), 0);
-        assert!(receiver.try_recv().is_ok());
-    }
-
-    #[tokio::test]
-    async fn peer_rpc_queue_helper_drops_when_full() {
-        let (sender, _receiver) = mpsc::channel(1);
-        sender
-            .try_send(ZCPacket::new_with_payload(b"existing"))
-            .unwrap();
-        let stats_manager = StatsManager::new();
-        let dropped_packets = stats_manager.get_simple_counter(MetricName::PeerRpcPacketQueueDrops);
-
-        assert!(!try_enqueue_peer_rpc_packet(
-            &sender,
-            ZCPacket::new_with_payload(b"overflow"),
-            &dropped_packets,
-            "test_peer_rpc",
-        ));
-
-        assert_eq!(dropped_packets.get(), 1);
-    }
-
-    #[tokio::test]
-    async fn peer_rpc_queue_helper_drops_when_closed() {
-        let (sender, receiver) = mpsc::channel(1);
-        drop(receiver);
-        let stats_manager = StatsManager::new();
-        let dropped_packets = stats_manager.get_simple_counter(MetricName::PeerRpcPacketQueueDrops);
-
-        assert!(!try_enqueue_peer_rpc_packet(
-            &sender,
-            ZCPacket::new_with_payload(b"closed"),
-            &dropped_packets,
-            "test_peer_rpc",
-        ));
-
-        assert_eq!(dropped_packets.get(), 1);
-    }
-
     struct TestCostCalculator {
         costs: HashMap<(PeerId, PeerId), i32>,
     }

easytier/src/tunnel/stats.rs

@@ -1,4 +1,7 @@
-use std::sync::atomic::{AtomicU32, AtomicU64, Ordering::Relaxed};
+use std::{
+    cell::UnsafeCell,
+    sync::atomic::{AtomicU32, Ordering::Relaxed},
+};
 
 pub struct WindowLatency {
     latency_us_window: Vec<AtomicU32>,
@@ -60,30 +63,34 @@ impl WindowLatency {
 
 #[derive(Debug)]
 pub struct Throughput {
-    tx_bytes: AtomicU64,
-    rx_bytes: AtomicU64,
-    tx_packets: AtomicU64,
-    rx_packets: AtomicU64,
+    tx_bytes: UnsafeCell<u64>,
+    rx_bytes: UnsafeCell<u64>,
+    tx_packets: UnsafeCell<u64>,
+    rx_packets: UnsafeCell<u64>,
 }
 
 impl Clone for Throughput {
     fn clone(&self) -> Self {
         Self {
-            tx_bytes: AtomicU64::new(self.tx_bytes()),
-            rx_bytes: AtomicU64::new(self.rx_bytes()),
-            tx_packets: AtomicU64::new(self.tx_packets()),
-            rx_packets: AtomicU64::new(self.rx_packets()),
+            tx_bytes: UnsafeCell::new(unsafe { *self.tx_bytes.get() }),
+            rx_bytes: UnsafeCell::new(unsafe { *self.rx_bytes.get() }),
+            tx_packets: UnsafeCell::new(unsafe { *self.tx_packets.get() }),
+            rx_packets: UnsafeCell::new(unsafe { *self.rx_packets.get() }),
         }
     }
 }
 
+// add sync::Send and sync::Sync traits to Throughput
+unsafe impl Send for Throughput {}
+unsafe impl Sync for Throughput {}
+
 impl Default for Throughput {
     fn default() -> Self {
         Self {
-            tx_bytes: AtomicU64::new(0),
-            rx_bytes: AtomicU64::new(0),
-            tx_packets: AtomicU64::new(0),
-            rx_packets: AtomicU64::new(0),
+            tx_bytes: UnsafeCell::new(0),
+            rx_bytes: UnsafeCell::new(0),
+            tx_packets: UnsafeCell::new(0),
+            rx_packets: UnsafeCell::new(0),
         }
     }
 }
@@ -94,68 +101,32 @@ impl Throughput {
     }
 
     pub fn tx_bytes(&self) -> u64 {
-        self.tx_bytes.load(Relaxed)
+        unsafe { *self.tx_bytes.get() }
     }
 
     pub fn rx_bytes(&self) -> u64 {
-        self.rx_bytes.load(Relaxed)
+        unsafe { *self.rx_bytes.get() }
     }
 
     pub fn tx_packets(&self) -> u64 {
-        self.tx_packets.load(Relaxed)
+        unsafe { *self.tx_packets.get() }
     }
 
     pub fn rx_packets(&self) -> u64 {
-        self.rx_packets.load(Relaxed)
+        unsafe { *self.rx_packets.get() }
     }
 
     pub fn record_tx_bytes(&self, bytes: u64) {
-        self.tx_bytes.fetch_add(bytes, Relaxed);
-        self.tx_packets.fetch_add(1, Relaxed);
+        unsafe {
+            *self.tx_bytes.get() += bytes;
+            *self.tx_packets.get() += 1;
+        }
     }
 
     pub fn record_rx_bytes(&self, bytes: u64) {
-        self.rx_bytes.fetch_add(bytes, Relaxed);
-        self.rx_packets.fetch_add(1, Relaxed);
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::Throughput;
-    use std::sync::Arc;
-
-    #[test]
-    fn throughput_records_concurrent_tx_and_rx() {
-        const THREADS: usize = 8;
-        const RECORDS_PER_THREAD: usize = 10_000;
-        const TX_BYTES_PER_RECORD: u64 = 3;
-        const RX_BYTES_PER_RECORD: u64 = 7;
-
-        let throughput = Arc::new(Throughput::new());
-
-        std::thread::scope(|scope| {
-            for _ in 0..THREADS {
-                let throughput = Arc::clone(&throughput);
-                scope.spawn(move || {
-                    for _ in 0..RECORDS_PER_THREAD {
-                        throughput.record_tx_bytes(TX_BYTES_PER_RECORD);
-                        throughput.record_rx_bytes(RX_BYTES_PER_RECORD);
-                    }
-                });
-            }
-        });
-
-        let expected_packets = (THREADS * RECORDS_PER_THREAD) as u64;
-        assert_eq!(throughput.tx_packets(), expected_packets);
-        assert_eq!(throughput.rx_packets(), expected_packets);
-        assert_eq!(
-            throughput.tx_bytes(),
-            expected_packets * TX_BYTES_PER_RECORD
-        );
-        assert_eq!(
-            throughput.rx_bytes(),
-            expected_packets * RX_BYTES_PER_RECORD
-        );
+        unsafe {
+            *self.rx_bytes.get() += bytes;
+            *self.rx_packets.get() += 1;
+        }
     }
 }