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

The millis-since-base staleness check (last > now - 180s) saturated the
cutoff to 0 early in process life, so metrics stamped at 0ms failed the
strict > 0 test and were wrongly evicted by the immediate first GC
tick. Switch to age-based (now - last < 180s), equivalent to the
original Instant semantics and robust under clock saturation, which
fixes the flaky peer_conn_secure_mode_pubkey_and_encryption test.

.cargo/audit.toml

@@ -1,3 +0,0 @@
-[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.20", features = ["derive"] }
+validator = { version = "0.18", 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.27"
+maxminddb = "0.24"
 once_cell = "1.18"
 
 axum = { version = "0.7", features = ["macros"] }

easytier-web/src/client_manager/mod.rs

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

@@ -215,6 +215,7 @@ smoltcp = { git = "https://github.com/smoltcp-rs/smoltcp.git", rev = "0a926767a6
     "async",
 ] }
 parking_lot = { version = "0.12.0" }
+fastant = "0.1"
 
 wildmatch = "2.3.4"
 
@@ -236,11 +237,12 @@ http_req = { git = "https://github.com/EasyTier/http_req.git", default-features
 ] }
 
 # for dns connector
-hickory-resolver = "0.26.1"
-hickory-proto = "0.26.1"
+hickory-resolver = "0.25.2"
+hickory-proto = "0.25.2"
 
 # for magic dns
