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

@@ -129,6 +129,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "anes"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"
+
 [[package]]
 name = "anstream"
 version = "0.6.15"
@@ -1139,6 +1145,12 @@ dependencies = [
  "toml 0.9.12+spec-1.1.0",
 ]
 
+[[package]]
+name = "cast"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
+
 [[package]]
 name = "cc"
 version = "1.2.10"
@@ -1248,6 +1260,33 @@ dependencies = [
  "windows-targets 0.52.6",
 ]
 
+[[package]]
+name = "ciborium"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e"
+dependencies = [
+ "ciborium-io",
+ "ciborium-ll",
+ "serde",
+]
+
+[[package]]
+name = "ciborium-io"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757"
+
+[[package]]
+name = "ciborium-ll"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9"
+dependencies = [
+ "ciborium-io",
+ "half",
+]
+
 [[package]]
 name = "cidr"
 version = "0.3.1"
@@ -1593,6 +1632,42 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "criterion"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f"
+dependencies = [
+ "anes",
+ "cast",
+ "ciborium",
+ "clap",
+ "criterion-plot",
+ "is-terminal",
+ "itertools 0.10.5",
+ "num-traits",
+ "once_cell",
+ "oorandom",
+ "plotters",
+ "rayon",
+ "regex",
+ "serde",
+ "serde_derive",
+ "serde_json",
+ "tinytemplate",
+ "walkdir",
+]
+
+[[package]]
+name = "criterion-plot"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1"
+dependencies = [
+ "cast",
+ "itertools 0.10.5",
+]
+
 [[package]]
 name = "critical-section"
 version = "1.2.0"
@@ -2266,6 +2341,7 @@ dependencies = [
  "clap_complete",
  "clap_complete_nushell",
  "console-subscriber",
+ "criterion",
  "crossbeam",
  "ctor 0.8.0",
  "dashmap",
@@ -2276,6 +2352,7 @@ dependencies = [
  "derive_builder",
  "derive_more 2.1.1",
  "encoding",
+ "fastant",
  "flume 0.12.0",
  "forwarded-header-value",
  "futures",
@@ -2848,6 +2925,16 @@ dependencies = [
  "pin-project-lite",
 ]
 
+[[package]]
+name = "fastant"
+version = "0.1.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e825441bfb2d831c47c97d05821552db8832479f44c571b97fededbf0099c07"
+dependencies = [
+ "small_ctor",
+ "web-time",
+]
+
 [[package]]
 name = "fastbloom"
 version = "0.9.0"
@@ -4495,6 +4582,17 @@ dependencies = [
  "once_cell",
 ]
 
+[[package]]
+name = "is-terminal"
+version = "0.4.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
+dependencies = [
+ "hermit-abi",
+ "libc",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "is-wsl"
 version = "0.4.0"
@@ -5613,7 +5711,7 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "680998035259dcfcafe653688bf2aa6d3e2dc05e98be6ab46afb089dc84f1df8"
 dependencies = [
- "proc-macro-crate 2.0.0",
+ "proc-macro-crate 3.5.0",
  "proc-macro2",
  "quote",
  "syn 2.0.117",
@@ -5855,6 +5953,12 @@ dependencies = [
  "portable-atomic",
 ]
 
+[[package]]
+name = "oorandom"
+version = "11.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
+
 [[package]]
 name = "opaque-debug"
 version = "0.3.1"
@@ -6485,6 +6589,34 @@ dependencies = [
  "time",
 ]
 
+[[package]]
+name = "plotters"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747"
+dependencies = [
+ "num-traits",
+ "plotters-backend",
+ "plotters-svg",
+ "wasm-bindgen",
+ "web-sys",
+]
+
+[[package]]
+name = "plotters-backend"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a"
+
+[[package]]
+name = "plotters-svg"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670"
+dependencies = [
+ "plotters-backend",
+]
+
 [[package]]
 name = "pnet"
 version = "0.35.0"
@@ -8678,6 +8810,12 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "small_ctor"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "88414a5ca1f85d82cc34471e975f0f74f6aa54c40f062efa42c0080e7f763f81"
+
 [[package]]
 name = "smallvec"
 version = "1.13.2"
@@ -9836,6 +9974,16 @@ dependencies = [
  "zerovec",
 ]
 
+[[package]]
+name = "tinytemplate"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc"
+dependencies = [
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "tinyvec"
 version = "1.8.0"

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

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