-hickory-server = { version = "0.26.1", features = [
+hickory-client = { version = "0.25.2", optional = true }
+hickory-server = { version = "0.25.2", features = [
     "resolver",
 ], optional = true }
 
@@ -340,6 +342,11 @@ futures-util = "0.3.31"
 maplit = "1.0.2"
 tempfile = "3.22.0"
 ctor = "0.8.0"
+criterion = { version = "0.5", features = ["html_reports"] }
+
+[[bench]]
+name = "counter_contention"
+harness = false
 
 [target.'cfg(target_os = "linux")'.dev-dependencies]
 defguard_wireguard_rs = "0.4.2"
@@ -400,7 +407,7 @@ jemalloc-prof = [
     "jemalloc-sys/stats",
 ]
 tracing = ["tokio/tracing", "dep:console-subscriber"]
-magic-dns = ["dep:hickory-server"]
+magic-dns = ["dep:hickory-client", "dep:hickory-server"]
 faketcp = ["dep:flume"]
 zstd = ["dep:zstd"]
 # For Network Extension on macOS

easytier/benches/counter_contention.rs

@@ -0,0 +1,443 @@
+//! Compare counter implementations under tokio-task contention.
+//!
+//! Groups:
+//!   - `contention_scaling` : N tokio tasks share one counter, total work fixed.
+//!     Variants: `single_atomic`, `cas_saturating`, `sharded_atomic`,
+//!     `thread_local_cell`, `unsafe_cell` (unsound, for reference only).
+//!   - `single_thread_write`: per-`add` cost with no contention (floor cost).
+//!   - `read_cost`          : per-`get()` cost.
+//!   - `counter_handle`     : the REAL production hot path. Measures the actual
+//!     `stats_manager::CounterHandle::add` (single-atomic `fetch_add` + lock-free
+//!     fastant `touch`) against reconstructed baselines:
+//!       * `prod` - real `CounterHandle` (this code's version)
+//!       * `baseline_cas_mutex` - pre-optimization: single `AtomicU64` with `fetch_update` (CAS) + `Mutex<Instant>` touch
+//!       * `baseline_fetchadd_mutex` - `fetch_add` + `Mutex<Instant>` touch (isolates the lock-free fastant touch)
+//!
+//! Run: `cargo bench -p easytier --bench counter_contention`
+
+use std::cell::{Cell, UnsafeCell};
+use std::hint::black_box;
+use std::sync::Arc;
+use std::sync::LazyLock;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::thread::available_parallelism;
+use std::time::Instant;
+
+use criterion::{
+    BenchmarkId, Criterion, Throughput, criterion_group, criterion_main, measurement::WallTime,
+};
+use easytier::common::stats_manager::{CounterHandle, MetricName, StatsManager};
+use parking_lot::Mutex;
+
+const COUNTER_SHARDS: usize = 16;
+const TOTAL_WORK: u64 = 8_000_000;
+// The handle path does a counter update plus a timestamp `touch` per `add`,
+// so it is heavier per op than the counter-only groups; use a smaller total to
+// keep the bench fast.
+const HANDLE_TOTAL_WORK: u64 = 2_000_000;
+const TASK_COUNTS: &[usize] = &[1, 2, 4, 8, 16, 32];
+
+trait Counter: Send + Sync {
+    fn add(&self, delta: u64);
+    fn get(&self) -> u64;
+}
+
+// ---------------------------------------------------------------------------
+// 1. SingleAtomic: one atomic, fetch_add. Baseline; contends across cores.
+// ---------------------------------------------------------------------------
+
+struct SingleAtomic(AtomicU64);
+
+impl Default for SingleAtomic {
+    fn default() -> Self {
+        Self(AtomicU64::new(0))
+    }
+}
+
+impl Counter for SingleAtomic {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        self.0.fetch_add(delta, Ordering::Relaxed);
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        self.0.load(Ordering::Relaxed)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 2. CasSaturating: fetch_update with saturating_add (the original PR `add`).
+//    A CAS loop that can retry under contention.
+// ---------------------------------------------------------------------------
+
+struct CasSaturating(AtomicU64);
+
+impl Default for CasSaturating {
+    fn default() -> Self {
+        Self(AtomicU64::new(0))
+    }
+}
+
+impl Counter for CasSaturating {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        let _ = self
+            .0
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| {
+                Some(c.saturating_add(delta))
+            });
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        self.0.load(Ordering::Relaxed)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 3. ShardedAtomic: 16 cache-aligned shards + per-thread shard index.
+//    Comparison-only variant (production `stats_manager::Counter` is
+//    single-atomic; sharding was evaluated and dropped as no benefit for the
+//    default 1-16 worker deployments).
+// ---------------------------------------------------------------------------
+
+thread_local! {
+    static SHARD_IDX: Cell<usize> = Cell::new({
+        static NEXT: AtomicUsize = AtomicUsize::new(0);
+        NEXT.fetch_add(1, Ordering::Relaxed) % COUNTER_SHARDS
+    });
+}
+
+#[repr(align(64))]
+struct Shard {
+    value: AtomicU64,
+}
+
+struct ShardedAtomic {
+    shards: Box<[Shard]>,
+}
+
+impl Default for ShardedAtomic {
+    fn default() -> Self {
+        let mut shards = Vec::with_capacity(COUNTER_SHARDS);
+        for _ in 0..COUNTER_SHARDS {
+            shards.push(Shard {
+                value: AtomicU64::new(0),
+            });
+        }
+        Self {
+            shards: shards.into_boxed_slice(),
+        }
+    }
+}
+
+impl Counter for ShardedAtomic {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        let i = SHARD_IDX.with(|c| c.get());
+        self.shards[i].value.fetch_add(delta, Ordering::Relaxed);
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        self.shards
+            .iter()
+            .map(|s| s.value.load(Ordering::Relaxed))
+            .sum()
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 4. ThreadLocalCell: per-thread Cell<u64> accumulation. Zero-atomic writes.
+//    `get()` flushes the caller thread's local into a shared aggregate, so the
+//    measured read cost reflects a flush-based read. Exact totals would require
+//    flushing every thread (not modeled here).
+// ---------------------------------------------------------------------------
+
+thread_local! {
+    static TLS_DELTA: Cell<u64> = const { Cell::new(0) };
+}
+
+struct ThreadLocalCell {
+    shared: AtomicU64,
+}
+
+impl Default for ThreadLocalCell {
+    fn default() -> Self {
+        Self {
+            shared: AtomicU64::new(0),
+        }
+    }
+}
+
+impl Counter for ThreadLocalCell {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        TLS_DELTA.with(|c| c.set(c.get() + delta));
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        let local = TLS_DELTA.with(|c| c.replace(0));
+        self.shared.fetch_add(local, Ordering::Relaxed) + local
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 5. UnsafeCellCounter: a plain u64 mutated through UnsafeCell with manual
+//    `unsafe impl Send/Sync`. This is UNSOUND under concurrent access (data
+//    race / UB) and is exactly what the original code did "for speed". It is
+//    included only to measure the speed ceiling the author was chasing, and to
+//    show that its `get()` returns wrong totals under contention (lost updates).
+// ---------------------------------------------------------------------------
+
+struct UnsafeCellCounter(UnsafeCell<u64>);
+
+// SAFETY: deliberately unsound; see above.
+unsafe impl Send for UnsafeCellCounter {}
+unsafe impl Sync for UnsafeCellCounter {}
+
+impl Default for UnsafeCellCounter {
+    fn default() -> Self {
+        Self(UnsafeCell::new(0))
+    }
+}
+
+impl Counter for UnsafeCellCounter {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        // SAFETY: UNSOUND under concurrent access (data race).
+        unsafe {
+            *self.0.get() += delta;
+        }
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        // SAFETY: UNSOUND under concurrent writers (data race).
+        unsafe { *self.0.get() }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Production counter handle + reconstructed baselines for the `counter_handle`
+// group. These measure the full hot path (`add` = counter update + `touch`
+// timestamp), which is what actually runs per packet in `peer_manager`.
+// ---------------------------------------------------------------------------
+
+// The real production `CounterHandle`. `CounterHandle::add` does a single
+// `AtomicU64::fetch_add` then a lock-free fastant `touch`.
+impl Counter for CounterHandle {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        // Fully-qualified to avoid infinite recursion through the trait method.
+        CounterHandle::add(self, delta);
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        CounterHandle::get(self)
+    }
+}
+
+// A faithful replica of the pre-optimization design: a SINGLE `AtomicU64`
+// (unsharded) plus a `Mutex<Instant>` timestamp. `use_cas` selects whether the
+// counter write is a `fetch_update` saturating CAS (the original PR `add`) or a
+// plain `fetch_add`.
+struct BaselineHandle {
+    counter: AtomicU64,
+    last_updated: Mutex<Instant>,
+    use_cas: bool,
+}
+
+impl Default for BaselineHandle {
+    fn default() -> Self {
+        Self {
+            counter: AtomicU64::new(0),
+            last_updated: Mutex::new(Instant::now()),
+            use_cas: false,
+        }
+    }
+}
+
+impl Counter for BaselineHandle {
+    #[inline(always)]
+    fn add(&self, delta: u64) {
+        if self.use_cas {
+            let _ = self
+                .counter
+                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| {
+                    Some(c.saturating_add(delta))
+                });
+        } else {
+            self.counter.fetch_add(delta, Ordering::Relaxed);
+        }
+        *self.last_updated.lock() = Instant::now();
+    }
+    #[inline(always)]
+    fn get(&self) -> u64 {
+        self.counter.load(Ordering::Relaxed)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Harness: a shared multi-thread tokio runtime sized to the host's parallelism.
+// ---------------------------------------------------------------------------
+
+static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
+    let workers = available_parallelism().map(|n| n.get()).unwrap_or(1);
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(workers)
+        .enable_all()
+        .build()
+        .expect("failed to build tokio runtime")
+});
+
+fn bench_contention<C: Counter + Default + 'static>(
+    group: &mut criterion::BenchmarkGroup<'_, WallTime>,
+    name: &str,
+    n_tasks: usize,
+    per_task: u64,
+) {
+    let counter: Arc<C> = Arc::new(C::default());
+    group.bench_with_input(BenchmarkId::new(name, n_tasks), &n_tasks, |b, &n| {
+        b.iter(|| {
+            let counter = counter.clone();
+            RUNTIME.block_on(async move {
+                let mut handles = Vec::with_capacity(n);
+                for _ in 0..n {
+                    let c = counter.clone();
+                    handles.push(tokio::spawn(async move {
+                        for _ in 0..per_task {
+                            c.add(black_box(1));
+                        }
+                    }));
+                }
+                for handle in handles {
+                    let _ = handle.await;
+                }
+                black_box(counter.get());
+            });
+        });
+    });
+}
+
+fn contention_scaling(c: &mut Criterion) {
+    let mut group = c.benchmark_group("contention_scaling");
+    group.throughput(Throughput::Elements(TOTAL_WORK));
+    for &n in TASK_COUNTS {
+        let per = TOTAL_WORK / n as u64;
+        bench_contention::<SingleAtomic>(&mut group, "single_atomic", n, per);
+        bench_contention::<CasSaturating>(&mut group, "cas_saturating", n, per);
+        bench_contention::<ShardedAtomic>(&mut group, "sharded_atomic", n, per);
+        bench_contention::<ThreadLocalCell>(&mut group, "thread_local_cell", n, per);
+        bench_contention::<UnsafeCellCounter>(&mut group, "unsafe_cell", n, per);
+    }
+    group.finish();
+}
+
+fn single_thread_write<C: Counter + Default>(
+    group: &mut criterion::BenchmarkGroup<'_, WallTime>,
+    name: &str,
+) {
+    let counter = C::default();
+    group.bench_function(name, |b| {
+        b.iter(|| {
+            counter.add(black_box(1));
+        });
+    });
+}
+
+fn single_thread_write_group(c: &mut Criterion) {
+    let mut group = c.benchmark_group("single_thread_write");
+    group.throughput(Throughput::Elements(1));
+    single_thread_write::<SingleAtomic>(&mut group, "single_atomic");
+    single_thread_write::<CasSaturating>(&mut group, "cas_saturating");
+    single_thread_write::<ShardedAtomic>(&mut group, "sharded_atomic");
+    single_thread_write::<ThreadLocalCell>(&mut group, "thread_local_cell");
+    single_thread_write::<UnsafeCellCounter>(&mut group, "unsafe_cell");
+    group.finish();
+}
+
+fn read_cost<C: Counter + Default>(
+    group: &mut criterion::BenchmarkGroup<'_, WallTime>,
+    name: &str,
+) {
+    let counter = C::default();
+    counter.add(1000);
+    group.bench_function(name, |b| {
+        b.iter(|| black_box(counter.get()));
+    });
+}
+
+fn read_cost_group(c: &mut Criterion) {
+    let mut group = c.benchmark_group("read_cost");
+    read_cost::<SingleAtomic>(&mut group, "single_atomic");
+    read_cost::<CasSaturating>(&mut group, "cas_saturating");
+    read_cost::<ShardedAtomic>(&mut group, "sharded_atomic");
+    read_cost::<ThreadLocalCell>(&mut group, "thread_local_cell");
+    read_cost::<UnsafeCellCounter>(&mut group, "unsafe_cell");
+    group.finish();
+}
+
+fn bench_handle(
+    group: &mut criterion::BenchmarkGroup<'_, WallTime>,
+    name: &str,
+    n_tasks: usize,
+    per_task: u64,
+    counter: Arc<dyn Counter>,
+) {
+    group.bench_with_input(BenchmarkId::new(name, n_tasks), &n_tasks, |b, &n| {
+        b.iter(|| {
+            let counter = counter.clone();
+            RUNTIME.block_on(async move {
+                let mut handles = Vec::with_capacity(n);
+                for _ in 0..n {
+                    let c = counter.clone();
+                    handles.push(tokio::spawn(async move {
+                        for _ in 0..per_task {
+                            c.add(black_box(1));
+                        }
+                    }));
+                }
+                for handle in handles {
+                    let _ = handle.await;
+                }
+                black_box(counter.get());
+            });
+        });
+    });
+}
+
+fn counter_handle(c: &mut Criterion) {
+    let mut group = c.benchmark_group("counter_handle");
+    group.throughput(Throughput::Elements(HANDLE_TOTAL_WORK));
+
+    // StatsManager::new() spawns a background cleanup task, which needs a tokio
+    // runtime context; bind it to our shared RUNTIME for the lifetime of the
+    // group.
+    let _rt_guard = RUNTIME.enter();
+    let stats = StatsManager::new();
+
+    let prod: Arc<dyn Counter> = Arc::new(stats.get_simple_counter(MetricName::TrafficBytesTx));
+    let cas: Arc<dyn Counter> = Arc::new(BaselineHandle {
+        use_cas: true,
+        ..Default::default()
+    });
+    let fam: Arc<dyn Counter> = Arc::new(BaselineHandle {
+        use_cas: false,
+        ..Default::default()
+    });
+
+    for &n in TASK_COUNTS {
+        let per = HANDLE_TOTAL_WORK / n as u64;
+        bench_handle(&mut group, "prod", n, per, prod.clone());
+        bench_handle(&mut group, "baseline_cas_mutex", n, per, cas.clone());
+        bench_handle(&mut group, "baseline_fetchadd_mutex", n, per, fam.clone());
+    }
+    group.finish();
+}
+
+// Keep the default measurement config; pass CLI flags to speed up a run, e.g.
+// `-- --measurement-time 2 --sample-size 30 --warm-up-time 500`.
+criterion_group! {
+    name = benches;
+    config = Criterion::default();
+    targets = contention_scaling, single_thread_write_group, read_cost_group, counter_handle
+}
+criterion_main!(benches);

easytier/locales/app.yml

@@ -114,11 +114,11 @@ core_clap:
     en: "encryption algorithm to use, supported: '', 'xor', 'chacha20', 'aes-gcm', 'aes-gcm-256', 'openssl-aes128-gcm', 'openssl-aes256-gcm', 'openssl-chacha20'. Empty string means default (aes-gcm)"
     zh-CN: "要使用的加密算法，支持：''（默认aes-gcm）、'xor'、'chacha20'、'aes-gcm'、'aes-gcm-256'、'openssl-aes128-gcm'、'openssl-aes256-gcm'、'openssl-chacha20'"
   multi_thread:
-    en: "use multi-thread runtime, default is single-thread"
-    zh-CN: "使用多线程运行时，默认为单线程"
+    en: "multi-thread tokio runtime (default on). Only affects launcher-based deployments (GUI/mobile/web/Windows service); the easytier-core CLI always runs single-threaded."
+    zh-CN: "多线程 tokio 运行时（默认开启）。仅对 launcher 部署（GUI/移动端/web/Windows 服务）生效；easytier-core CLI 始终为单线程。"
   multi_thread_count:
-    en: "the number of threads to use, default is 2, only effective when multi-thread is enabled, must be greater than 2"
-    zh-CN: "使用的线程数，默认为2，仅在多线程模式下有效。取值必须大于2"
+    en: "the number of worker threads, default 2, only effective when multi-thread is enabled, minimum 2"
+    zh-CN: "worker 线程数，默认 2，仅在启用多线程时生效，最小为 2"
   disable_ipv6:
     en: "do not use ipv6"
     zh-CN: "不使用IPv6"

easytier/src/common/dns.rs

@@ -3,40 +3,33 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 
 use anyhow::Context;
-use hickory_proto::rr::RData;
-use hickory_resolver::config::{
-    ConnectionConfig, LookupIpStrategy, NameServerConfig, ResolverConfig, ResolverOpts,
-};
-use hickory_resolver::net::runtime::TokioRuntimeProvider;
+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_resolver::system_conf::read_system_conf;
-use hickory_resolver::TokioResolver;
+use hickory_resolver::{Resolver, TokioResolver};
 use once_cell::sync::Lazy;
 use tokio::net::lookup_host;
 
 use super::error::Error;
 
 pub fn get_default_resolver_config() -> ResolverConfig {
-    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()],
-            ),
-        ],
-    )
+    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
 }
 
 pub static ALLOW_USE_SYSTEM_DNS_RESOLVER: Lazy<AtomicBool> = Lazy::new(|| AtomicBool::new(true));
 
-pub static RESOLVER: Lazy<Arc<TokioResolver>> =
+pub static RESOLVER: Lazy<Arc<Resolver<GenericConnector<TokioRuntimeProvider>>>> =
     Lazy::new(|| {
         let system_cfg = read_system_conf();
         let mut cfg = get_default_resolver_config();
@@ -48,11 +41,9 @@ pub static RESOLVER: Lazy<Arc<TokioResolver>> =
             opt = s.1;
         }
         opt.ip_strategy = LookupIpStrategy::Ipv4AndIpv6;
-        let resolver = TokioResolver::builder_with_config(cfg, TokioRuntimeProvider::default())
-            .with_options(opt)
-            .build()
-            .expect("failed to build DNS resolver");
-        Arc::new(resolver)
+        let builder = TokioResolver::builder_with_config(cfg, TokioConnectionProvider::default())
+            .with_options(opt);
+        Arc::new(builder.build())
     });
 
 pub async fn resolve_txt_record(domain_name: &str) -> Result<String, Error> {
@@ -62,16 +53,12 @@ pub async fn resolve_txt_record(domain_name: &str) -> Result<String, Error> {
         .await
         .with_context(|| format!("txt_lookup failed, domain_name: {}", domain_name))?;
 
-    let Some(RData::TXT(txt_record)) = response
-        .answers()
+    let txt_record = response
         .iter()
         .next()
-        .map(|record| &record.data)
-    else {
-        return Err(anyhow::anyhow!("no txt record found, domain_name: {}", domain_name).into());
-    };
+        .with_context(|| format!("no txt record found, domain_name: {}", domain_name))?;
 
-    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,10 @@
 use dashmap::DashMap;
 use serde::{Deserialize, Serialize};
-use std::cell::UnsafeCell;
 use std::fmt;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::sync::OnceLock;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
 use tokio::time::interval;
 use tokio_util::task::AbortOnDropHandle;
 
@@ -374,136 +375,106 @@ impl Default for LabelSet {
     }
 }
 
-/// UnsafeCounter provides a high-performance counter using UnsafeCell
+/// Counter provides a high-performance atomic counter
 #[derive(Debug)]
-pub struct UnsafeCounter {
-    value: UnsafeCell<u64>,
+pub struct Counter {
+    value: AtomicU64,
 }
 
-impl Default for UnsafeCounter {
+impl Default for Counter {
     fn default() -> Self {
         Self::new()
     }
 }
 
-impl UnsafeCounter {
+impl Counter {
     pub fn new() -> Self {
         Self {
-            value: UnsafeCell::new(0),
+            value: AtomicU64::new(0),
         }
     }
 
     pub fn new_with_value(initial: u64) -> Self {
         Self {
-            value: UnsafeCell::new(initial),
+            value: AtomicU64::new(initial),
         }
     }
 
     /// Increment the counter by the given amount
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn add(&self, delta: u64) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = (*ptr).saturating_add(delta);
-        }
+    pub fn add(&self, delta: u64) {
+        self.value.fetch_add(delta, Ordering::Relaxed);
     }
 
     /// Increment the counter by 1
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn inc(&self) {
-        unsafe {
-            self.add(1);
-        }
+    pub fn inc(&self) {
+        self.add(1);
     }
 
     /// Get the current value of the counter
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is modifying this counter simultaneously.
-    pub unsafe fn get(&self) -> u64 {
-        let ptr = self.value.get();
-        unsafe { *ptr }
+    pub fn get(&self) -> u64 {
+        self.value.load(Ordering::Relaxed)
     }
 
     /// Reset the counter to zero
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn reset(&self) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = 0;
-        }
+    pub fn reset(&self) {
+        self.value.store(0, Ordering::Relaxed);
     }
 
     /// Set the counter to a specific value
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this counter simultaneously.
-    pub unsafe fn set(&self, value: u64) {
-        let ptr = self.value.get();
-        unsafe {
-            *ptr = value;
-        }
+    pub fn set(&self, value: u64) {
+        self.value.store(value, Ordering::Relaxed);
     }
 }
 
-// UnsafeCounter is Send + Sync because the safety is guaranteed by the caller
-unsafe impl Send for UnsafeCounter {}
-unsafe impl Sync for UnsafeCounter {}
+/// Epoch used to convert a monotonic clock reading into a storable `u64`
+/// millisecond count for `MetricData::last_updated`. Lazily initialized on first
+/// use. Backed by `fastant`, which uses the TSC on x86_64 Linux (and falls back
+/// to `std::time::Instant` elsewhere), making `now_millis()` cheap enough to
+/// call per packet.
+fn time_base() -> fastant::Instant {
+    static BASE: OnceLock<fastant::Instant> = OnceLock::new();
+    *BASE.get_or_init(fastant::Instant::now)
+}
+
+fn now_millis() -> u64 {
+    fastant::Instant::now()
+        .saturating_duration_since(time_base())
+        .as_millis() as u64
+}
 
 /// MetricData contains both the counter and last update timestamp
-/// Uses UnsafeCell for lock-free access
 #[derive(Debug)]
 struct MetricData {
-    counter: UnsafeCounter,
-    last_updated: UnsafeCell<Instant>,
+    counter: Counter,
+    last_updated: AtomicU64,
 }
 
 impl MetricData {
     fn new() -> Self {
         Self {
-            counter: UnsafeCounter::new(),
-            last_updated: UnsafeCell::new(Instant::now()),
+            counter: Counter::new(),
+            last_updated: AtomicU64::new(now_millis()),
         }
     }
 
     fn new_with_value(initial: u64) -> Self {
         Self {
-            counter: UnsafeCounter::new_with_value(initial),
-            last_updated: UnsafeCell::new(Instant::now()),
+            counter: Counter::new_with_value(initial),
+            last_updated: AtomicU64::new(now_millis()),
         }
     }
 
-    /// Update the last_updated timestamp
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is accessing this timestamp simultaneously.
-    unsafe fn touch(&self) {
-        let ptr = self.last_updated.get();
-        unsafe {
-            *ptr = Instant::now();
-        }
+    /// Update the last_updated timestamp. Lock-free.
+    fn touch(&self) {
+        self.last_updated.store(now_millis(), Ordering::Relaxed);
     }
 
-    /// Get the last updated timestamp
-    /// # Safety
-    /// This method is unsafe because it uses UnsafeCell. The caller must ensure
-    /// that no other thread is modifying this timestamp simultaneously.
-    unsafe fn get_last_updated(&self) -> Instant {
-        let ptr = self.last_updated.get();
-        unsafe { *ptr }
+    /// Last update time as milliseconds since `time_base()`.
+    fn last_updated_millis(&self) -> u64 {
+        self.last_updated.load(Ordering::Relaxed)
     }
 }
 
-// MetricData is Send + Sync because the safety is guaranteed by the caller
-unsafe impl Send for MetricData {}
-unsafe impl Sync for MetricData {}
-
 /// MetricKey uniquely identifies a metric with its name and labels
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct MetricKey {
@@ -546,39 +517,31 @@ impl CounterHandle {
 
     /// Increment the counter by the given amount
     pub fn add(&self, delta: u64) {
-        unsafe {
-            self.metric_data.counter.add(delta);
-            self.metric_data.touch();
-        }
+        self.metric_data.counter.add(delta);
+        self.metric_data.touch();
     }
 
     /// Increment the counter by 1
     pub fn inc(&self) {
-        unsafe {
-            self.metric_data.counter.inc();
-            self.metric_data.touch();
-        }
+        self.metric_data.counter.inc();
+        self.metric_data.touch();
     }
 
     /// Get the current value of the counter
     pub fn get(&self) -> u64 {
-        unsafe { self.metric_data.counter.get() }
+        self.metric_data.counter.get()
     }
 
     /// Reset the counter to zero
     pub fn reset(&self) {
-        unsafe {
-            self.metric_data.counter.reset();
-            self.metric_data.touch();
-        }
+        self.metric_data.counter.reset();
+        self.metric_data.touch();
     }
 
     /// Set the counter to a specific value
     pub fn set(&self, value: u64) {
-        unsafe {
-            self.metric_data.counter.set(value);
-            self.metric_data.touch();
-        }
+        self.metric_data.counter.set(value);
+        self.metric_data.touch();
     }
 }
 
@@ -614,9 +577,15 @@ impl StatsManager {
             loop {
                 interval.tick().await;
 
-                let Some(cutoff_time) = Instant::now().checked_sub(Duration::from_secs(180)) else {
-                    continue;
-                };
+                // Drop metrics untouched for 180s and with no live handles.
+                // Compare in the millis-since-base domain so neither the hot
+                // path nor GC reconstructs an `Instant` or locks.
+                //
+                // Use an age-based check (`now - last < STALE`) rather than
+                // `last > now - STALE`: early in process life `now_millis()` is
+                // tiny, so `now - STALE` saturates to 0 and a metric stamped at
+                // 0 would fail a strict `> 0` test and be wrongly evicted.
+                let now = now_millis();
 
                 let Some(counters) = counters_clone.upgrade() else {
                     break;
@@ -624,7 +593,7 @@ impl StatsManager {
 
                 counters.retain(|_, metric_data: &mut Arc<MetricData>| {
                     Arc::strong_count(metric_data) > 1
-                        || unsafe { metric_data.get_last_updated() > cutoff_time }
+                        || now.saturating_sub(metric_data.last_updated_millis()) < 180_000
                 });
                 counters.shrink_to_fit();
             }
@@ -662,7 +631,7 @@ impl StatsManager {
             let key = entry.key();
             let metric_data = entry.value();
 
-            let value = unsafe { metric_data.counter.get() };
+            let value = metric_data.counter.get();
 
             metrics.push(MetricSnapshot {
                 name: key.name,
@@ -695,7 +664,7 @@ impl StatsManager {
         let key = MetricKey::new(name, labels.clone());
 
         if let Some(metric_data) = self.counters.get(&key) {
-            let value = unsafe { metric_data.counter.get() };
+            let value = metric_data.counter.get();
             Some(MetricSnapshot {
                 name,
                 labels: labels.clone(),
@@ -793,20 +762,18 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_unsafe_counter() {
-        let counter = UnsafeCounter::new();
+    async fn test_counter() {
+        let counter = Counter::new();
 
-        unsafe {
-            assert_eq!(counter.get(), 0);
-            counter.inc();
-            assert_eq!(counter.get(), 1);
-            counter.add(5);
-            assert_eq!(counter.get(), 6);
-            counter.set(10);
-            assert_eq!(counter.get(), 10);
-            counter.reset();
-            assert_eq!(counter.get(), 0);
-        }
+        assert_eq!(counter.get(), 0);
+        counter.inc();
+        assert_eq!(counter.get(), 1);
+        counter.add(5);
+        assert_eq!(counter.get(), 6);
+        counter.set(10);
+        assert_eq!(counter.get(), 10);
+        counter.reset();
+        assert_eq!(counter.get(), 0);
     }
 
     #[tokio::test]
@@ -947,12 +914,14 @@ mod tests {
         let counter = stats.get_simple_counter(MetricName::TrafficBytesForwarded);
         counter.set(1);
 
-        let cutoff_time = Instant::now().checked_add(Duration::from_secs(1)).unwrap();
+        // Cutoff 1s in the future, so every metric is stale by timestamp; only
+        // a live handle keeps a metric.
+        let cutoff_millis = now_millis() + 1_000;
         stats
             .counters
             .retain(|_, metric_data: &mut Arc<MetricData>| {
                 Arc::strong_count(metric_data) > 1
-                    || unsafe { metric_data.get_last_updated() > cutoff_time }
+                    || metric_data.last_updated_millis() > cutoff_millis
             });
 
         assert_eq!(stats.metric_count(), 1);
@@ -963,11 +932,33 @@ mod tests {
             .counters
             .retain(|_, metric_data: &mut Arc<MetricData>| {
                 Arc::strong_count(metric_data) > 1
-                    || unsafe { metric_data.get_last_updated() > cutoff_time }
+                    || metric_data.last_updated_millis() > cutoff_millis
             });
         assert_eq!(stats.metric_count(), 0);
     }
 
+    #[tokio::test]
+    async fn test_counter_handle_concurrent_increment() {
+        const THREADS: usize = 8;
+        const INCREMENTS_PER_THREAD: usize = 10_000;
+
+        let stats = StatsManager::new();
+        let counter = stats.get_simple_counter(MetricName::TrafficPacketsForwarded);
+
+        std::thread::scope(|scope| {
+            for _ in 0..THREADS {
+                let counter = counter.clone();
+                scope.spawn(move || {
+                    for _ in 0..INCREMENTS_PER_THREAD {
+                        counter.inc();
+                    }
+                });
+            }
+        });
+
+        assert_eq!(counter.get(), (THREADS * INCREMENTS_PER_THREAD) as u64);
+    }
+
     #[tokio::test]
     async fn test_stats_rpc_data_structures() {
         // Test GetStatsRequest

easytier/src/connector/dns_connector.rs

@@ -13,7 +13,7 @@ use crate::{
 };
 use anyhow::Context;
 use dashmap::DashSet;
-use hickory_resolver::proto::rr::{RData, rdata::SRV};
+use hickory_resolver::proto::rr::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,10 +129,7 @@ impl DnsTunnelConnector {
                         format!("srv_lookup failed, srv_domain: {}", srv_domain)
                     })?;
                     tracing::info!(?response, ?srv_domain, "srv_lookup response");
-                    for record in response.answers() {
-                        let RData::SRV(record) = &record.data else {
-                            continue;
-                        };
+                    for record in response.iter() {
                         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/easytier-core.rs

@@ -23,6 +23,9 @@ pub static malloc_conf: &[u8] = b"retain:false\0";
 
 rust_i18n::i18n!("locales", fallback = "en");
 
+// The easytier-core CLI intentionally uses a single-thread runtime. The
+// `multi_thread` flag only affects launcher-based deployments (GUI / mobile /
+// web / Windows service); see launcher.rs:223.
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> std::process::ExitCode {
     core::main().await

easytier/src/instance/dns_server/config.rs

@@ -92,11 +92,12 @@ impl TryFrom<&Record> for rr::Record {
 
     fn try_from(value: &Record) -> Result<Self, Self::Error> {
         let name = value.name()?;
-        let ttl = value.ttl.as_secs() as u32;
+        let mut record = Self::update0(name, value.ttl.as_secs() as u32, value.rr_type());
+        record.set_dns_class(rr::DNSClass::IN);
         match value.rr_type {
             RecordType::A => {
                 let addr: Ipv4Addr = value.value.parse()?;
-                Ok(Self::from_rdata(name, ttl, RData::A(rr::rdata::a::A(addr))))
+                record.set_data(RData::A(rr::rdata::a::A(addr)));
             }
             RecordType::SOA => {
                 let soa = value.value.split_whitespace().collect::<Vec<_>>();
@@ -110,7 +111,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()?;
-                Ok(Self::from_rdata(name, ttl, RData::SOA(rr::rdata::soa::SOA::new(
+                record.set_data(RData::SOA(rr::rdata::soa::SOA::new(
                     mname,
                     rname,
                     serial,
@@ -118,10 +119,11 @@ 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::net::runtime::TokioRuntimeProvider;
+use hickory_resolver::name_server::TokioConnectionProvider;
 use hickory_resolver::system_conf::read_system_conf;
-use hickory_server::net::runtime::TokioTime;
-use hickory_server::server::Server as HickoryServer;
+use hickory_server::ServerFuture;
+use hickory_server::authority::{AuthorityObject, Catalog, ZoneType};
 use hickory_server::server::{Request, RequestHandler, ResponseHandler, ResponseInfo};
 use hickory_server::store::forwarder::ForwardConfig;
-use hickory_server::store::{forwarder::ForwardZoneHandler, in_memory::InMemoryZoneHandler};
-use hickory_server::zone_handler::{AxfrPolicy, Catalog, ZoneHandler, ZoneType};
+use hickory_server::store::{forwarder::ForwardAuthority, in_memory::InMemoryAuthority};
+use std::io;
 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: HickoryServer<CatalogRequestHandler>,
+    server: ServerFuture<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, T: hickory_server::net::runtime::Time>(
+    async fn handle_request<R: ResponseHandler>(
         &self,
         request: &Request,
         response_handle: R,
@@ -60,14 +60,14 @@ impl RequestHandler for CatalogRequestHandler {
         self.catalog
             .read()
             .await
-            .handle_request::<R, T>(request, response_handle)
+            .handle_request(request, response_handle)
             .await
     }
 }
 
-pub fn build_authority(domain: &str, records: &[Record]) -> Result<InMemoryZoneHandler> {
+pub fn build_authority(domain: &str, records: &[Record]) -> Result<InMemoryAuthority> {
     let zone = rr::Name::from_str(domain)?;
-    let mut authority = InMemoryZoneHandler::empty(zone, ZoneType::Primary, AxfrPolicy::Deny);
+    let mut authority = InMemoryAuthority::empty(zone, ZoneType::Primary, false);
     for record in records.iter() {
         let r = record.try_into()?;
         authority.upsert_mut(r, 0);
@@ -97,16 +97,18 @@ impl Server {
                 .name_servers()
                 .iter()
                 .filter(|&x| {
-                    !config.excluded_forward_nameservers().contains(&x.ip)
+                    !config
+                        .excluded_forward_nameservers()
+                        .contains(&x.socket_addr.ip())
                 })
                 .cloned()
                 .collect::<Vec<_>>()
                 .into(),
             options: Some(system_conf.1),
         };
-        let auth = ForwardZoneHandler::builder_with_config(
+        let auth = ForwardAuthority::builder_with_config(
             forward_config,
-            TokioRuntimeProvider::default(),
+            TokioConnectionProvider::default(),
         )
         .build()
         .unwrap();
@@ -115,7 +117,7 @@ impl Server {
 
         let catalog = Arc::new(RwLock::new(catalog));
         let handler = CatalogRequestHandler::new(catalog.clone());
-        let server = HickoryServer::new(handler);
+        let server = ServerFuture::new(handler);
 
         Ok(Self {
             server,
@@ -185,7 +187,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), 1024);
+                .register_listener(tcp_listener, Duration::from_secs(5));
         }
 
         if let Some(address) = self.general_config.listen_udp() {
@@ -201,11 +203,11 @@ impl Server {
         Ok(())
     }
 
-    pub async fn upsert(&self, name: LowerName, authority: Arc<dyn ZoneHandler>) {
+    pub async fn upsert(&self, name: LowerName, authority: Arc<dyn AuthorityObject>) {
         self.catalog.write().await.upsert(name, vec![authority]);
     }
 
-    pub async fn remove(&self, name: &LowerName) -> Option<Vec<Arc<dyn ZoneHandler>>> {
+    pub async fn remove(&self, name: &LowerName) -> Option<Vec<Arc<dyn AuthorityObject>>> {
         self.catalog.write().await.remove(name)
     }
 
@@ -214,16 +216,11 @@ impl Server {
         update: &Request,
         response_edns: Option<Edns>,
         response_handle: R,
-    ) -> ResponseInfo {
+    ) -> io::Result<ResponseInfo> {
         self.catalog
             .write()
             .await
-            .update(
-                update,
-                response_edns.as_ref(),
-                <TokioTime as hickory_server::net::runtime::Time>::current_time(),
-                response_handle,
-            )
+            .update(update, response_edns, response_handle)
             .await
     }
 
@@ -240,12 +237,7 @@ impl Server {
         self.catalog
             .read()
             .await
-            .lookup(
-                request,
-                response_edns.as_ref(),
-                <TokioTime as hickory_server::net::runtime::Time>::current_time(),
-                response_handle,
-            )
+            .lookup(request, response_edns, response_handle)
             .await
     }
 
@@ -265,14 +257,11 @@ mod tests {
         GeneralConfigBuilder, RecordBuilder, RecordType, RunConfigBuilder,
     };
     use anyhow::Result;
+    use hickory_client::client::{Client, ClientHandle};
     use hickory_proto::rr;
-    use hickory_resolver::TokioResolver;
-    use hickory_resolver::config::{
-        ConnectionConfig, NameServerConfig, ResolverConfig, ResolverOpts,
-    };
-    use hickory_resolver::net::runtime::TokioRuntimeProvider;
+    use hickory_proto::runtime::TokioRuntimeProvider;
+    use hickory_proto::udp::UdpClientStream;
     use maplit::hashmap;
-    use std::net::Ipv4Addr;
     use std::time::Duration;
 
     #[tokio::test]
@@ -325,34 +314,23 @@ mod tests {
         server.run().await?;
 
         let local_addr = server.udp_local_addr().unwrap();
-        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)
+        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,
+            )
             .await?;
+        drop(background_task);
 
         println!("Response: {:?}", response);
 
         assert_eq!(response.answers().len(), 1);
-        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));
+        let expected_record: rr::Record = configured_record.try_into()?;
+        assert_eq!(response.answers().first().unwrap(), &expected_record);
 
         server.shutdown().await?;
         Ok(())

easytier/src/instance/dns_server/server_instance.rs

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

easytier/src/instance/dns_server/tests.rs

@@ -1,16 +1,13 @@
-use std::net::Ipv4Addr;
+use std::net::{Ipv4Addr, SocketAddr};
 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_resolver::TokioResolver;
-use hickory_resolver::config::{
-    ConnectionConfig, NameServerConfig, ResolverConfig, ResolverOpts,
-};
-use hickory_resolver::net::runtime::TokioRuntimeProvider;
-use hickory_resolver::net::{DnsError, NetError};
+use hickory_proto::runtime::TokioRuntimeProvider;
+use hickory_proto::udp::UdpClientStream;
 use tokio::sync::Notify;
 use tokio_util::sync::CancellationToken;
 
@@ -69,56 +66,55 @@ pub async fn prepare_env_with_tld_dns_zone(
 }
 
 pub async fn check_dns_record(fake_ip: &Ipv4Addr, domain: &str, expected_ip: &str) {
-    let resolver = build_test_resolver(fake_ip);
-    let response = resolver
-        .lookup(rr::Name::from_str(domain).unwrap(), rr::RecordType::A)
+    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| 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);
+    assert_eq!(response.answers().len(), 1, "{:?}", response.answers());
     let resp = response.answers().first().unwrap();
-    let rr::RData::A(ip) = &resp.data else {
-        panic!("unexpected DNS response: {response:?}");
-    };
     assert_eq!(
-        ip.0,
+        resp.clone().into_parts().rdata.into_a().unwrap().0,
         expected_ip.parse::<Ipv4Addr>().unwrap()
     );
 }
 
 pub async fn check_dns_record_missing(fake_ip: &Ipv4Addr, domain: &str) {
-    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) => {
+    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| {
             panic!("DNS query for missing record failed unexpectedly for domain '{domain}': {e}")
-        }
-    }
-}
-
-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()
+        });
+    background_task.abort();
+    let _ = background_task.await;
+    assert!(response.answers().is_empty(), "{:?}", response.answers());
 }
 
 #[tokio::test]

easytier/src/tunnel/stats.rs

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